LangGraph 实战：构建图式多 Agent 工作流

来源：langchain-ai/langgraph · LangGraph 官方文档 | 整理时间：2026-04-04

什么是 LangGraph？

LangGraph 是 LangChain 团队推出的图式 Agent 编排框架。它把 Agent 工作流建模为有向图（Directed Graph），每个节点是一个处理步骤，边定义了步骤之间的流转关系。

为什么选 LangGraph？

• 精确控制：不像纯 Agent 那样不可预测，你可以定义每一步做什么
• 状态管理：状态在节点间自动传递，支持持久化和恢复
• 人机交互：在关键节点暂停，等待人类审批后继续
• 多 Agent：支持多个 Agent 在同一个图中协作

核心概念

StateGraph

LangGraph 的核心是 StateGraph——一个带状态的图：

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    query: str
    research_result: str
    draft: str
    review_feedback: str
    final_output: str

节点（Node）

每个节点是一个函数，接收状态，返回更新的状态：

def research(state: AgentState) -> dict:
    """研究阶段：收集信息"""
    query = state["query"]
    result = search_and_analyze(query)
    return {"research_result": result}

def write_draft(state: AgentState) -> dict:
    """写作阶段：生成初稿"""
    draft = generate_draft(state["research_result"])
    return {"draft": draft}

def review(state: AgentState) -> dict:
    """审查阶段：检查质量"""
    feedback = review_content(state["draft"])
    return {"review_feedback": feedback}

边（Edge）

边定义节点之间的流转，支持条件路由：

def should_revise(state: AgentState) -> str:
    """根据审查反馈决定是否需要修改"""
    if "approve" in state["review_feedback"].lower():
        return "done"
    return "revise"

# 添加条件边
graph.add_conditional_edges(
    "review",
    should_revise,
    {"done": END, "revise": "write_draft"}
)

第一个 LangGraph 应用

安装

pip install langgraph langchain-anthropic

完整示例：研究助手

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

class ResearchState(TypedDict):
    topic: str
    research_data: str
    outline: str
    article: str
    feedback: str
    revision_count: int

def research_node(state: ResearchState) -> dict:
    """步骤 1：研究主题"""
    prompt = f"Research the following topic and provide key findings: {state['topic']}"
    result = llm.invoke(prompt).content
    return {"research_data": result}

def outline_node(state: ResearchState) -> dict:
    """步骤 2：生成大纲"""
    prompt = f"Create an article outline based on this research:\n{state['research_data']}"
    outline = llm.invoke(prompt).content
    return {"outline": outline}

def write_node(state: ResearchState) -> dict:
    """步骤 3：撰写文章"""
    prompt = f"""Write an article based on:
    Topic: {state['topic']}
    Research: {state['research_data']}
    Outline: {state['outline']}
    """
    article = llm.invoke(prompt).content
    return {"article": article}

def review_node(state: ResearchState) -> dict:
    """步骤 4：审查文章"""
    prompt = f"""Review this article for quality, accuracy, and completeness.
    If good enough, say 'APPROVED'. Otherwise, provide specific feedback.

    Article: {state['article']}"""
    feedback = llm.invoke(prompt).content
    return {
        "feedback": feedback,
        "revision_count": state.get("revision_count", 0) + 1,
    }

def should_continue(state: ResearchState) -> str:
    """决定是否继续修改"""
    if "APPROVED" in state["feedback"] or state["revision_count"] >= 3:
        return "done"
    return "revise"

# 构建图
graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("outline", outline_node)
graph.add_node("write", write_node)
graph.add_node("review", review_node)

# 定义边
graph.set_entry_point("research")
graph.add_edge("research", "outline")
graph.add_edge("outline", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges(
    "review",
    should_continue,
    {"done": END, "revise": "write"},
)

# 编译并运行
app = graph.compile()
result = app.invoke({
    "topic": "AI Agent 在软件开发中的应用",
    "revision_count": 0,
})

print(result["article"])

人机交互（Human-in-the-Loop）

在关键节点暂停，等待人类审批：

from langgraph.checkpoint.memory import MemorySaver

# 添加检查点支持
checkpointer = MemorySaver()
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["write"],  # 在 write 节点前暂停
)

# 第一次运行，会在 write 之前暂停
config = {"configurable": {"thread_id": "thread-1"}}
result = app.invoke({"topic": "AI Agents"}, config)

# 人类审查大纲后，继续执行
result = app.invoke(None, config)  # 传入 None 继续

多 Agent 模式

监督者模式（Supervisor）

from langgraph.graph import StateGraph, END

def supervisor(state):
    """监督者决定分配给哪个 Agent"""
    task_type = classify_task(state["query"])
    if task_type == "research":
        return "researcher"
    elif task_type == "code":
        return "coder"
    else:
        return "writer"

graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", research_agent)
graph.add_node("coder", code_agent)
graph.add_node("writer", writer_agent)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", supervisor, {
    "researcher": "researcher",
    "coder": "coder",
    "writer": "writer",
})

层级模式（Hierarchical）

总监督者
├── 前端团队监督者
│   ├── UI Agent
│   └── 样式 Agent
└── 后端团队监督者
    ├── API Agent
    └── 数据库 Agent

部署选项

相关链接

• Agent 框架全景
• 核心模式详解
• LangGraph GitHub：https://github.com/langchain-ai/langgraph
• LangGraph 官方文档：https://langchain-ai.github.io/langgraph/
• LangGraph 教程：https://langchain-ai.github.io/langgraph/tutorials/