---
title: "Multi-Agent Architecture with LangGraph"
url: "https://vanducng.dev/2025/07/10/Multi-Agent-Architecture-with-LangGraph/"
date: "2025-07-10"
series: "langchain-langgraph"
categories: ["AI"]
tags: ["llm","python","langchain","langgraph","agents"]
---

# Multi-Agent Architecture with LangGraph

Single agents hit a ceiling. When tasks require diverse expertise - research, coding, analysis, writing - a single agent either becomes overloaded with tools or produces mediocre results across domains. Multi-agent systems solve this by decomposing work among specialized agents, each focused on what it does best.

## The Specialization Principle

Human organizations discovered centuries ago that specialization improves outcomes. A team with a researcher, developer, and technical writer produces better documentation than three generalists working independently. Each person develops deep expertise in their domain and delivers higher quality work in less time.

The same principle applies to AI agents. A single agent with 20 tools must make increasingly difficult decisions about which tool to use. Its context window fills with tool descriptions, examples, and results from diverse domains. Performance degrades as cognitive load increases.

Multi-agent systems mirror organizational design. Instead of one overloaded generalist, you deploy specialists who excel at narrow tasks. A research agent becomes expert at information gathering. A coding agent focuses on implementation. A review agent concentrates on quality assessment. Each agent has a focused prompt, relevant tools, and domain-specific examples.

This isn't just about division of labor - it's about **emergent capabilities**. When agents can communicate and hand off work, the system can tackle tasks none could handle individually.

## Why Multiple Agents?

Single-agent architectures face fundamental limits:

```mermaid
graph TD
    subgraph Single["Single Agent"]
        A1[One Agent] --> B1[20+ Tools]
        B1 --> C1[Context Overload]
        C1 --> D1[Poor Tool Selection]
    end

    subgraph Multi["Multi-Agent"]
        A2[Orchestrator] --> B2[Research Agent]
        A2 --> C2[Code Agent]
        A2 --> D2[Writer Agent]
        B2 --> E2[3-4 Tools Each]
        C2 --> E2
        D2 --> E2
    end

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class A1,B1,C1,D1 orangeClass
    class A2,B2,C2,D2,E2 blueClass
```

| Aspect | Single Agent | Multi-Agent |
|--------|--------------|-------------|
| Tool count | 20+ tools, confusion | 3-5 per agent, focused |
| Context | Everything in one prompt | Distributed, specialized |
| Failure | Entire task fails | Graceful degradation |
| Scaling | Harder as complexity grows | Add new specialists |

Multi-agent systems also enable parallelism - while one agent researches, another can write.

## Multi-Agent Patterns

Choosing the right pattern depends on your coordination needs:

| Pattern | Coordination | Best For |
|---------|--------------|----------|
| **Orchestrator** | Central planner delegates | Predictable workflows, clear task decomposition |
| **Supervisor** | Monitor quality, reassign on failure | Quality-critical work, iterative refinement |
| **Peer-to-Peer** | Agents communicate directly | Creative collaboration, debate, synthesis |

### Pattern 1: Orchestrator

The orchestrator pattern mirrors traditional project management. A central coordinator analyzes the task, creates a plan, and delegates to specialists:

```python
def create_plan(state: OrchestratorState) -> dict:
    """Orchestrator breaks task into steps for specialists."""
    prompt = f"""Break this task into steps:
    Task: {state['task']}
    Specialists: researcher, coder, writer
    Return JSON: [{{"agent": "...", "task": "..."}}]"""

    plan = json.loads(llm.invoke(prompt).content)
    return {"plan": plan, "current_step": 0}

def route_to_agent(state: OrchestratorState) -> str:
    """Route to next specialist or synthesize if done."""
    if state["current_step"] >= len(state["plan"]):
        return "synthesize"
    return state["plan"][state["current_step"]]["agent"]
```

The orchestrator's strength is predictability - you can trace exactly which agent handled which part of the task.

### Pattern 2: Supervisor with Workers

The supervisor monitors progress and can reassign or retry:

```python
class SupervisorState(TypedDict):
    task: str
    messages: Annotated[list, add]
    agent_outputs: dict
    quality_scores: dict
    iterations: int

def supervisor_node(state: SupervisorState) -> dict:
    """Supervise agents, evaluate quality, and decide next steps."""

    # Evaluate completed work
    outputs = state.get("agent_outputs", {})

    if not outputs:
        # Initial delegation
        return {
            "messages": [AIMessage(content="Delegating to research agent first")]
        }

    # Quality check
    prompt = f"""Evaluate the quality of this work on a scale of 1-10:

Work output:
{outputs}

Return only a number 1-10."""

    response = llm.invoke([HumanMessage(content=prompt)])

    try:
        score = int(response.content.strip())
    except ValueError:
        score = 5

    if score < 7 and state.get("iterations", 0) < 3:
        # Request revision
        return {
            "messages": [AIMessage(content=f"Quality score {score}/10. Requesting revision.")],
            "iterations": state.get("iterations", 0) + 1
        }

    # Accept and move forward
    return {
        "messages": [AIMessage(content=f"Accepted with score {score}/10")],
        "quality_scores": {**state.get("quality_scores", {}), "final": score}
    }

def should_continue(state: SupervisorState) -> str:
    """Determine if we need more work or can finalize."""
    iterations = state.get("iterations", 0)
    scores = state.get("quality_scores", {})

    if scores.get("final", 0) >= 7:
        return "finalize"
    if iterations >= 3:
        return "finalize"  # Give up after 3 attempts
    return "revise"
```

```mermaid
graph TD
    A[Supervisor] --> B{Evaluate Quality}
    B -->|Score < 7| C[Request Revision]
    C --> D[Worker Revises]
    D --> A
    B -->|Score >= 7| E[Accept & Continue]
    B -->|Max Iterations| E
    E --> F[Next Task or Finalize]

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class A blueClass
    class B,C,D orangeClass
    class E,F greenClass
```

### Pattern 3: Peer-to-Peer Collaboration

Agents communicate directly without a central coordinator:

```python
class CollaborativeState(TypedDict):
    topic: str
    discussion: Annotated[list[dict], add]
    consensus: str | None
    round: int

def researcher_agent(state: CollaborativeState) -> dict:
    """Research agent contributes facts and evidence."""
    context = "\n".join([
        f"{msg['agent']}: {msg['content']}"
        for msg in state.get("discussion", [])
    ])

    prompt = f"""You are a research agent in a collaborative discussion.
Topic: {state['topic']}

Previous discussion:
{context or 'No discussion yet.'}

Provide factual research insights. Be specific and cite sources when possible.
Keep response under 150 words."""

    response = llm.invoke([HumanMessage(content=prompt)])

    return {
        "discussion": [{"agent": "researcher", "content": response.content}]
    }

def critic_agent(state: CollaborativeState) -> dict:
    """Critic agent challenges assumptions and identifies weaknesses."""
    context = "\n".join([
        f"{msg['agent']}: {msg['content']}"
        for msg in state.get("discussion", [])
    ])

    prompt = f"""You are a critical analyst in a collaborative discussion.
Topic: {state['topic']}

Previous discussion:
{context}

Identify weaknesses, gaps, or counterarguments. Be constructive.
Keep response under 150 words."""

    response = llm.invoke([HumanMessage(content=prompt)])

    return {
        "discussion": [{"agent": "critic", "content": response.content}]
    }

def synthesizer_agent(state: CollaborativeState) -> dict:
    """Synthesizer combines insights into a coherent conclusion."""
    context = "\n".join([
        f"{msg['agent']}: {msg['content']}"
        for msg in state.get("discussion", [])
    ])

    prompt = f"""You are synthesizing a collaborative discussion.
Topic: {state['topic']}

Discussion:
{context}

Create a balanced synthesis that addresses the key points and critiques.
Provide a clear conclusion."""

    response = llm.invoke([HumanMessage(content=prompt)])

    return {
        "consensus": response.content,
        "round": state.get("round", 0) + 1
    }

# Build collaborative graph
collab_graph = StateGraph(CollaborativeState)
collab_graph.add_node("researcher", researcher_agent)
collab_graph.add_node("critic", critic_agent)
collab_graph.add_node("synthesizer", synthesizer_agent)

# Round-robin discussion
collab_graph.add_edge(START, "researcher")
collab_graph.add_edge("researcher", "critic")
collab_graph.add_edge("critic", "synthesizer")

def check_rounds(state: CollaborativeState) -> str:
    if state.get("round", 0) >= 2:
        return END
    return "researcher"

collab_graph.add_conditional_edges("synthesizer", check_rounds)

collaborative_agents = collab_graph.compile()
```

## Agent Communication

### Shared State vs Message Passing

Two approaches for inter-agent communication:

```python
# Approach 1: Shared State
# All agents read/write to common state
class SharedState(TypedDict):
    research_findings: list[str]
    code_artifacts: list[str]
    review_comments: list[str]

# Approach 2: Message Passing
# Agents send explicit messages
class MessageState(TypedDict):
    inbox: dict[str, list[dict]]  # agent_name -> messages
    outbox: dict[str, list[dict]]

def send_message(state: MessageState, from_agent: str, to_agent: str, content: str) -> dict:
    """Send a message from one agent to another."""
    message = {
        "from": from_agent,
        "content": content,
        "timestamp": datetime.now().isoformat()
    }

    current_inbox = state.get("inbox", {})
    agent_inbox = current_inbox.get(to_agent, [])
    agent_inbox.append(message)

    return {"inbox": {**current_inbox, to_agent: agent_inbox}}
```

### Handoffs Between Agents

Clean handoffs transfer context and responsibility:

```python
from typing import Literal

class HandoffState(TypedDict):
    task: str
    current_agent: str
    context: dict
    history: Annotated[list[dict], add]

def create_handoff(
    from_agent: str,
    to_agent: str,
    summary: str,
    context: dict
) -> dict:
    """Create a structured handoff between agents."""
    return {
        "current_agent": to_agent,
        "context": context,
        "history": [{
            "from": from_agent,
            "to": to_agent,
            "summary": summary,
            "timestamp": datetime.now().isoformat()
        }]
    }

def research_with_handoff(state: HandoffState) -> dict:
    """Research agent that hands off to coder."""
    # Do research work
    findings = "Found 3 relevant APIs for the task..."

    # Prepare handoff
    return create_handoff(
        from_agent="researcher",
        to_agent="coder",
        summary="Research complete. Found APIs: X, Y, Z. Recommend starting with X.",
        context={
            "findings": findings,
            "recommended_approach": "Use API X with pagination",
            "constraints": ["Rate limit: 100/min", "Auth required"]
        }
    )

def coder_receives_handoff(state: HandoffState) -> dict:
    """Coder agent receives context from researcher."""
    context = state.get("context", {})

    prompt = f"""You are a coding agent. You received this handoff:

Findings: {context.get('findings', 'None')}
Approach: {context.get('recommended_approach', 'None')}
Constraints: {context.get('constraints', [])}

Implement the solution based on this context."""

    response = llm.invoke([HumanMessage(content=prompt)])

    return {
        "context": {**context, "implementation": response.content}
    }
```

## State Management Across Agents

### Scoped State

Not all state should be shared. Use scoped access:

```python
class ScopedState(TypedDict):
    # Global - all agents can access
    task: str
    final_result: str

    # Agent-specific - only certain agents write
    research_data: dict  # Only researcher writes
    code_output: dict    # Only coder writes
    review_notes: dict   # Only reviewer writes

def make_scoped_node(agent_name: str, writable_keys: list[str]):
    """Create a node that can only write to specific state keys."""

    def scoped_node(state: ScopedState) -> dict:
        # Agent does its work
        result = do_agent_work(state, agent_name)

        # Filter to only writable keys
        filtered = {k: v for k, v in result.items() if k in writable_keys}

        return filtered

    return scoped_node

# Create agents with scoped write access
researcher = make_scoped_node("researcher", ["research_data"])
coder = make_scoped_node("coder", ["code_output"])
reviewer = make_scoped_node("reviewer", ["review_notes", "final_result"])
```

### Persistent Memory Across Sessions

For long-running multi-agent systems, persist state:

```python
from langgraph.checkpoint.sqlite import SqliteSaver
import json

class PersistentMultiAgentState(TypedDict):
    project_id: str
    agents_memory: dict[str, dict]  # Per-agent persistent memory
    shared_knowledge: list[dict]
    conversation_history: Annotated[list, add]

def save_agent_memory(state: PersistentMultiAgentState, agent_name: str, memory: dict) -> dict:
    """Save agent-specific memory that persists across sessions."""
    current_memory = state.get("agents_memory", {})
    agent_memory = current_memory.get(agent_name, {})

    # Merge new memory
    updated = {**agent_memory, **memory}

    return {
        "agents_memory": {**current_memory, agent_name: updated}
    }

def load_agent_memory(state: PersistentMultiAgentState, agent_name: str) -> dict:
    """Load agent's persistent memory."""
    return state.get("agents_memory", {}).get(agent_name, {})

# Use with checkpointer for persistence
checkpointer = SqliteSaver.from_conn_string("multi_agent.db")

multi_agent_system = graph.compile(checkpointer=checkpointer)

# Each project maintains its own state
config = {"configurable": {"thread_id": f"project-{project_id}"}}
result = multi_agent_system.invoke(initial_state, config=config)
```

## Building a Complete Multi-Agent System

Let's build a research and writing system with three specialized agents:

```python
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from typing import TypedDict, Annotated, Literal
from operator import add

# Shared state
class ResearchWritingState(TypedDict):
    topic: str
    research_notes: list[str]
    outline: list[str]
    draft: str
    review_feedback: str
    final_article: str
    current_phase: str

llm = ChatOpenAI(model="gpt-4o")

# Research Agent
@tool
def search_academic(query: str) -> str:
    """Search academic papers for information."""
    return f"Found 5 papers on: {query}. Key findings: ..."

@tool
def search_news(query: str) -> str:
    """Search recent news articles."""
    return f"Recent news on {query}: 3 articles found..."

research_tools = [search_academic, search_news]
research_llm = llm.bind_tools(research_tools)

def research_agent(state: ResearchWritingState) -> dict:
    """Gather information on the topic."""
    prompt = f"""You are a research agent. Gather comprehensive information on:
Topic: {state['topic']}

Use available tools to find academic and news sources.
Compile your findings as detailed research notes."""

    messages = [HumanMessage(content=prompt)]
    response = research_llm.invoke(messages)

    # Process tool calls if any
    notes = [response.content] if response.content else []

    if response.tool_calls:
        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_fn = {t.name: t for t in research_tools}[tool_name]
            result = tool_fn.invoke(tool_call["args"])
            notes.append(f"[{tool_name}]: {result}")

    return {
        "research_notes": notes,
        "current_phase": "outlining"
    }

# Writer Agent
def writer_agent(state: ResearchWritingState) -> dict:
    """Create outline and draft based on research."""
    notes = "\n".join(state.get("research_notes", []))

    # First create outline
    outline_prompt = f"""Based on this research, create an article outline:

Research Notes:
{notes}

Create a structured outline with 4-6 main sections."""

    outline_response = llm.invoke([HumanMessage(content=outline_prompt)])
    outline = outline_response.content.split("\n")

    # Then write draft
    draft_prompt = f"""Write a complete article based on this outline:

Outline:
{outline_response.content}

Research:
{notes}

Write an engaging, informative article of 500-800 words."""

    draft_response = llm.invoke([HumanMessage(content=draft_prompt)])

    return {
        "outline": outline,
        "draft": draft_response.content,
        "current_phase": "reviewing"
    }

# Editor Agent
def editor_agent(state: ResearchWritingState) -> dict:
    """Review and refine the draft."""
    review_prompt = f"""Review this article draft for:
1. Accuracy (based on research notes)
2. Clarity and flow
3. Engagement
4. Grammar and style

Draft:
{state['draft']}

Research Notes:
{chr(10).join(state.get('research_notes', [])[:3])}

Provide specific feedback and suggestions."""

    feedback = llm.invoke([HumanMessage(content=review_prompt)])

    # Apply edits
    edit_prompt = f"""Revise this article based on feedback:

Original Draft:
{state['draft']}

Feedback:
{feedback.content}

Produce the final polished article."""

    final = llm.invoke([HumanMessage(content=edit_prompt)])

    return {
        "review_feedback": feedback.content,
        "final_article": final.content,
        "current_phase": "complete"
    }

# Build the multi-agent graph
graph = StateGraph(ResearchWritingState)

graph.add_node("researcher", research_agent)
graph.add_node("writer", writer_agent)
graph.add_node("editor", editor_agent)

# Sequential workflow
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "editor")
graph.add_edge("editor", END)

research_writing_system = graph.compile()
```

```mermaid
graph LR
    A[START] --> B[Research Agent]
    B --> C[Writer Agent]
    C --> D[Editor Agent]
    D --> E[END]

    subgraph Research["Research Phase"]
        B1[Search Academic] --> B
        B2[Search News] --> B
    end

    subgraph Writing["Writing Phase"]
        C --> C1[Create Outline]
        C1 --> C2[Write Draft]
    end

    subgraph Editing["Editing Phase"]
        D --> D1[Review]
        D1 --> D2[Polish]
    end

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class A,E greenClass
    class B,B1,B2 blueClass
    class C,C1,C2 orangeClass
    class D,D1,D2 blueClass
```

### Running the System

```python
# Execute the multi-agent workflow
result = research_writing_system.invoke({
    "topic": "The impact of large language models on software development",
    "research_notes": [],
    "outline": [],
    "draft": "",
    "review_feedback": "",
    "final_article": "",
    "current_phase": "researching"
})

print("=== Research Notes ===")
for note in result["research_notes"]:
    print(f"- {note[:100]}...")

print("\n=== Final Article ===")
print(result["final_article"])
```

## Conflict Resolution

When agents disagree, you need resolution strategies:

```python
class ConflictState(TypedDict):
    proposals: dict[str, str]  # agent -> proposal
    votes: dict[str, str]      # agent -> voted_proposal
    resolution: str

def collect_proposals(state: ConflictState) -> dict:
    """Each agent submits a proposal."""
    # Agents have already submitted via their nodes
    return {}

def voting_round(state: ConflictState) -> dict:
    """Agents vote on proposals (can't vote for own)."""
    proposals = state.get("proposals", {})
    votes = {}

    for agent in proposals:
        # Each agent evaluates and votes
        other_proposals = {k: v for k, v in proposals.items() if k != agent}

        if other_proposals:
            prompt = f"""You are {agent}. Vote for the best proposal:

{chr(10).join([f'{k}: {v}' for k, v in other_proposals.items()])}

Return only the name of the agent whose proposal you vote for."""

            response = llm.invoke([HumanMessage(content=prompt)])
            votes[agent] = response.content.strip()

    return {"votes": votes}

def resolve_conflict(state: ConflictState) -> dict:
    """Determine winner or synthesize if tied."""
    votes = state.get("votes", {})
    proposals = state.get("proposals", {})

    # Count votes
    vote_counts = {}
    for voted_for in votes.values():
        vote_counts[voted_for] = vote_counts.get(voted_for, 0) + 1

    if not vote_counts:
        # No votes, synthesize
        resolution = "No consensus - using first proposal"
        return {"resolution": list(proposals.values())[0] if proposals else resolution}

    # Find winner
    winner = max(vote_counts, key=vote_counts.get)

    # Check for tie
    max_votes = vote_counts[winner]
    tied = [k for k, v in vote_counts.items() if v == max_votes]

    if len(tied) > 1:
        # Tie - synthesize
        tied_proposals = {k: proposals.get(k, "") for k in tied}
        synthesis_prompt = f"""Synthesize these tied proposals:

{chr(10).join([f'{k}: {v}' for k, v in tied_proposals.items()])}

Create a combined solution that takes the best from each."""

        response = llm.invoke([HumanMessage(content=synthesis_prompt)])
        return {"resolution": response.content}

    return {"resolution": proposals.get(winner, "No resolution")}
```

## Key Takeaways

1. **Multi-agent beats single-agent for complex tasks**: Specialization improves quality; parallelism improves speed.

2. **Choose the right pattern**: Orchestrator for clear hierarchies, supervisor for quality control, peer-to-peer for creative collaboration.

3. **State scoping matters**: Not all agents need access to all state. Scope writes to prevent conflicts.

4. **Handoffs transfer context**: Clean handoffs with summaries and structured context enable smooth transitions.

5. **Persist for long-running systems**: Use checkpointing to maintain state across sessions and survive failures.

6. **Resolve conflicts systematically**: Voting, synthesis, or hierarchy - pick a strategy and implement it explicitly.

---

*Next: [From Prototype to Production: LangGraph Systems](https://vanducng.dev/2025/07/24/From-Prototype-to-Production-LangGraph-Systems/) - We'll cover deployment, monitoring, error handling, and building production-ready agent architectures.*