LangGraph多Agent编排实战：用Python构建生产级AI工作流的6个核心模式

LangGraph多Agent编排：为什么单Agent永远不够

单个AI Agent只能处理简单任务，一旦涉及多步骤决策、条件分支、人工审批、状态回滚，代码就变成一团意大利面。LangGraph基于有向图（DAG）模型，让你用声明式方式定义Agent之间的协作流程——每个节点是一个Agent或函数，每条边是状态传递路径。2026年，LangGraph已支持StateGraph→条件路由→人机协作→检查点持久化→子图嵌套→流式输出全链路能力。

本文将从6个核心模式出发，带你完成图定义→条件路由→人工审批→状态持久化→子图组合→流式输出的全链路实战。

---

核心概念

概念	说明
StateGraph	基于状态的有向图，节点处理状态、边传递状态
Node	图中的处理单元，接收状态、返回状态更新
Edge	节点间的连接，分为普通边和条件边
Conditional Edge	根据状态动态选择下一个节点的边
Checkpoint	状态快照，支持暂停/恢复/回滚
Human-in-the-Loop	人工审批节点，暂停执行等待人工输入
Subgraph	嵌套的子图，支持模块化组合
Stream Mode	流式输出模式，实时返回中间结果

---

问题分析：多Agent编排解决的5类痛点

1. 流程不可控：单Agent的ReAct循环无法表达复杂业务流程
2. 状态丢失：长对话中Agent无法记住之前的关键决策
3. 人工审批缺失：关键操作无法暂停等待人工确认
4. 错误不可恢复：执行失败后无法回滚到上一个稳定状态
5. 协作混乱：多个Agent之间的职责边界模糊、调用顺序不可控

---

分步实操：6个LangGraph多Agent编排核心模式

模式1：StateGraph基础与图定义

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"]):
        return {"user_intent": "search"}
    elif any(kw in last_message for kw in ["计算", "分析", "统计", "calculate"]):
        return {"user_intent": "calculate"}
    else:
        return {"user_intent": "chat"}

def search_agent(state: AgentState) -> dict:
    query = state["messages"][-1].content
    results = [f"搜索结果1: {query}的相关信息", f"搜索结果2: {query}的深度分析"]
    return {"search_results": results}

def calculate_agent(state: AgentState) -> dict:
    return {"final_answer": "计算结果: 42"}

def chat_agent(state: AgentState) -> dict:
    return {"final_answer": "你好！我是AI助手，有什么可以帮你？"}

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": "帮我搜索Python最新特性"}],
    "user_intent": "",
    "search_results": [],
    "final_answer": "",
})
print(result["search_results"])

模式2：条件路由与多轮对话

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"关于{topic}的研究大纲:\n1. 背景介绍\n2. 核心概念\n3. 实践案例\n4. 总结展望"
    return {"topic": topic, "outline": outline}

def writer(state: ResearchState) -> dict:
    draft = f"基于大纲《{state['outline']}》撰写的初稿..."
    return {"draft": draft}

def reviewer(state: ResearchState) -> dict:
    if state.get("revision_count", 0) >= 2:
        return {"review_feedback": "approved"}
    return {"review_feedback": "需要修改：增加更多实例和数据支撑"}

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"修订版（第{state.get('revision_count', 0) + 1}次修改）: {state['draft']}\n已增加实例和数据。"
    return {"draft": revised, "revision_count": state.get("revision_count", 0) + 1}

def publisher(state: ResearchState) -> dict:
    return {"final_answer": f"发布文章: {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()

模式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"]):
        return {"risk_level": "high"}
    elif any(kw in request for kw in ["修改", "更新", "update"]):
        return {"risk_level": "medium"}
    return {"risk_level": "low"}

def human_approval(state: ApprovalState) -> dict:
    if state["risk_level"] == "high":
        decision = interrupt(f"高风险操作需要审批: {state['request']}")
        return {"approved": decision.get("approved", False)}
    return {"approved": True}

def executor(state: ApprovalState) -> dict:
    if state["approved"]:
        return {"result": f"执行成功: {state['request']}"}
    return {"result": f"操作被拒绝: {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": "删除生产数据库的所有测试数据", "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

模式4：检查点持久化与状态恢复

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"已完成步骤: {state_snapshot.values.get('steps_completed', [])}")

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

模式5：子图嵌套与模块化组合

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检查通过: {state['code'][:50]}..."}

def security_scanner(state: CodeReviewState) -> dict:
    return {"security_result": "安全扫描: 未发现漏洞"}

def style_checker(state: CodeReviewState) -> dict:
    return {"style_result": "代码风格: 符合PEP8规范"}

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": "所有测试通过 (42/42)"}

def deployer(state: DevPipelineState) -> dict:
    return {"deploy_result": "部署成功: v1.0.0 已上线"}

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()

模式6：流式输出与实时反馈

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

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

def responder(state: StreamChatState) -> dict:
    return {"response": f"关于'{state['query']}'的详细回答: LangGraph支持流式输出，可以实时返回中间结果..."}

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": "LangGraph怎么用？", "messages": [], "thinking": "", "response": ""}):
    for node_name, node_output in event.items():
        print(f"[{node_name}] {node_output}")

---

避坑指南

坑1：状态Schema定义不完整

# ❌ 错误：缺少字段默认值
class BadState(TypedDict):
    messages: Annotated[list, add_messages]
    result: str  # 调用时必须提供，否则KeyError

# ✅ 正确：使用Optional或提供默认值
from typing import Optional

class GoodState(TypedDict):
    messages: Annotated[list, add_messages]
    result: str
    metadata: Optional[dict]  # 可选字段

坑2：条件路由返回值不匹配

# ❌ 错误：路由函数返回的值不在映射表中
def bad_router(state) -> str:
    return "unknown_node"  # 映射表中没有这个key

# ✅ 正确：确保所有可能的返回值都在映射表中
def good_router(state) -> Literal["search", "calculate", "chat"]:
    if "搜索" in state["messages"][-1].content:
        return "search"
    elif "计算" in state["messages"][-1].content:
        return "calculate"
    return "chat"

坑3：忘记添加Checkpointer导致状态丢失

# ❌ 错误：不使用checkpointer，中断后无法恢复
app = graph.compile()  # 无状态

# ✅ 正确：使用checkpointer持久化状态
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())

坑4：interrupt使用不当

# ❌ 错误：在非checkpointer图中使用interrupt
app = graph.compile()  # 无checkpointer
# interrupt()会抛出异常

# ✅ 正确：必须配合checkpointer使用
app = graph.compile(checkpointer=MemorySaver())

坑5：子图状态与父图状态不匹配

# ❌ 错误：子图状态字段与父图完全不同
class ParentState(TypedDict):
    code: str
class ChildState(TypedDict):
    data: bytes  # 字段不兼容

# ✅ 正确：子图状态是父图状态的子集或兼容扩展
class ChildState(TypedDict):
    code: str  # 与父图字段对应
    extra_field: str

---

报错排查

序号	报错信息	原因	解决方法
1	KeyError: 'field_name'	状态Schema缺少字段	确保invoke时提供所有必填字段
2	InvalidEdgeError	边指向不存在的节点	检查add_edge的节点名是否已add_node
3	GraphRecursionError	图中存在无限循环	添加最大迭代次数或终止条件
4	interrupt() called without checkpointer	未配置checkpointer	编译时传入checkpointer参数
5	StateUpdateError: incompatible types	状态更新类型不匹配	检查节点返回的dict键值类型
6	NodeNotFoundError	条件路由返回未注册的节点	确保路由返回值在映射表中
7	CheckpointError: thread_id required	未提供thread_id	invoke时传入configurable.thread_id
8	SubgraphStateError	子图与父图状态不兼容	统一状态Schema或使用转换层
9	StreamTimeoutError	流式输出超时	检查节点是否有死循环或长时间阻塞
10	SerializationError	状态包含不可序列化对象	确保状态中只有JSON可序列化类型

---

进阶优化

1. 自定义Reducer：使用Annotated[type, custom_reducer]实现复杂的状态合并逻辑
2. 并行节点：使用add_edge(START, ["node_a", "node_b"])实现节点并行执行
3. 图可视化：使用app.get_graph().draw_mermaid()生成Mermaid流程图
4. 生产级Checkpointer：使用PostgreSQL或Redis替代MemorySaver
5. 异步执行：使用AsyncStateGraph和ainvoke实现异步Agent

---

对比分析

维度	LangGraph	CrewAI	AutoGen	LangChain Agent
图编排能力	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐	⭐⭐
状态持久化	⭐⭐⭐⭐⭐	⭐⭐	⭐⭐⭐	⭐⭐
人机协作	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐
条件路由	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐	⭐
子图嵌套	⭐⭐⭐⭐⭐	⭐⭐	⭐	⭐
学习曲线	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐

---

总结：LangGraph让你从"命令式Agent调用"进化为"声明式图编排"。StateGraph→条件路由→人机协作→检查点持久化→子图嵌套→流式输出六位一体，是2026年Python多Agent编排的首选框架。核心原则：状态即数据流、节点即处理单元、边即控制流。

---

在线工具推荐

• JSON格式化：/zh-CN/json/format
• Hash计算：/zh-CN/encode/hash
• cURL转代码：/zh-CN/dev/curl-to-code