LangGraph 与状态机

📅 发布于 2026-07-06

用图结构构建可控的 Agent 工作流

11.1 LangGraph：把 Agent 流程变成图

LangChain 团队推出的 LangGraph 是 2024 年最火的 Agent 框架之一。核心思想：把 Agent 工作流建模为有向图。

Agent 的每一步是图中的一个节点（Node），步骤之间的跳转是边（Edge）。整个 Agent 就是一张图，可视化、可调试、可控。

流程图：LangGraph 概念图

• START（入口节点） -> 规划节点（planner）
  • 需搜索 -> 搜索节点（searcher）
  • 需计算 -> 代码节点（coder）
• 搜索节点（searcher） -> 检查节点（reviewer）
• 代码节点（coder） -> 检查节点（reviewer）
• 检查节点（reviewer）
  • 需继续（虚线）-> 规划节点（planner）
  • 完成 -> END（出口节点）

11.2 LangGraph 的核心概念

五个核心概念

1. State（状态） 全局状态对象，所有节点共享。用 TypedDict 定义，包含消息历史、中间结果等。每个节点可以读写 State。

2. Node（节点） 图中的执行单元。每个节点是一个函数，接收 State，返回更新后的 State。节点 = 一步操作。

3. Edge（边） 节点之间的连接。普通边 = 固定跳转；条件边 = 根据 State 动态决定下一个节点。

4. Conditional Edge（条件边） 根据当前 State 动态路由。类似 if-else：if state['need_search'] -> search_node, else -> code_node。

5. Checkpoint（检查点） 每个节点执行后自动保存 State 快照。支持时间旅行（回到任意检查点）、人工干预（暂停在某节点等人审核后继续）、错误恢复（从失败点重试）。

11.3 用 LangGraph 构建 ReAct Agent

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessage
import operator

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 消息列表（只增不覆盖）
    tool_results: list  # 工具结果

# 2. 定义节点函数
def call_llm(state: AgentState):
    """LLM 推理节点：决定下一步做什么"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def call_tool(state: AgentState):
    """工具执行节点：执行 LLM 选择的工具"""
    last_msg = state["messages"][-1]
    tool_call = last_msg.tool_calls[0]
    result = execute_tool(tool_call)
    return {"messages": [ToolMessage(result)]}

# 3. 定义条件路由
def should_continue(state: AgentState):
    """条件边：判断是否继续循环"""
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"  # 需要调用工具
    return END           # 不需要，结束

# 4. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_llm)
workflow.add_node("tools", call_tool)

# 设置入口
workflow.set_entry_point("agent")

# 添加边和条件边
workflow.add_conditional_edges(
    "agent",        # 源节点
    should_continue, # 路由函数
    {
        "tools": "tools",  # 返回 "tools" -> 跳到 tools 节点
        END: END           # 返回 END -> 结束
    }
)
workflow.add_edge("tools", "agent")  # tools 执行完回到 agent

# 5. 编译并运行
app = workflow.compile(checkpointer=MemorySaver())

result = app.invoke(
    {"messages": [HumanMessage("北京明天天气？")]},
    config={"configurable": {"thread_id": "1"}}
)

代码解析

State 定义 Annotated[list, operator.add] 表示 messages 列表用"追加"而非"覆盖"语义。新消息追加到列表，不替换旧消息。

条件边 should_continue 函数检查最后一条消息是否包含 tool_calls。有则路由到 tools 节点，无则结束。

循环结构 agent -> (条件) -> tools -> agent -> ... 形成循环。这就是 ReAct 的循环在 LangGraph 中的表达。

Checkpointer MemorySaver() 保存每步状态。通过 thread_id 可以回到任意检查点，支持人工干预。

11.4 Human-in-the-Loop：人工干预

LangGraph 的 Checkpoint 机制让"人工干预"变得自然：

流程图：人工干预流程

• Agent 执行 -> 检查点（暂停） -> 人工审核（Human）
• 人工审核（Human） -> 通过？
  • 通过 -> 继续执行 -> （虚线）-> Agent 执行
  • 修改 -> 修改 State 后继续 -> （虚线）-> Agent 执行
  • 拒绝 -> 终止

# 在工具执行前暂停，等待人工审核
app = workflow.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"]  # 在 tools 节点前暂停
)

# 第一次调用 -> 执行到 tools 前暂停
result = app.invoke(
    {"messages": [HumanMessage("删除 /tmp 目录")]},
    config={"configurable": {"thread_id": "1"}}
)
# 此时 Agent 想调用 rm 命令，暂停等待人工审核

# 人工审核后，决定继续或修改
# 方式1: 直接继续（批准）
result = app.invoke(None, config={"configurable": {"thread_id": "1"}})

# 方式2: 修改 State 后继续（纠正）
app.update_state(
    config={"configurable": {"thread_id": "1"}},
    values={"messages": [HumanMessage("不要删除，改为列出文件")]}
)
result = app.invoke(None, config={"configurable": {"thread_id": "1"}})

11.5 LangGraph vs LangChain Agent

对比项	LangChain Agent	LangGraph
流程控制	黑盒，LLM 自主决定	白盒，图结构显式定义
可调试性	难，只能看最终输出	强，每步可检查 State
人工干预	不支持	原生支持（Checkpoint）
并行执行	不支持	支持（多个节点并行）
适用场景	简单、快速的 Agent	复杂、需要控制的生产级 Agent

11.6 LangGraph 完整实战代码

前面几节我们分别学习了 State、Node、Edge 和 Checkpoint 的概念，现在把它们组合起来，写一个完整可运行的 LangGraph 项目。这个示例模拟一个智能客服 Agent：接收用户问题，判断是否需要查数据库、调用外部 API 或直接回答，执行后在审查节点检查结果质量，不达标则重试，达标则返回最终回复。

from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
import operator

# -- 1. 定义全局状态 --
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]   # 对话历史（追加策略）
    step_count: int                            # 当前循环步数
    tool_results: list                         # 工具返回结果
    needs_review: bool                         # 是否需要人工审查

# -- 2. 定义节点函数 --
def agent_node(state: AgentState) -> dict:
    """LLM 推理节点：分析用户意图，决定调用哪个工具"""
    step = state.get("step_count", 0) + 1
    # 实际项目中这里调用 llm.invoke(state["messages"])
    decision = "search_db" if step < 3 else "answer"
    return {"messages": [{"role": "ai", "content": f"step={step}, decision={decision}"}], "step_count": step}

def search_db_node(state: AgentState) -> dict:
    """数据库查询节点：根据 LLM 决策检索知识库"""
    # 实际项目中这里连接数据库执行查询
    result = {"found": True, "data": "订单 #12345 已发货，预计 7月3日 送达"}
    return {"tool_results": [result], "messages": [{"role": "tool", "content": str(result)}]}

def answer_node(state: AgentState) -> dict:
    """最终回复节点：整合所有信息生成回答"""
    final = f"根据查询结果：{state.get('tool_results', [])}"
    return {"messages": [{"role": "ai", "content": final}]}

def review_node(state: AgentState) -> dict:
    """质量审查节点：检查回复是否满足要求"""
    last = state["messages"][-1]
    quality_ok = len(str(last.get("content", ""))) > 5
    return {"needs_review": not quality_ok, "step_count": state.get("step_count", 0)}

# -- 3. 定义条件路由 --
def should_continue(state: AgentState) -> str:
    if state.get("step_count", 0) >= 5:
        return "end"
    last_msg = state["messages"][-1]
    if "decision=search_db" in str(last_msg.get("content", "")):
        return "search_db"
    return "answer"

def review_decision(state: AgentState) -> str:
    if state.get("needs_review", False) and state.get("step_count", 0) < 5:
        return "retry"
    return "finish"

# -- 4. 构建 StateGraph --
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("search_db", search_db_node)
workflow.add_node("answer", answer_node)
workflow.add_node("review", review_node)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, {
    "search_db": "search_db",
    "answer": "answer",
    "end": END
})
workflow.add_edge("search_db", "review")
workflow.add_edge("answer", "review")
workflow.add_conditional_edges("review", review_decision, {
    "retry": "agent",
    "finish": END
})

# -- 5. 编译并运行 --
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

result = app.invoke(
    {"messages": [{"role": "user", "content": "我的订单 #12345 到哪了？"}], "step_count": 0, "tool_results": [], "needs_review": False},
    config={"configurable": {"thread_id": "customer-001"}}
)
print(result["messages"][-1])

上面这段代码展示了 LangGraph 的完整开发流程：定义 State -> 编写节点函数 -> 定义条件路由 -> 组装图 -> 编译运行。几个关键设计点值得注意：第一，step_count 字段用于防止无限循环，当步数超过 5 时强制结束，这是生产环境中的必备安全措施。第二，review 节点充当质量守门员，不满意的回复会被打回 agent 节点重试，形成了一个自我纠错的闭环。第三，MemorySaver 配合 thread_id 实现了状态持久化，同一个 thread 的多次调用可以无缝续接，这对于多轮对话场景至关重要。

在实际项目中，你只需要把 agent_node 中的伪逻辑替换为真实的 LLM 调用，把 search_db_node 替换为你的数据库查询逻辑，就可以直接部署使用了。整个图结构的定义不需要改动，这就是 LangGraph "一次建模、反复使用"的优势所在。

11.7 LangGraph vs 其他框架对比

除了 LangGraph，市面上还有多个流行的 Agent 框架，它们各有侧重。理解差异有助于在项目中做出正确的技术选型。下面从五个关键维度对比四大主流框架。

对比维度	LangGraph	LangChain Agent	AutoGen	CrewAI
图结构支持	原生有向图，支持环和并行	无图结构，链式调用	对话图，以消息传递为核心	任务图，按角色分工串联
状态管理	TypedDict 全局 State + Checkpoint	仅消息历史，无持久状态	对话上下文即状态	任务结果传递，无统一 State
人在循环（HITL）	原生支持，interrupt_before/after	不支持	支持，通过 human_reply 消息	有限支持，需自定义逻辑
并行执行	支持，多节点可并行触发	不支持，严格串行	支持，多 Agent 可并行对话	有限支持，任务级别可并行
适用场景	复杂生产级 Agent、需要精细控制	快速原型、简单链式任务	多 Agent 对话协作、研究探索	角色扮演型团队协作、SOP 驱动

从对比可以看出，LangGraph 在图结构表达力和状态可控性方面优势明显：它既支持循环（ReAct 模式）又支持并行（多节点同时触发），加上 Checkpoint 机制让 Human-in-the-Loop 变得开箱即用。AutoGen 更适合多 Agent 自由对话的研究场景，CrewAI 则擅长按预设角色分工的任务流水线。选型建议：需要精细控制流程选 LangGraph，快速验证选 LangChain，多 Agent 协作选 AutoGen，角色分工明确选 CrewAI。

流程图：框架复杂度对比

• LangGraph（复杂度 最高，控制力 最强）
• AutoGen（复杂度 较高，控制力 较强）
• CrewAI（复杂度 中等，控制力 中等）
• LangChain（复杂度 较低，控制力 较弱）
• END（根据场景选择合适框架）

流程方向：复杂 -> 简单

11.8 高级特性：子图与人机协同

子图（Subgraph）：模块化管理复杂流程

当 Agent 工作流变得复杂时，把所有节点放在一个图里会导致可读性和可维护性下降。子图允许你把一组相关节点封装成一个独立子图，然后作为单个节点嵌入主图。这样做有三个好处：一是逻辑分层，主图只关心高层流程，子图处理细节；二是复用性，同一个子图可以在不同主图中重复使用；三是独立测试，子图可以单独编译和调试，不影响主图。

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

# -- 子图状态：研究子流程 --
class ResearchState(TypedDict):
    queries: list          # 查询列表
    findings: Annotated[list, operator.add]  # 研究结果

# -- 子图节点 --
def generate_queries(state: ResearchState) -> dict:
    """根据用户问题生成多个搜索查询"""
    return {"queries": ["LangGraph 教程", "LangGraph 实战", "LangGraph vs LangChain"]}

def execute_search(state: ResearchState) -> dict:
    """执行搜索并收集结果"""
    results = []
    for q in state["queries"]:
        # 实际项目中调用搜索 API
        results.append({"query": q, "snippet": f"关于 {q} 的搜索结果..."})
    return {"findings": results}

def summarize(state: ResearchState) -> dict:
    """汇总搜索结果"""
    summary = "; ".join([r["snippet"] for r in state["findings"]])
    return {"findings": [{"summary": summary}]}

# -- 构建子图 --
research_subgraph = StateGraph(ResearchState)
research_subgraph.add_node("gen_queries", generate_queries)
research_subgraph.add_node("search", execute_search)
research_subgraph.add_node("summarize", summarize)
research_subgraph.add_edge(START, "gen_queries")
research_subgraph.add_edge("gen_queries", "search")
research_subgraph.add_edge("search", "summarize")
research_subgraph.add_edge("summarize", END)
research_app = research_subgraph.compile()

# -- 主图：将子图作为节点嵌入 --
class MainState(TypedDict):
    topic: str
    research_result: str
    final_report: str

def research_node(state: MainState) -> dict:
    """调用研究子图"""
    sub_result = research_app.invoke({"queries": [], "findings": []})
    return {"research_result": str(sub_result["findings"][-1])}

def write_report(state: MainState) -> dict:
    """根据研究结果撰写报告"""
    return {"final_report": f"研究报告：{state['research_result']}"}

main_graph = StateGraph(MainState)
main_graph.add_node("research", research_node)
main_graph.add_node("report", write_report)
main_graph.add_edge(START, "research")
main_graph.add_edge("research", "report")
main_graph.add_edge("report", END)
main_app = main_graph.compile()

这段代码展示了两层图结构：内层 research_subgraph 负责"生成查询 -> 执行搜索 -> 汇总结果"三步，外层主图调用子图后将结果传给 write_report 节点生成最终报告。子图的 State（ResearchState）与主图的 State（MainState）完全独立，通过 research_node 函数做桥接转换，这种设计让两层的逻辑互不污染。

Human-in-the-Loop：三种人机协同模式

LangGraph 的人机协同不止"暂停-审核-继续"一种模式，实际项目中有三种常见模式：审批模式（执行前等人批准）、纠正模式（暂停后人工修改 State 再继续）、对话模式（Agent 遇到不确定时主动向人提问）。下面用代码演示这三种模式的实现。

from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class HitlState(TypedDict):
    action: str
    approved: bool
    feedback: str
    result: str

# -- 模式1：审批模式 -- 执行前等待人工批准 --
def risky_action(state: HitlState) -> dict:
    """高风险操作节点（如删除文件、发送邮件）"""
    return {"result": f"已执行操作: {state['action']}"}

def execute_approval_mode():
    """编译时设置 interrupt_before，在 risky_action 前暂停"""
    wg = StateGraph(HitlState)
    wg.add_node("risky", risky_action)
    wg.add_edge(START, "risky")
    wg.add_edge("risky", END)
    app = wg.compile(
        checkpointer=MemorySaver(),
        interrupt_before=["risky"]  # 关键：执行前暂停
    )
    return app

# -- 模式2：纠正模式 -- 人工修改 State 后继续 --
def execute_correction_mode(app, thread_id: str):
    """暂停后人工修改 State，再继续执行"""
    cfg = {"configurable": {"thread_id": thread_id}}
    # 第一次调用：执行到暂停点
    state = app.invoke({"action": "删除 /tmp/all", "approved": False, "feedback": "", "result": ""}, config=cfg)
    # 人工审查后修改 State
    app.update_state(cfg, values={"action": "列出 /tmp/all 文件而非删除"})
    # 继续执行
    result = app.invoke(None, config=cfg)
    return result

# -- 模式3：对话模式 -- Agent 主动向人提问 --
def ask_human(state: HitlState) -> dict:
    """Agent 遇到不确定时，生成问题等待人类回答"""
    question = "我不确定你想删除哪个目录，请指定："
    # 实际项目中通过 UI 或 chat 将问题发给用户
    return {"result": question, "feedback": "等待用户回答"}

def execute_dialog_mode():
    """Agent 在条件边中判断不确定时跳转到 ask_human 节点"""
    wg = StateGraph(HitlState)
    wg.add_node("risky", risky_action)
    wg.add_node("ask", ask_human)
    wg.add_edge(START, "risky")
    # 条件边：不确定时跳到 ask 节点
    wg.add_conditional_edges("risky", lambda s: "ask" if not s.get("approved") else END,
                            {"ask": "ask", END: END})
    wg.add_edge("ask", END)
    return wg.compile(checkpointer=MemorySaver())

三种模式各有适用场景：审批模式适合高风险操作（如删除数据、发送邮件），纠正模式适合 Agent 理解有偏差时人工修正，对话模式适合 Agent 信息不足需要向人补充询问。生产项目中可以组合使用，比如先对话模式收集信息，再审批模式确认执行。

Streaming：流式输出实时反馈

用户体验角度，长时间运行的 Agent 必须提供实时反馈，否则用户会以为程序卡住了。LangGraph 原生支持 Streaming，可以流式输出每个节点的结果、State 变化以及 LLM 的 token 级流式输出。

from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator

class StreamState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str

def thinking_node(state: StreamState) -> dict:
    """模拟 LLM 思考过程"""
    return {"messages": [{"role": "ai", "content": "正在分析问题..."}], "current_step": "thinking"}

def searching_node(state: StreamState) -> dict:
    """模拟搜索过程"""
    return {"messages": [{"role": "tool", "content": "找到 3 条相关结果"}], "current_step": "searching"}

def responding_node(state: StreamState) -> dict:
    """生成最终回复"""
    return {"messages": [{"role": "ai", "content": "基于搜索结果，答案是..."}], "current_step": "responding"}

# 构建图
wg = StateGraph(StreamState)
wg.add_node("think", thinking_node)
wg.add_node("search", searching_node)
wg.add_node("respond", responding_node)
wg.add_edge(START, "think")
wg.add_edge("think", "search")
wg.add_edge("search", "respond")
wg.add_edge("respond", END)

app = wg.compile(checkpointer=MemorySaver())

# -- 方式1：stream 模式 -- 逐节点输出 --
print("=== stream 模式 ===")
for event in app.stream(
    {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
    config={"configurable": {"thread_id": "stream-001"}}
):
    node_name = list(event.keys())[0]
    print(f"[{node_name}] -> {event[node_name]}")

# -- 方式2：stream_values 模式 -- 每步输出完整 State --
print("\n=== stream_values 模式 ===")
for state in app.stream_values(
    {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
    config={"configurable": {"thread_id": "stream-002"}}
):
    print(f"current_step={state.get('current_step')}, messages_count={len(state.get('messages', []))}")

# -- 方式3：stream_events 模式 -- 细粒度事件流（含 LLM token） --
print("\n=== stream_events 模式 ===")
async for event in app.astream_events(
    {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
    config={"configurable": {"thread_id": "stream-003"}},
    version="v2"
):
    if event["event"] == "on_chain_start":
        print(f"节点开始: {event['name']}")
    elif event["event"] == "on_chain_end":
        print(f"节点完成: {event['name']}, 输出: {event.get('data', {}).get('output', '')[:80]}")

三种 Streaming 方式从粗到细各有用途：stream() 以节点为单位输出，适合展示"正在执行哪一步"的进度条；stream_values() 每步输出完整 State，适合调试和监控状态变化；astream_events() 是最细粒度的事件流，能捕获 LLM 的 token 级输出，适合实现打字机效果的实时回复。在生产项目中，推荐前端用 astream_events 做 token 级展示，后端用 stream 做节点级日志记录。

流程图：状态机流转图

• START（入口） -> Thinking（思考节点）
• Thinking（思考节点） -> Decision（条件判断）
  • 需搜索 -> Search（搜索节点）
  • 直接回答 -> Answer（回答节点）
• Search（搜索节点） -> Parallel（并行节点）
• Parallel（并行节点） -> Review（审查节点）
• Answer（回答节点） -> Review（审查节点）
• Review（审查节点）
  • 不通过（虚线）-> Thinking（思考节点）
  • 通过 -> END（出口）

面试八股

Q：LangGraph 的核心思想是什么？与 LangChain Agent 有什么区别？

A： 核心思想：把 Agent 工作流建模为有向图。节点 = 执行步骤，边 = 跳转逻辑。

区别：LangChain Agent 是黑盒（LLM 自主决定流程），难调试、难干预。LangGraph 是白盒（图结构显式定义），每步可检查 State、支持人工干预、支持并行。

简单场景用 LangChain Agent，复杂生产场景用 LangGraph。

Q：LangGraph 的五个核心概念是什么？

A： 1. State（状态）：全局共享状态，所有节点可读写。用 TypedDict 定义。 2. Node（节点）：执行单元，一个函数，接收 State 返回更新后的 State。 3. Edge（边）：节点间连接，普通边=固定跳转。 4. Conditional Edge（条件边）：根据 State 动态路由，类似 if-else。 5. Checkpoint（检查点）：每步自动保存 State 快照，支持时间旅行、人工干预、错误恢复。

Q：LangGraph 如何实现 Human-in-the-Loop？

A： 通过 Checkpoint + interrupt_before/interrupt_after 实现：

1. 编译图时指定 interrupt_before=["tools"]，表示在 tools 节点前暂停
2. Agent 执行到暂停点时停止，等待人工操作
3. 人工可以：直接继续（批准）、修改 State 后继续（纠正）、终止执行（拒绝）
4. 通过 thread_id 关联同一次执行的状态快照，支持从任意点恢复

Q：LangGraph 中 State 的 Annotated[list, operator.add] 是什么意思？

A： Annotated[list, operator.add] 定义了 State 字段的合并策略。

operator.add 表示"追加"语义：当节点返回新的 messages 时，新消息追加到已有列表，而不是覆盖。

这对消息历史很重要——每个节点都应该看到完整的对话历史，而不是只看到上一步的结果。

Q：LangGraph 如何实现 ReAct 循环？

A： 通过图结构：

1. agent 节点（调用 LLM 推理）
2. tools 节点（执行工具）
3. 条件边 should_continue：检查最后一条消息有无 tool_calls
4. agent -> (条件) -> tools -> agent -> ... 形成循环
5. 当 LLM 不再要求调用工具时，条件边路由到 END，循环结束

ReAct 的 Thought-Action-Observation 循环在 LangGraph 中体现为图中的环（cycle）。

Q：LangGraph 中 StateGraph 的核心数据结构是什么？为什么用 TypedDict？

A： 核心数据结构是 StateGraph，它以 TypedDict 作为 State 的类型定义。

使用 TypedDict 的原因有三点：1. 类型安全，IDE 可以自动补全字段名和类型，减少拼写错误；2. 合并策略声明，通过 Annotated[list, operator.add] 可以指定字段的合并方式（追加 vs 覆盖）；3. 可序列化，TypedDict 本质是 dict，可以直接序列化为 JSON 保存到 Checkpoint 中。

StateGraph 在编译时会根据 TypedDict 的字段定义自动处理节点返回值与现有 State 的合并逻辑，无需手动编写 merge 函数。

Q：ConditionEdge 和普通 Edge 的区别是什么？什么场景下用 ConditionEdge？

A： 普通 Edge：固定跳转，A 节点执行完必定跳到 B 节点。语法：graph.add_edge("A", "B")。

Conditional Edge：动态路由，根据当前 State 的内容决定下一个节点。语法：graph.add_conditional_edges("A", router_fn, {"path1": "node1", "path2": "node2"})。

使用场景：任何需要根据运行时状态做分支的地方都应该用 ConditionEdge。典型场景包括：ReAct 循环中判断是否继续调用工具、多工具选择（根据 LLM 输出决定调用哪个工具）、错误处理（根据错误类型决定重试还是终止）、质量审查（根据评分决定通过还是重试）。

Q：Checkpoint 在 LangGraph 中的作用是什么？MemorySaver 和其他 Checkpointer 有什么区别？

A： Checkpoint 作用：在每个节点执行后自动保存 State 快照，实现三大功能——时间旅行（回到任意历史状态）、人工干预（暂停在指定节点等待人工操作后继续）、错误恢复（从失败点重试而非从头开始）。

MemorySaver：内存级 Checkpointer，数据存在进程内存中，重启后丢失。适合开发调试和单机部署。

SqliteSaver / PostgresSaver：持久化到数据库，重启后数据仍在。适合生产环境，支持多实例共享状态。

选型建议：开发用 MemorySaver，生产用 PostgresSaver。切换只需改一行 compile(checkpointer=xxx)，图定义完全不变。

Q：LangGraph 中 Human-in-the-Loop 的三种模式分别是什么？如何实现？

A： 模式1：审批模式。编译时设置 interrupt_before=["node_name"]，在指定节点前自动暂停，等待人工批准后继续。适合高风险操作。

模式2：纠正模式。暂停后通过 app.update_state(config, values={...}) 修改 State 内容，再调用 app.invoke(None, config) 继续。适合 Agent 理解偏差时人工修正。

模式3：对话模式。在条件边中判断是否需要人工输入，跳转到专门的人机交互节点等待用户回答。适合 Agent 信息不足需要补充询问的场景。

三种模式可以组合使用，通过 thread_id 关联同一次执行的状态快照。

Q：项目选型时，LangGraph 和 LangChain Agent 怎么选？给出决策标准。

A： 选 LangChain Agent 的场景：1. 快速原型验证，不在乎流程控制；2. 简单的单步或少步 Agent，不需要循环和并行；3. 不需要人工干预和状态持久化；4. 团队已经在使用 LangChain 生态，不想引入新依赖。

选 LangGraph 的场景：1. 需要精细控制 Agent 每一步的流程；2. 需要人工干预（审批、纠正、对话）；3. 需要并行执行多个节点；4. 需要状态持久化和错误恢复；5. 生产环境部署，需要可调试性和可观测性；6. 复杂工作流，涉及子图、多 Agent 协作。

一句话标准：能用 LangChain Agent 在 10 分钟跑通的简单场景就用它，但凡需要"控制力"就上 LangGraph。

Q：LangGraph 的节点和边，和传统工作流有什么区别？

A： 核心区别 = 灵活性 + 循环能力：

传统工作流：节点和跳转边都是固定写死的，一步错就全盘卡壳，也不支持循环执行。适用于流程确定、场景固定的简单业务。

LangGraph：

1. 支持条件边——下一个执行节点不是固定的，由大模型的推理结果动态决定。类似 if-else 但决策者是 LLM 而非硬编码逻辑。
2. 支持循环执行——Agent 可以自主反复试错、迭代优化，直到完成任务。这是 Agent 能自主完成复杂任务的核心原因。

一句话话术：传统工作流是确定性流水线，LangGraph 是不确定性自主推理链。条件边让 Agent 能"自主决策"下一步，循环执行让 Agent 能"试错优化"直到完成。

课后练习

第 1 题（单选）

LangGraph 的核心思想是什么？

A. 用消息传递实现 Agent 通信 B. 把 Agent 工作流建模为有向图 C. 用角色扮演组织 Agent 协作 D. 用 SOP 标准流程驱动 Agent

答案与解析

答案：B

解析：LangGraph 核心思想：把 Agent 工作流建模为有向图。节点=执行步骤，边=跳转逻辑，条件边=动态路由。

第 2 题（单选）

LangGraph 中条件边（Conditional Edge）的作用是？

A. 固定从一个节点跳到另一个节点 B. 根据当前 State 动态决定下一个节点 C. 保存状态快照 D. 定义节点函数

答案与解析

答案：B

解析：条件边根据当前 State 动态路由，类似 if-else。例如：检查是否有 tool_calls，有则跳到 tools 节点，无则结束。

第 3 题（单选）

LangGraph 的 Checkpoint 机制不支持以下哪个功能？

A. 时间旅行（回到任意检查点） B. 人工干预（暂停等待审核） C. 自动生成工具代码 D. 错误恢复（从失败点重试）

答案与解析

答案：C

解析：Checkpoint 支持时间旅行、人工干预、错误恢复。不支持自动生成工具代码——工具需要人工定义。

第 4 题（单选）

Annotated[list, operator.add] 在 LangGraph State 中的含义是？

A. 每次覆盖整个列表 B. 新返回的列表追加到已有列表 C. 只保留最新的一个元素 D. 按字典序排序

答案与解析

答案：B

解析：operator.add 表示"追加"合并策略：节点返回的新消息追加到已有列表，而非覆盖。确保每个节点能看到完整历史。

第 5 题（单选）

LangGraph 相比 LangChain Agent 的最大优势是？

A. 运行速度更快 B. 流程可控、可调试、支持人工干预 C. 不需要 LLM 就能运行 D. 支持更多的工具类型

答案与解析

答案：B

解析：LangGraph 最大优势是白盒可控：图结构显式定义流程、每步可检查 State、支持人工干预和并行执行。适合生产级复杂场景。

第 6 题（多选）

以下哪些是 LangGraph 的核心概念？（多选）

A. State（状态） B. Node（节点） C. Edge（边） D. Gradient（梯度）

答案与解析

答案：A、B、C

解析：LangGraph 五大核心概念：State、Node、Edge、Conditional Edge、Checkpoint。Gradient 是深度学习概念，不属于 LangGraph。

第 7 题（多选）

关于 LangGraph 的 Human-in-the-Loop，以下说法正确的是？（多选）

A. 通过 interrupt_before 在指定节点前暂停 B. 人工可以批准、修改或拒绝继续执行 C. 通过 thread_id 关联同一次执行的状态 D. 暂停后会丢失所有状态需要重新开始

答案与解析

答案：A、B、C

解析：暂停后状态保存在 Checkpoint 中，不会丢失。通过 thread_id 可以恢复到暂停点继续执行。

第 8 题（多选）

LangGraph 相比 LangChain Agent 的改进包括？（多选）

A. 流程从黑盒变为白盒（图结构显式定义） B. 支持人工干预 C. 支持并行执行 D. 不再需要 LLM 参与决策

答案与解析

答案：A、B、C

解析：LangGraph 仍然需要 LLM 参与决策（在 agent 节点中调用 LLM）。改进的是流程可控性、可调试性、人工干预和并行支持。

第 9 题（单选）

LangGraph 中 StateGraph 的核心数据结构使用什么类型来定义 State？

A. dataclass B. TypedDict C. Pydantic BaseModel D. 普通 dict

答案与解析

答案：B

解析：StateGraph 使用 TypedDict 作为 State 的类型定义。TypedDict 提供类型安全、支持 Annotated 合并策略声明，且本质是 dict 可直接序列化到 Checkpoint。

第 10 题（单选）

关于 LangGraph 的 ConditionEdge，以下说法正确的是？

A. ConditionEdge 只能连接两个固定节点 B. ConditionEdge 根据当前 State 动态决定下一个节点 C. ConditionEdge 用于保存状态快照 D. ConditionEdge 只能在图的入口使用

答案与解析

答案：B

解析：ConditionEdge 根据当前 State 的内容动态路由，通过路由函数返回字符串键值映射到不同节点。适合 ReAct 循环判断、多工具选择、错误处理等分支场景。

第 11 题（多选）

关于 LangGraph 的 Checkpointer，以下说法正确的是？（多选）

A. MemorySaver 将状态保存在内存中，重启后丢失 B. PostgresSaver 可以实现跨实例共享状态 C. Checkpointer 支持时间旅行（回到任意历史状态） D. 切换不同 Checkpointer 需要修改图结构定义

答案与解析

答案：A、B、C

解析：切换 Checkpointer 只需修改 compile(checkpointer=xxx) 一行代码，图定义完全不变。MemorySaver 适合开发，PostgresSaver 适合生产。两者都支持时间旅行和人工干预。

第 12 题（多选）

关于 LangGraph 子图（Subgraph），以下说法正确的是？（多选）

A. 子图可以封装一组相关节点作为独立模块 B. 子图的 State 必须与主图 State 完全相同 C. 子图可以独立编译和测试 D. 子图通过节点函数桥接主图与子图的 State 转换

答案与解析

答案：A、C、D

解析：子图的 State 与主图的 State 是独立的，通过节点函数做桥接转换。子图的好处是逻辑分层、复用性和独立测试。

第 13 题（多选）

LangGraph 支持哪些 Streaming 输出方式？（多选）

A. stream() -- 以节点为单位输出 B. stream_values() -- 每步输出完整 State C. astream_events() -- 细粒度事件流（含 LLM token 级输出） D. stream_tokens() -- 仅输出 LLM 的 token

答案与解析

答案：A、B、C

解析：LangGraph 原生支持三种 Streaming 方式：stream() 节点级、stream_values() 状态级、astream_events() 事件级（含 token）。没有 stream_tokens() 这个方法。