Multi-Agent Patterns
Choose the right pattern for agent collaboration – handoffs, workflows, teams, and programmatic control
Multi-agent patterns enable different types of collaboration between agents, from dynamic routing to fixed workflows to collaborative teams.
from praisonaiagents import Agent
triage = Agent(name="Triage", instructions="Route the request to the right specialist")
triage.start("Help me debug a failing CI pipeline")The user describes a task; handoffs, workflows, or teams pick the collaboration pattern.
Quick Start
- Simple Handoffs
python from praisonaiagents import Agent # Create specialist agent specialist = Agent( name="specialist", instructions="You are an expert in your domain.", ) # Router agent with handoff capability router = Agent( name="router", instructions="Route requests to appropriate specialists.", handoffs=[specialist] # LLM decides when to handoff ) result = router.run("Help me with a complex task") - With Configuration
python from praisonaiagents import Agent, HandoffConfig, ContextPolicy # Configure handoff behavior handoff_config = HandoffConfig( context_policy=ContextPolicy.SUMMARY, detect_cycles=True, timeout_seconds=300 ) router = Agent( name="router", instructions="Route requests intelligently.", handoffs=[specialist], handoff_config=handoff_config )
—Multi-agent patterns enable different types of collaboration between agents.
from praisonaiagents import Agent
specialist = Agent(name="specialist", instructions="You are an expert in your domain.")
router = Agent(name="router", instructions="Route requests to specialists.", handoffs=[specialist])
router.start("Help me with a complex task")The user sends one request; the router agent chooses handoffs, workflows, or team patterns as needed.
| Pattern | Use Case | Control | Example |
|---|---|---|---|
| Handoffs | Dynamic routing | LLM decides | Customer service triage |
| Programmatic | Explicit control | Code decides | Error handling workflow |
| AgentFlow | Fixed sequence | Predefined order | Data processing pipeline |
| AgentTeam | Collaboration | Shared context | Multi-expert analysis |
Configuration Options
Handoff ConfigurationComplete handoff configuration reference
Common Patterns
Pattern 1: Handoffs (LLM-Driven)
When to use: LLM decides which specialist agent to call based on the user’s request.
Basic Handoff
from praisonaiagents import Agent
researcher = Agent(
name="researcher",
instructions="Research topics thoroughly using web search.",
tools=["web_search"]
)
writer = Agent(
name="writer",
instructions="Write engaging articles based on research.",
)
# Router agent with handoff capabilities
router = Agent(
name="router",
instructions="Route requests to appropriate specialists.",
handoffs=[researcher, writer]
)
result = router.run("Write an article about AI trends after researching")Advanced Config
from praisonaiagents import Agent, HandoffConfig, ContextPolicy
handoff_config = HandoffConfig(
context_policy=ContextPolicy.SUMMARY,
max_context_tokens=2000,
detect_cycles=True,
max_depth=5
)
router = Agent(
name="intelligent_router",
instructions="Route complex requests through specialists.",
handoffs=[researcher, analyzer, writer],
handoff_config=handoff_config
)
result = router.run("Create a market analysis report")Pattern 2: Programmatic Handoffs
When to use: Your application code needs explicit control over which agent handles a task.
Basic Programmatic
from praisonaiagents import Agent
reviewer = Agent(
name="code_reviewer",
instructions="Review code for bugs and improvements.",
)
fixer = Agent(
name="code_fixer",
instructions="Fix code issues based on review feedback.",
)
async def process_code_review(code: str):
# First, review the code
review_result = await reviewer.arun(f"Review this code:\n{code}")
# Check if issues were found (your logic)
if "issues found" in review_result.lower():
# Explicitly handoff to fixer
fix_result = await reviewer.handoff_to_async(
fixer,
f"Fix issues:\nReview: {review_result}\nCode: {code}"
)
return fix_result
return review_result
result = await process_code_review("def divide(a, b): return a / b")Async Concurrent
from praisonaiagents import Agent
import asyncio
async def process_dataset(datasets: list):
coordinator = Agent(name="coordinator")
validator = Agent(name="validator")
processor = Agent(name="processor")
# Process multiple datasets concurrently
validation_tasks = []
for dataset in datasets:
task = coordinator.handoff_to_async(
validator,
f"Validate dataset: {dataset}"
)
validation_tasks.append(task)
validation_results = await asyncio.gather(*validation_tasks)
# Sequential processing of validated data
processed_results = []
for i, validation_result in enumerate(validation_results):
if "valid" in validation_result.lower():
processed = await coordinator.handoff_to_async(
processor,
f"Process this validated data: {datasets[i]}"
)
processed_results.append(processed)
return processed_resultsPattern 3: AgentFlow (Sequential/Parallel)
When to use: Tasks follow a predictable sequence or can be executed in parallel.
Sequential
from praisonaiagents import Agent, AgentFlow, Task
extractor = Agent(name="data_extractor")
transformer = Agent(name="data_transformer")
loader = Agent(name="data_loader")
extract_task = Task(
name="extract_data",
description="Extract customer data",
agent=extractor
)
transform_task = Task(
name="transform_data",
description="Clean and standardize data",
agent=transformer
)
load_task = Task(
name="load_data",
description="Load data into database",
agent=loader
)
# Sequential workflow
etl_pipeline = AgentFlow(
name="etl_pipeline",
agents=[extractor, transformer, loader],
tasks=[extract_task, transform_task, load_task]
)
result = etl_pipeline.run("Process today's data batch")Parallel
from praisonaiagents import Agent, AgentFlow, Task
sentiment_agent = Agent(name="sentiment_analyzer")
keyword_agent = Agent(name="keyword_extractor")
summary_agent = Agent(name="summarizer")
sentiment_task = Task(
name="analyze_sentiment",
description="Analyze feedback sentiment",
agent=sentiment_agent
)
keyword_task = Task(
name="extract_keywords",
description="Extract key topics",
agent=keyword_agent
)
summary_task = Task(
name="summarize_feedback",
description="Create executive summary",
agent=summary_agent
)
# Parallel workflow
analysis_flow = AgentFlow(
name="parallel_analysis",
agents=[sentiment_agent, keyword_agent, summary_agent],
tasks=[sentiment_task, keyword_task, summary_task],
parallel=True # Enable parallel execution
)
result = analysis_flow.run("Analyze customer feedback")Pattern 4: AgentTeam (Collaborative)
When to use: Multiple agents need to work together on the same problem.
from praisonaiagents import Agent, AgentTeam, Task
market_analyst = Agent(
name="market_analyst",
instructions="Analyze market trends and competition.",
tools=["web_search", "data_analysis"]
)
financial_analyst = Agent(
name="financial_analyst",
instructions="Analyze financial data and risk factors.",
tools=["financial_tools", "calculator"]
)
strategy_consultant = Agent(
name="strategy_consultant",
instructions="Synthesize analysis into strategic recommendations.",
)
analysis_task = Task(
name="market_analysis",
description="Comprehensive market analysis for product launch",
agent=market_analyst
)
# Collaborative team
expert_team = AgentTeam(
name="strategic_analysis_team",
agents=[market_analyst, financial_analyst, strategy_consultant],
tasks=[analysis_task],
collaboration_mode="consensus"
)
result = expert_team.run("Analyze market opportunity for solar panels")Best Practices
Pattern Selection
Choose the right pattern for your use case:
- Start Simple: Use single agents first, add patterns as complexity grows
- LLM Routing: Use handoffs when the AI should decide the flow
- Deterministic: Use AgentFlow for predictable, repeatable processes
- Collaborative: Use AgentTeam when agents need to build on each other’s work
- Explicit Control: Use programmatic handoffs for error handling and conditionals
Context Management
Optimize context sharing:
- Use
ContextPolicy.SUMMARYfor most handoffs (safe default) - Use
ContextPolicy.FULLonly when complete history is essential - Set
max_context_tokensto control costs and latency - Enable
detect_cycles=Trueto prevent infinite loops
handoff_config = HandoffConfig(
context_policy=ContextPolicy.SUMMARY,
max_context_tokens=1000,
detect_cycles=True
)Error Handling
Handle common errors:
# HandoffTimeoutError: Handoff timed out
handoff_config = HandoffConfig(timeout_seconds=600)
# HandoffCycleError: Cycle detected
handoff_config = HandoffConfig(detect_cycles=True)
# HandoffDepthError: Max depth exceeded
handoff_config = HandoffConfig(max_depth=15)Performance Optimization
Optimize for your use case:
- Handoffs: Best for dynamic routing (10-100ms overhead)
- AgentFlow: Best for predictable pipelines (minimal overhead)
- AgentTeam: Best for collaborative analysis (higher context cost)
- Programmatic: Best for explicit control (lowest overhead)
# Efficient context sharing
handoff_config = HandoffConfig(
context_policy=ContextPolicy.SUMMARY,
max_context_tokens=1000
)
# Async for I/O-bound operations
result = await agent.handoff_to_async(specialist, task)