praval.composition

Composition utilities for decorator-based agents.

This module provides utilities for composing and orchestrating agents decorated with the @agent decorator.

Functions

agent_pipeline(*agents[, channel])	Compose agents into a pipeline that processes data sequentially.
conditional_agent(condition_func)	Decorator for conditional agent execution.
start_agents(*agent_funcs[, initial_data, ...])	Convenience function to start multiple agents with initial data.
throttled_agent(delay_seconds)	Decorator to throttle agent execution.

Classes

AgentSession(session_name)

Context manager for coordinated agent sessions.

praval.composition.agent_pipeline(*agents, channel='pipeline')

Compose agents into a pipeline that processes data sequentially.

Parameters:

• *agents (Callable) – Functions decorated with @agent

• channel (str) – Channel name for pipeline communication

Return type:

Callable

Returns:

Function that triggers the pipeline with initial data

Example

pipeline = agent_pipeline(explorer, analyzer, reporter) pipeline({“task”: “analyze sentiment”})

praval.composition.conditional_agent(condition_func)

Decorator for conditional agent execution.

Parameters:

condition_func (Callable[[Any], bool]) – Function that takes a spore and returns bool

Example

@conditional_agent(lambda spore: spore.knowledge.get(“priority”) == “high”) @agent(“urgent_processor”) def process_urgent(spore):

return {“processed”: True}

praval.composition.throttled_agent(delay_seconds)

Decorator to throttle agent execution.

Parameters:

delay_seconds (float) – Minimum seconds between executions

Example

@throttled_agent(2.0) # Max once every 2 seconds @agent(“slow_processor”) def process_slowly(spore):

return {“processed”: True}

class praval.composition.AgentSession(session_name)

Bases: object

Context manager for coordinated agent sessions.

Example

with AgentSession(“knowledge_mining”) as session:

session.add_agents(explorer, analyzer, curator) session.broadcast({“task”: “mine concepts about AI”})

Parameters:

session_name (str)

__init__(session_name)

Parameters:

session_name (str)

add_agent(agent_func)

Add an agent to this session.

Return type:

AgentSession

Parameters:

agent_func (Callable)

add_agents(*agent_funcs)

Add multiple agents to this session.

Return type:

AgentSession

Parameters:

agent_funcs (Callable)

broadcast(data)

Broadcast data to all agents in this session.

Return type:

str

Parameters:

data (Dict[str, Any])

get_stats()

Get session statistics.

Return type:

Dict[str, Any]

praval.composition.start_agents(*agent_funcs, initial_data=None, channel='startup')

Convenience function to start multiple agents with initial data.

Parameters:

• *agent_funcs (Callable) – Functions decorated with @agent

• initial_data (Optional[Dict[str, Any]]) – Initial data to broadcast (optional)

• channel (str) – Channel to use for startup communication

Return type:

str

Returns:

Spore ID of startup broadcast

Example

start_agents(explorer, analyzer, curator,

initial_data={“task”: “analyze market trends”})