multi-agent-orchestration

Case Manager Agent (Coordinator)
    ├→ Contract Analyzer Agent
    │   └ Task: Review contract terms
    ├→ Precedent Research Agent
    │   └ Task: Find relevant case law
    ├→ Risk Assessor Agent
    │   └ Task: Identify legal risks
    └→ Document Drafter Agent
        └ Task: Prepare legal documents

Support Coordinator
    ├→ Issue Classifier Agent
    │   └ Task: Categorize customer issue
    ├→ Knowledge Base Agent
    │   └ Task: Find relevant documentation
    ├→ Escalation Agent
    │   └ Task: Determine if human escalation needed
    └→ Solution Synthesizer Agent
        └ Task: Prepare comprehensive response

from crewai import Agent, Task, Crew
 
# Define agents
analyst = Agent(
    role="Financial Analyst",
    goal="Analyze financial data and provide insights",
    backstory="Expert in financial markets with 10+ years experience"
)
 
researcher = Agent(
    role="Market Researcher",
    goal="Research market trends and competition",
    backstory="Data-driven researcher specializing in market analysis"
)
 
# Define tasks
analysis_task = Task(
    description="Analyze Q3 financial results for {company}",
    agent=analyst,
    tools=[financial_tool, data_tool]
)
 
research_task = Task(
    description="Research competitive landscape in {market}",
    agent=researcher,
    tools=[web_search_tool, industry_data_tool]
)
 
# Create crew and execute
crew = Crew(
    agents=[analyst, researcher],
    tasks=[analysis_task, research_task],
    process=Process.sequential
)
 
result = crew.kickoff(inputs={"company": "TechCorp", "market": "AI"})

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
 
# Define agents
analyst = AssistantAgent(
    name="analyst",
    system_message="You are a financial analyst..."
)
 
researcher = AssistantAgent(
    name="researcher",
    system_message="You are a market researcher..."
)
 
# Create group chat
groupchat = GroupChat(
    agents=[analyst, researcher],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"
)
 
# Manage group conversation
manager = GroupChatManager(groupchat=groupchat)
 
# User proxy to initiate conversation
user = UserProxyAgent(name="user")
 
# Have conversation
user.initiate_chat(
    manager,
    message="Analyze if Company X should invest in Y market"
)

from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import create_agent_executor
 
# Define state
class AgentState:
    research_findings: str
    analysis: str
    recommendations: str
 
# Create graph
graph = StateGraph(AgentState)
 
# Add nodes for each agent
graph.add_node("researcher", research_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("writer", writer_agent)
 
# Define edges (workflow)
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
 
# Set entry/exit points
graph.set_entry_point("researcher")
graph.set_finish_point("writer")
 
# Compile and run
workflow = graph.compile()
result = workflow.invoke({"topic": "AI trends"})

from swarm import Agent, Swarm
 
# Define agents
triage_agent = Agent(
    name="Triage Agent",
    instructions="Determine which specialist to route the customer to"
)
 
billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing and payment questions"
)
 
technical_agent = Agent(
    name="Technical Support",
    instructions="Handle technical issues"
)
 
# Define handoff functions
def route_to_billing(reason: str):
    return billing_agent
 
def route_to_technical(reason: str):
    return technical_agent
 
# Add tools to triage agent
triage_agent.functions = [route_to_billing, route_to_technical]
 
# Execute swarm
client = Swarm()
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "I have a billing question"}]
)

# Task 1: Research
research_output = research_agent.work("Analyze AI market trends")
 
# Task 2: Analysis (uses research output)
analysis = analyst_agent.work(f"Analyze these findings: {research_output}")
 
# Task 3: Report (uses analysis)
report = writer_agent.work(f"Write report on: {analysis}")

import asyncio
 
async def parallel_teams():
    # All agents work in parallel
    market_task = market_agent.work_async("Analyze market")
    technical_task = tech_agent.work_async("Analyze technology")
    user_task = user_agent.work_async("Analyze user needs")
 
    # Wait for all to complete
    market_results, tech_results, user_results = await asyncio.gather(
        market_task, technical_task, user_task
    )
 
    # Synthesize results
    return synthesize(market_results, tech_results, user_results)

manager_agent.orchestrate({
    "market_analysis": {
        "agents": [competitor_analyst, trend_analyst],
        "task": "Comprehensive market analysis"
    },
    "technical_evaluation": {
        "agents": [architecture_agent, security_agent],
        "task": "Technical feasibility assessment"
    },
    "synthesis": {
        "agents": [strategy_agent],
        "task": "Create strategic recommendations"
    }
})

agents = [bull_agent, bear_agent, neutral_agent]
question = "Should we invest in this startup?"
 
# Debate round 1
arguments = {agent: agent.argue(question) for agent in agents}
 
# Debate round 2 (respond to others)
counter_arguments = {
    agent: agent.respond(arguments) for agent in agents
}
 
# Reach consensus
consensus = mediator_agent.synthesize_consensus(counter_arguments)

agent_a.send_message(agent_b, {
    "type": "request",
    "action": "analyze_document",
    "document": doc_content,
    "context": {"deadline": "urgent"}
})

# Agent A writes to shared memory
shared_memory.write("findings", {"market_size": "$5B", "growth": "20%"})
 
# Agent B reads from shared memory
findings = shared_memory.read("findings")

manager.broadcast("update_all_agents", {
    "new_deadline": "tomorrow",
    "priority": "critical"
})

# Agents negotiate roles based on task
agents = [agent1, agent2, agent3]
task = "complex financial analysis"
 
# Agents determine best structure
negotiated_structure = self_organize(agents, task)
# Returns optimal workflow for this task

# Monitor progress
if progress < expected_rate:
    # Increase resources
    workflow.add_agent(specialist_agent)
elif quality < threshold:
    # Increase validation
    workflow.insert_review_step()

# After team execution
execution_trace = crew.get_execution_trace()
 
# Extract learnings
learnings = extract_patterns(execution_trace)
 
# Update agent knowledge
for agent, learning in learnings.items():
    agent.update_knowledge(learning)