第2章图形工作流¶

学习目标¶

掌握节点（Node）、边（Edge）、状态（State）三大基本概念
区分有向无环图（DAG）与循环图，理解循环在 Agent 中的价值
能够使用条件边实现动态路由
理解并行执行的语义与状态合并机制
完成图的编译与执行，并能流式消费输出

问题导入¶

上一章我们跑通了第一个 LangGraph 图，但只用了 add_edge 把节点串成一条直线。真实业务里，工作流往往不是直线："如果检索结果为空就回退到人工兜底""如果模型输出含代码就走去执行节点，否则直接回复"——这些都需要条件边。再比如，为了加速，我们希望同时调用搜索和数据库两个工具——这需要并行。本章把图的"骨架"讲透，让你能画出任意复杂度的控制流。

核心概念¶

1. 节点 Node¶

节点是图中的"做事单元"，本质上是一个函数：接收当前 State，返回要更新的字段。定义节点时只需保证：

函数签名是 def node(state: State) -> dict:。
返回值是一个 dict，键必须是 State 中已声明的字段。
每个节点在图中有唯一名字（add_node(name, fn)）。

节点内部可以调用 LLM、工具、数据库、外部 API——LangGraph 不关心你做什么，只关心"你改了哪些状态"。

2. 边 Edge¶

边描述"执行完当前节点后去哪"。LangGraph 提供三种边：

普通边 add_edge(A, B)：A 执行完一定去 B，确定性跳转。
条件边 add_conditional_edges(A, router_fn, mapping)：A 执行完调用 router_fn(state)，根据返回值查 mapping 决定下一节点，实现动态路由。
入口边 add_edge(START, A)：图的起点；出口边 add_edge(B, END)：到 END 即结束。

3. 状态 State¶

State 是节点间共享的数据容器，用 TypedDict 定义。关键字段可标注 Annotated[T, reducer]，指定合并策略。常用 reducer：

add_messages：追加消息（多轮对话）。
自定义 operator.add：列表追加。
不标注 reducer：直接覆盖。

State 的设计直接决定图的复杂度。原则是：只放跨节点共享的数据，节点内部临时变量不要塞进 State。

4. 循环图与 DAG¶

DAG（有向无环图）：节点按拓扑顺序执行一遍，适合流水线。LCEL 天然是 DAG。
循环图：节点可被反复执行，适合 ReAct 这类"思考-行动-观察"循环。

LangGraph 支持循环，但需注意终止条件。常见做法是用条件边判断：当"任务完成"时走 END，否则继续循环。为防止死循环，可设置 recursion_limit（默认 25 步）。

5. 条件边与动态路由¶

条件边是 LangGraph 表达力的核心。router_fn 读取 State，返回一个字符串（或节点名），mapping 把字符串映射到目标节点。典型用法：

graph.add_conditional_edges(
    "agent",
    lambda state: "tools" if state["messages"][-1].tool_calls else "end",
    {"tools": "tools", "end": END},
)

这让 Agent 能根据"模型这一轮是否要调工具"动态决定下一步。

6. 并行执行¶

LangGraph 支持两种并行：

扇出（fan-out）：从一个节点连多条边到不同节点，它们会被并行调度。
扇入（fan-out → fan-in）：多个并行节点执行完后汇合到一个节点。

并行时状态合并依赖 reducer。例如两个并行节点都向 messages 写入，add_messages 会把两份结果都保留；若用默认覆盖策略，后到的会覆盖先到的，造成数据丢失。

7. 图编译与执行¶

compile() 把 builder 转为可执行的 Runnable。编译时可注入：

checkpointer：状态持久化（下一章详述）。
interrupt_before / interrupt_after：在指定节点前后暂停，用于人在回路。

执行方式：

graph.invoke(input)：阻塞执行，返回最终 State。
graph.stream(input)：生成器，逐节点输出增量。
graph.ainvoke / graph.astream：异步版本。

原理图¶

flowchart LR
    START([START]) --> agent[agent<br/>调用 LLM]
    agent -- "有 tool_calls" --> tools[tools<br/>执行工具]
    tools --> agent
    agent -- "无 tool_calls" --> judge{judge<br/>条件边}
    judge -- "需要总结" --> summary[summary<br/>总结节点]
    judge -- "可直接回复" --> END
    summary --> END([END])

    classDef loop fill:#fef3c7,stroke:#f59e0b;
    class agent,tools loop;

图：一个带循环与条件边的典型 Agent 图。agent 与 tools 构成 ReAct 循环（黄色）；循环结束后由 judge 条件边决定是去 summary 还是直接结束。

示例代码¶

下面构建一个"带工具循环 + 条件收尾"的完整工作流。

import os
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. 状态定义
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 工具：模拟天气查询
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    # 实际项目调用真实天气 API
    return f"{city} 今天晴，25℃。"

tools_list = [get_weather]
llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
llm_with_tools = llm.bind_tools(tools_list)

# 3. 节点：Agent 思考
def agent(state: State) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 4. 节点：执行工具
def call_tools(state: State) -> dict:
    # 取出最后一条消息中的工具调用，逐个执行
    from langchain_core.messages import ToolMessage
    last_msg = state["messages"][-1]
    results = []
    for call in last_msg.tool_calls:
        # 通过名字找到对应工具并调用
        fn = {t.name: t for t in tools_list}[call["name"]]
        results.append(ToolMessage(content=fn.invoke(call["args"]), tool_call_id=call["id"]))
    return {"messages": results}

# 5. 节点：总结（演示条件收尾）
def summary(state: State) -> dict:
    last = state["messages"][-1].content
    return {"messages": [{"role": "assistant", "content": f"总结：{last[:30]}..."}]}

# 6. 条件路由函数
def should_continue(state: State) -> Literal["tools", "summary", "end"]:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    # 如果回复较短，直接结束；较长则总结
    if len(last.content) > 50:
        return "summary"
    return "end"

# 7. 构建图
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", call_tools)
builder.add_node("summary", summary)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "summary": "summary",
    "end": END,
})
builder.add_edge("tools", "agent")        # 工具执行完回到 agent，形成循环
builder.add_edge("summary", END)

graph = builder.compile()

# 8. 流式执行，观察每一步
if __name__ == "__main__":
    for event in graph.stream({
        "messages": [{"role": "user", "content": "北京天气怎么样？然后给我一段 60 字的旅行建议。"}],
    }, stream_mode="updates"):
        for node_name, value in event.items():
            print(f"--- 节点 {node_name} 输出 ---")
            print(value)

运行流程：用户提问 → agent 调用 LLM（输出含 get_weather 工具调用）→ 条件边路由到 tools → 执行工具得到天气 → 回到 agent → LLM 基于天气生成 60 字建议（无工具调用且超过 50 字）→ 条件边路由到 summary → 总结后结束。一次调用就完整跑完了"循环 + 条件收尾"。

企业案例¶

案例 1：复杂审批流编排¶

某保险公司理赔流程包含"录入→初审→复核→（金额大于 5 万时）专家会签→出险确认→赔付"六个环节。团队用 LangGraph 建模：每个环节是一个节点，金额阈值通过条件边分流，专家会签用并行节点（多位专家同时评审）+ fan-in 汇总。此前用状态机库实现需 300 行，迁移到 LangGraph 后约 120 行，且天然支持持久化——会签环节常需等待数小时，检查点让"挂起-恢复"成为内置能力。

案例 2：智能问答的检索-生成-校验三段式¶

知识问答团队把流程拆为 检索 → 生成 → 校验：校验节点判断生成内容是否与检索片段冲突，冲突则回到检索（换关键词重试），最多重试 3 次后走兜底回复。条件边 + recursion_limit 完美表达了这一逻辑，且每次重试的中间状态都被记录，便于离线分析"为什么这一轮没答对"。

常见问题¶

Q1：循环图会不会无限执行？ 默认 recursion_limit=25，超过会抛 RecursionError。可在 invoke 时传入 config={"recursion_limit": 50} 调整。更重要的是设计好终止条件（条件边导向 END）。

Q2：并行节点执行顺序是否确定？ 并行节点会被并发调度，完成顺序不确定，但汇合节点会等所有前驱完成后执行。汇合时状态按 reducer 合并，因此只要 reducer 设计正确，结果与执行顺序无关。

Q3：条件边的 mapping 可以省略吗？ 可以。若不传 mapping，router_fn 的返回值会被直接当作目标节点名。传 mapping 是为了把"业务标签"与"节点名"解耦，便于重命名。

Q4：一个节点能同时连到多个目标吗？ 能。用 add_conditional_edges 返回多个目标，或用普通边扇出。但注意：条件边每次只选一条路径；要真正并行多路径，用扇出 + 多条普通边。

Q5：stream_mode 有哪些取值？ 常用 values（每步完整 State）、updates（每步增量）、messages（逐 token 流式 LLM 输出）。调试时 updates 最直观。

面试题¶

1. 简述节点、边、状态三者的关系。 答：节点是改状态的函数，边是节点间的控制流，状态是节点间共享的数据容器。节点只返回要更新的字段，由 reducer 合并到状态中，边决定下一个执行的节点。

2. 条件边与普通边的区别？举例说明。 答：普通边是确定性跳转（A→B 恒成立）；条件边根据当前状态动态选择目标，例如 Agent 输出含 tool_calls 时走 tools 节点，否则走 END。

3. 为什么并行场景必须用 reducer？ 答：并行节点并发写同一字段，若无 reducer 后写会覆盖先写；reducer（如 add_messages）显式定义合并策略，保证结果与执行顺序无关。

4. recursion_limit 防的是什么？默认值？ 答：防止循环图无限执行。默认 25，可通过 config 调整。但根本之道是设计好终止条件。

5. invoke 与 stream 的区别？调试时推荐哪个？ 答：invoke 阻塞返回最终状态；stream 生成器逐节点输出增量。调试推荐 stream(stream_mode="updates")，能清晰看到每个节点的输出。

练习题¶

修改示例，增加一个 translate 节点：当用户提问包含英文时，先翻译成中文再交给 agent。提示：新增条件边在 START 之后做语言检测。
实现一个 fan-out / fan-in 图：节点 split 同时调用 search_web 与 search_db，二者都完成后汇合到 merge 节点合并结果。注意为结果列表设置 reducer。
给示例的循环加一个"最多重试 2 次工具调用"的限制，超过则强制走 END 并返回"无法完成"提示。

小结¶

本章把图形工作流的骨架讲透：节点负责改状态，边负责控流向，状态负责承数据。条件边让图能动态路由，循环让图能 ReAct，并行让图能加速。掌握这些后，你已能表达绝大多数控制流。下一章我们将聚焦"基本构件"——LangGraph 预置的 ToolNode、AgentNode 等，它们能让你少写大量样板代码。

第2章 图形工作流¶