User Query → Researcher → Analyst → Writer → Editor → Output

              Manager Agent
              /     |     \
    Worker 1   Worker 2   Worker 3
    (Search)   (Analyze)  (Summarize)
              \     |     /
              Aggregator Agent

Coder ↔ Reviewer ↔ Tester
  ↓        ↓        ↓
      Consensus

Agent 1 → Candidate Solution 1
Agent 2 → Candidate Solution 2
Agent 3 → Candidate Solution 3
   ↓
Selector (pick best)

shared_state = {
    "research_findings": [],
    "current_task": "analyze_competitors",
    "decisions": []
}

# All agents read/write to shared state
researcher.execute(shared_state)
analyst.execute(shared_state)

# Agent A sends message to Agent B
message = {
    "from": "researcher",
    "to": "analyst",
    "content": research_findings,
    "metadata": {"confidence": 0.9}
}

message_queue.send(message)

# Agents subscribe to events
event_bus.subscribe("research_complete", analyst.on_research_complete)
event_bus.subscribe("analysis_complete", writer.on_analysis_complete)

# Agent publishes event when done
event_bus.publish("research_complete", research_data)

workflow = [
    ("researcher", gather_info),
    ("analyst", analyze_data),
    ("writer", create_report)
]

for agent_name, task in workflow:
    result = agents[agent_name].execute(task, context)
    context.update(result)

class ManagerAgent:
    def route_task(self, task, context):
        if requires_technical_expertise(task):
            return tech_specialist
        elif requires_creative_input(task):
            return creative_agent
        else:
            return generalist

proposals = [agent.propose_solution(task) for agent in agents]
scores = [agent.evaluate(proposals) for agent in agents]
best = proposals[argmax(mean(scores))]

from crewai import Agent, Task, Crew

# Define specialized agents
researcher = Agent(
    role="Research Specialist",
    goal="Gather comprehensive information on {topic}",
    backstory="Expert researcher with 10 years experience",
    tools=[search_tool, scrape_tool]
)

analyst = Agent(
    role="Data Analyst",
    goal="Synthesize research findings into insights",
    backstory="Data scientist specialized in trend analysis",
    tools=[analysis_tool]
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, compelling reports",
    backstory="Professional writer with technical expertise",
    tools=[writing_tool]
)

# Define tasks
research_task = Task(
    description="Research {topic} thoroughly",
    agent=researcher,
    expected_output="Comprehensive research findings with sources"
)

analysis_task = Task(
    description="Analyze research findings for key insights",
    agent=analyst,
    context=[research_task],  # Depends on research_task
    expected_output="List of key insights and trends"
)

writing_task = Task(
    description="Write executive summary based on analysis",
    agent=writer,
    context=[research_task, analysis_task],
    expected_output="500-word executive summary"
)

# Create crew and execute
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    verbose=True
)

result = crew.kickoff(inputs={"topic": "AI market trends"})

from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    input: str
    research: str
    analysis: str
    output: str

def research_node(state):
    research = researcher_agent.run(state["input"])
    return {"research": research}

def analysis_node(state):
    analysis = analyst_agent.run(state["research"])
    return {"analysis": analysis}

def writing_node(state):
    output = writer_agent.run(state["analysis"])
    return {"output": output}

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", END)

app = workflow.compile()

# Execute
result = app.invoke({"input": "Analyze AI market trends"})

Task Complexity?
│
├─ Simple, linear → Single Agent
│
├─ Complex, requires specialization?
│  │
│  ├─ Sequential steps → Pipeline Pattern
│  ├─ Parallel subtasks → Hierarchical Pattern
│  ├─ Need review → Peer Collaboration
│  └─ Explore solutions → Swarm Pattern
│
└─ Uncertain → Start with Single Agent, refactor to Multi if needed

# Track agent execution
class TrackedAgent(Agent):
    def execute(self, task, context):
        start = time.time()
        logger.info(f"{self.name} starting: {task}")

        result = super().execute(task, context)

        duration = time.time() - start
        logger.info(f"{self.name} completed in {duration}s")

        metrics.record({
            "agent": self.name,
            "task": task,
            "duration": duration,
            "tokens": result.token_count,
            "cost": result.cost
        })

        return result