LangGraph Multi-Agent Orchestration: 6 Core Patterns for Production-Ready AI Workflows with Python

LangGraph Multi-Agent Orchestration: Why a Single Agent Is Never Enough

A single AI Agent can only handle simple tasks. Once you need multi-step decisions, conditional branching, human approval, or state rollback, your code becomes spaghetti. LangGraph uses a directed graph (DAG) model to let you declaratively define collaboration flows between Agents — each node is an Agent or function, each edge is a state transfer path. In 2026, LangGraph supports the full pipeline: StateGraph → conditional routing → human-in-the-loop → checkpoint persistence → subgraph nesting → streaming output.

This article walks through 6 core patterns, covering the full pipeline from graph definition → conditional routing → human approval → state persistence → subgraph composition → streaming output.

---

Core Concepts

Concept	Description
StateGraph	State-based directed graph; nodes process state, edges transfer state
Node	Processing unit in the graph; receives state, returns state updates
Edge	Connection between nodes;分为普通边和条件边
Conditional Edge	Edge that dynamically selects the next node based on state
Checkpoint	State snapshot; supports pause/resume/rollback
Human-in-the-Loop	Human approval node; pauses execution awaiting human input
Subgraph	Nested subgraph; supports modular composition
Stream Mode	Streaming output mode; returns intermediate results in real-time

---

Problem Analysis: 5 Pain Points Solved by Multi-Agent Orchestration

1. Uncontrollable Flow: Single Agent's ReAct loop cannot express complex business processes
2. Lost State: Agents cannot remember key decisions from earlier in long conversations
3. Missing Human Approval: Critical operations cannot pause for human confirmation
4. Irrecoverable Errors: Failed execution cannot roll back to a previous stable state
5. Chaotic Collaboration: Blurred responsibility boundaries and uncontrolled call ordering between multiple Agents

---

Step-by-Step: 6 Core LangGraph Multi-Agent Orchestration Patterns

Pattern 1: StateGraph Basics and Graph Definition

pip install langgraph==0.4.0 langchain-openai==0.3.0

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_intent: str
    search_results: list[str]
    final_answer: str

def classify_intent(state: AgentState) -> dict:
    last_message = state["messages"][-1].content
    if any(kw in last_message for kw in ["search", "find", "lookup"]):
        return {"user_intent": "search"}
    elif any(kw in last_message for kw in ["calculate", "analyze", "compute"]):
        return {"user_intent": "calculate"}
    else:
        return {"user_intent": "chat"}

def search_agent(state: AgentState) -> dict:
    query = state["messages"][-1].content
    results = [f"Result 1: info about {query}", f"Result 2: deep analysis of {query}"]
    return {"search_results": results}

def calculate_agent(state: AgentState) -> dict:
    return {"final_answer": "Calculation result: 42"}

def chat_agent(state: AgentState) -> dict:
    return {"final_answer": "Hello! I'm an AI assistant. How can I help you?"}

def route_intent(state: AgentState) -> str:
    return state["user_intent"]

graph = StateGraph(AgentState)

graph.add_node("classify", classify_intent)
graph.add_node("search", search_agent)
graph.add_node("calculate", calculate_agent)
graph.add_node("chat", chat_agent)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {
    "search": "search",
    "calculate": "calculate",
    "chat": "chat",
})
graph.add_edge("search", END)
graph.add_edge("calculate", END)
graph.add_edge("chat", END)

app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "Help me search for the latest Python features"}],
    "user_intent": "",
    "search_results": [],
    "final_answer": "",
})
print(result["search_results"])

Pattern 2: Conditional Routing and Multi-Turn Conversations

from typing import Literal

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    topic: str
    outline: str
    draft: str
    review_feedback: str
    revision_count: int

def researcher(state: ResearchState) -> dict:
    topic = state.get("topic", state["messages"][-1].content)
    outline = f"Research outline for {topic}:\n1. Background\n2. Core Concepts\n3. Case Studies\n4. Conclusion"
    return {"topic": topic, "outline": outline}

def writer(state: ResearchState) -> dict:
    draft = f"First draft based on outline '{state['outline']}'..."
    return {"draft": draft}

def reviewer(state: ResearchState) -> dict:
    if state.get("revision_count", 0) >= 2:
        return {"review_feedback": "approved"}
    return {"review_feedback": "Needs revision: add more examples and data support"}

def should_revise(state: ResearchState) -> Literal["revise", "publish"]:
    if state.get("review_feedback") == "approved":
        return "publish"
    return "revise"

def reviser(state: ResearchState) -> dict:
    revised = f"Revised (revision #{state.get('revision_count', 0) + 1}): {state['draft']}\nAdded examples and data."
    return {"draft": revised, "revision_count": state.get("revision_count", 0) + 1}

def publisher(state: ResearchState) -> dict:
    return {"final_answer": f"Published article: {state['draft']}"}

research_graph = StateGraph(ResearchState)

research_graph.add_node("research", researcher)
research_graph.add_node("write", writer)
research_graph.add_node("review", reviewer)
research_graph.add_node("revise", reviser)
research_graph.add_node("publish", publisher)

research_graph.add_edge(START, "research")
research_graph.add_edge("research", "write")
research_graph.add_edge("write", "review")
research_graph.add_conditional_edges("review", should_revise, {
    "revise": "revise",
    "publish": "publish",
})
research_graph.add_edge("revise", "review")
research_graph.add_edge("publish", END)

research_app = research_graph.compile()

Pattern 3: Human-in-the-Loop

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    messages: Annotated[list, add_messages]
    request: str
    risk_level: str
    approved: bool
    result: str

def risk_assessor(state: ApprovalState) -> dict:
    request = state["request"]
    if any(kw in request for kw in ["delete", "reset", "clear", "drop"]):
        return {"risk_level": "high"}
    elif any(kw in request for kw in ["modify", "update", "change"]):
        return {"risk_level": "medium"}
    return {"risk_level": "low"}

def human_approval(state: ApprovalState) -> dict:
    if state["risk_level"] == "high":
        decision = interrupt(f"High-risk operation requires approval: {state['request']}")
        return {"approved": decision.get("approved", False)}
    return {"approved": True}

def executor(state: ApprovalState) -> dict:
    if state["approved"]:
        return {"result": f"Executed successfully: {state['request']}"}
    return {"result": f"Operation rejected: {state['request']}"}

approval_graph = StateGraph(ApprovalState)
approval_graph.add_node("assess", risk_assessor)
approval_graph.add_node("approve", human_approval)
approval_graph.add_node("execute", executor)

approval_graph.add_edge(START, "assess")
approval_graph.add_edge("assess", "approve")
approval_graph.add_edge("approve", "execute")
approval_graph.add_edge("execute", END)

checkpointer = MemorySaver()
approval_app = approval_graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-001"}}
result = approval_app.invoke(
    {"request": "Delete all test data from production database", "messages": [], "risk_level": "", "approved": False, "result": ""},
    config=config,
)

for state in approval_app.get_state_history(config):
    if state.next:
        approval_app.invoke(
            Command(resume={"approved": True}),
            config=config,
        )
        break

Pattern 4: Checkpoint Persistence and State Recovery

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

class LongTaskState(TypedDict):
    messages: Annotated[list, add_messages]
    task_id: str
    steps_completed: list[str]
    current_step: str
    error: str

def step_one(state: LongTaskState) -> dict:
    return {"current_step": "step_one", "steps_completed": state.get("steps_completed", []) + ["step_one"]}

def step_two(state: LongTaskState) -> dict:
    return {"current_step": "step_two", "steps_completed": state.get("steps_completed", []) + ["step_two"]}

def step_three(state: LongTaskState) -> dict:
    return {"current_step": "step_three", "steps_completed": state.get("steps_completed", []) + ["step_three"]}

long_task_graph = StateGraph(LongTaskState)
long_task_graph.add_node("one", step_one)
long_task_graph.add_node("two", step_two)
long_task_graph.add_node("three", step_three)

long_task_graph.add_edge(START, "one")
long_task_graph.add_edge("one", "two")
long_task_graph.add_edge("two", "three")
long_task_graph.add_edge("three", END)

conn = sqlite3.connect(":memory:", check_same_thread=False)
sqlite_checkpointer = SqliteSaver(conn)
long_task_app = long_task_graph.compile(checkpointer=sqlite_checkpointer)

task_config = {"configurable": {"thread_id": "long-task-001"}}
long_task_app.invoke({"task_id": "task-1", "messages": [], "steps_completed": [], "current_step": "", "error": ""}, config=task_config)

state_snapshot = long_task_app.get_state(task_config)
print(f"Completed steps: {state_snapshot.values.get('steps_completed', [])}")

for history_state in long_task_app.get_state_history(task_config):
    print(f"Step: {history_state.values.get('current_step', 'N/A')}, Time: {history_state.created_at}")

Pattern 5: Subgraph Nesting and Modular Composition

class CodeReviewState(TypedDict):
    messages: Annotated[list, add_messages]
    code: str
    lint_result: str
    security_result: str
    style_result: str
    final_review: str

def linter(state: CodeReviewState) -> dict:
    return {"lint_result": f"Lint check passed: {state['code'][:50]}..."}

def security_scanner(state: CodeReviewState) -> dict:
    return {"security_result": "Security scan: no vulnerabilities found"}

def style_checker(state: CodeReviewState) -> dict:
    return {"style_result": "Code style: PEP8 compliant"}

review_subgraph = StateGraph(CodeReviewState)
review_subgraph.add_node("lint", linter)
review_subgraph.add_node("security", security_scanner)
review_subgraph.add_node("style", style_checker)

review_subgraph.add_edge(START, "lint")
review_subgraph.add_edge(START, "security")
review_subgraph.add_edge(START, "style")
review_subgraph.add_edge("lint", END)
review_subgraph.add_edge("security", END)
review_subgraph.add_edge("style", END)

review_sub_app = review_subgraph.compile()

class DevPipelineState(TypedDict):
    messages: Annotated[list, add_messages]
    code: str
    review_result: str
    test_result: str
    deploy_result: str

def code_review_node(state: DevPipelineState) -> dict:
    review_output = review_sub_app.invoke({
        "code": state["code"],
        "messages": [],
        "lint_result": "",
        "security_result": "",
        "style_result": "",
        "final_review": "",
    })
    return {"review_result": f"Lint: {review_output['lint_result']} | Security: {review_output['security_result']} | Style: {review_output['style_result']}"}

def test_runner(state: DevPipelineState) -> dict:
    return {"test_result": "All tests passed (42/42)"}

def deployer(state: DevPipelineState) -> dict:
    return {"deploy_result": "Deployment successful: v1.0.0 is live"}

pipeline_graph = StateGraph(DevPipelineState)
pipeline_graph.add_node("review", code_review_node)
pipeline_graph.add_node("test", test_runner)
pipeline_graph.add_node("deploy", deployer)

pipeline_graph.add_edge(START, "review")
pipeline_graph.add_edge("review", "test")
pipeline_graph.add_edge("test", "deploy")
pipeline_graph.add_edge("deploy", END)

pipeline_app = pipeline_graph.compile()

Pattern 6: Streaming Output and Real-Time Feedback

class StreamChatState(TypedDict):
    messages: Annotated[list, add_messages]
    query: str
    thinking: str
    response: str

def thinker(state: StreamChatState) -> dict:
    return {"thinking": f"Analyzing: {state['query']}"}

def responder(state: StreamChatState) -> dict:
    return {"response": f"Detailed answer about '{state['query']}': LangGraph supports streaming output, returning intermediate results in real-time..."}

stream_graph = StateGraph(StreamChatState)
stream_graph.add_node("think", thinker)
stream_graph.add_node("respond", responder)

stream_graph.add_edge(START, "think")
stream_graph.add_edge("think", "respond")
stream_graph.add_edge("respond", END)

stream_app = stream_graph.compile()

for event in stream_app.stream({"query": "How to use LangGraph?", "messages": [], "thinking": "", "response": ""}):
    for node_name, node_output in event.items():
        print(f"[{node_name}] {node_output}")

---

Pitfall Guide

Pitfall 1: Incomplete State Schema Definition

# ❌ Wrong: missing field defaults
class BadState(TypedDict):
    messages: Annotated[list, add_messages]
    result: str  # Must be provided on invoke, otherwise KeyError

# ✅ Correct: use Optional or provide defaults
from typing import Optional

class GoodState(TypedDict):
    messages: Annotated[list, add_messages]
    result: str
    metadata: Optional[dict]  # Optional field

Pitfall 2: Conditional Routing Return Value Mismatch

# ❌ Wrong: router returns value not in mapping
def bad_router(state) -> str:
    return "unknown_node"  # Not in the mapping table

# ✅ Correct: ensure all possible returns are in the mapping
def good_router(state) -> Literal["search", "calculate", "chat"]:
    if "search" in state["messages"][-1].content:
        return "search"
    elif "calculate" in state["messages"][-1].content:
        return "calculate"
    return "chat"

Pitfall 3: Forgetting Checkpointer Causes State Loss

# ❌ Wrong: no checkpointer, cannot resume after interruption
app = graph.compile()  # Stateless

# ✅ Correct: use checkpointer for state persistence
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())

Pitfall 4: Improper interrupt Usage

# ❌ Wrong: using interrupt in a graph without checkpointer
app = graph.compile()  # No checkpointer
# interrupt() will throw an exception

# ✅ Correct: must use with checkpointer
app = graph.compile(checkpointer=MemorySaver())

Pitfall 5: Subgraph State Mismatch with Parent Graph

# ❌ Wrong: subgraph state fields completely different from parent
class ParentState(TypedDict):
    code: str
class ChildState(TypedDict):
    data: bytes  # Incompatible fields

# ✅ Correct: subgraph state is a subset or compatible extension of parent
class ChildState(TypedDict):
    code: str  # Matches parent field
    extra_field: str

---

Error Troubleshooting

#	Error Message	Cause	Solution
1	KeyError: 'field_name'	State schema missing field	Ensure all required fields are provided on invoke
2	InvalidEdgeError	Edge points to non-existent node	Check that add_edge node names have been add_node'd
3	GraphRecursionError	Infinite loop in graph	Add max iteration count or termination condition
4	interrupt() called without checkpointer	No checkpointer configured	Pass checkpointer parameter when compiling
5	StateUpdateError: incompatible types	State update type mismatch	Check node return dict key-value types
6	NodeNotFoundError	Conditional routing returns unregistered node	Ensure routing return values are in the mapping
7	CheckpointError: thread_id required	No thread_id provided	Pass configurable.thread_id on invoke
8	SubgraphStateError	Subgraph and parent state incompatible	Unify state schema or use conversion layer
9	StreamTimeoutError	Streaming output timeout	Check for infinite loops or long-blocking nodes
10	SerializationError	State contains non-serializable objects	Ensure only JSON-serializable types in state

---

Advanced Optimization

1. Custom Reducer: Use Annotated[type, custom_reducer] for complex state merge logic
2. Parallel Nodes: Use add_edge(START, ["node_a", "node_b"]) for parallel node execution
3. Graph Visualization: Use app.get_graph().draw_mermaid() to generate Mermaid flowcharts
4. Production Checkpointer: Use PostgreSQL or Redis instead of MemorySaver
5. Async Execution: Use AsyncStateGraph and ainvoke for async Agents

---

Comparison

Dimension	LangGraph	CrewAI	AutoGen	LangChain Agent
Graph Orchestration	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐	⭐⭐
State Persistence	⭐⭐⭐⭐⭐	⭐⭐	⭐⭐⭐	⭐⭐
Human-in-the-Loop	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐
Conditional Routing	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐	⭐
Subgraph Nesting	⭐⭐⭐⭐⭐	⭐⭐	⭐	⭐
Learning Curve	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐

---

Summary: LangGraph evolves you from "imperative Agent calls" to "declarative graph orchestration". StateGraph → conditional routing → human-in-the-loop → checkpoint persistence → subgraph nesting → streaming output — six pillars working together, LangGraph is the go-to choice for Python multi-agent orchestration in 2026. Core principles: state is data flow, nodes are processing units, edges are control flow.

---

Recommended Online Tools

• JSON Formatter: /en/json/format
• Hash Calculator: /en/encode/hash
• cURL to Code: /en/dev/curl-to-code