10 · LangGraph 实战：智能体编排

本章目标：用 LangGraph 把第 2 章的手写循环"画"成图，并实现三件手写循环很难做的事——工具循环、人工介入（Human-in-the-loop）、多智能体协作。

本章你将学到

用图实现 ReAct 循环（模型节点 ⇄ 工具节点）；
用官方预置的 create_react_agent 快速搭建 Agent；
用 interrupt 实现"执行敏感操作前暂停等人批准"；
Supervisor（主管）模式下多 Agent 分工协作。

前置知识

第 09 章的 State / 节点 / 边 / Checkpointer；
第 02 章的 Agent 循环心智模型。

10.1 把"手写循环"画成图

第 2 章的循环可以画成一张非常简单的图：

              ┌──────────────────────────────┐
              │                              │
   START ──► agent ──(有 tool_calls)──► tools │
              │                              │
              └──────(无 tool_calls)─────────► END

agent 节点：调用绑定工具的模型，可能返回文本或 tool_calls；
tools 节点：执行工具，把结果回填；
循环：tools 执行完把边连回 agent，让模型基于工具结果继续推理；
出口：模型这次没有 tool_calls，直接到 END。

下面完整实现"航班助手"，图版。

"""LangGraph 实现的 ReAct 航班助手（图版手写循环）。"""
import os
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


# ---------- 1. 定义工具 ----------
@tool
def get_flight_status(flight_no: str) -> str:
    """查询航班实时状态。

    Args:
        flight_no: 航班号，如 CA1501
    """
    return {
        "CA1501": "已起飞，预计 11:35 抵达",
        "MU5102": "延误约 40 分钟",
        "CZ3901": "准点",
    }.get(flight_no.upper(), f"未查询到 {flight_no}")


@tool
def get_flight_price(flight_no: str, travel_date: str) -> str:
    """查询经济舱票价。

    Args:
        flight_no: 航班号，如 CA1501
        travel_date: 出行日期，YYYY-MM-DD
    """
    return f"{flight_no} 在 {travel_date} 经济舱票价约 780 元"


tools = [get_flight_status, get_flight_price]


# ---------- 2. State ----------
class State(TypedDict):
    messages: Annotated[list, add_messages]   # 对话历史用归约器累积


# ---------- 3. 节点 ----------
def agent_node(state: State) -> dict:
    """模型节点：读全部消息，输出回答或工具调用。"""
    resp = model.bind_tools(tools).invoke(state["messages"])
    return {"messages": [resp]}


# ---------- 4. 构图 ----------
builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))     # 官方预置：执行 tool_calls 并回填 ToolMessage

builder.add_edge(START, "agent")
# 关键：模型返回 tool_calls → 去 tools；否则 → END
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")             # 工具结果回到模型，形成循环

graph = builder.compile(checkpointer=MemorySaver())


# ---------- 5. 运行 ----------
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "flight-1"}}
    result = graph.invoke(
        {"messages": [("user", "帮我查 CA1501 现在什么状态，以及明天票价？")]},
        config=config,
    )
    print(result["messages"][-1].content)

运行一次，模型会先调 get_flight_status、再调 get_flight_price，最后汇总回答——两次工具调用发生在同一个图的一次执行里，靠的就是"agent ⇄ tools"这条循环边。

10.2 预置 Agent：create_react_agent

上面那张图（agent + tools + 条件边）太常用了，LangGraph 直接内置：

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model, tools, checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "user-7"}}
result = agent.invoke(
    {"messages": [("user", "MU5102 现在延误了吗？")]}, config=config
)
print(result["messages"][-1].content)

create_react_agent 底层就是 10.1 那张图，还内置了：

系统提示词（让模型按 ReAct 风格"先想后做"）；
工具异常捕获（工具抛异常回填错误消息，模型自己换策略）；

可配置 recursion_limit（最大步数保护，防死循环）：

agent.invoke({"messages": [...]}, config={"configurable": {"thread_id": "u1"}, "recursion_limit": 15})

从手写循环（第 2 章）→ 手写图（10.1）→ create_react_agent（10.2），你看到的是同一件事的三层实现。理解底层后，用预置组件心里才不慌。

10.3 Human-in-the-loop：敏感操作人工审批

真实业务里，Agent 不能什么都自己干——退票、转账、发消息给客户这类操作需要人确认。LangGraph 的 interrupt 让你在任意节点"挂起"，等人批准后再继续。

类比：工作流引擎的"审批节点"，流程停在等待人工动作，批准后才往下走。

10.3.1 用 interrupt 挂起与恢复

"""人工审批示例：转账前暂停等批准。"""
import os
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

load_dotenv()


class State(TypedDict):
    messages: Annotated[list, add_messages]
    to: str
    amount: float


def transfer_node(state: State) -> dict:
    """执行转账。执行前 interrupt 暂停，等人返回 'approve' 或 'reject'。"""
    decision = interrupt(
        {"type": "confirm_transfer", "to": state["to"], "amount": state["amount"]}
    )
    if decision == "approve":
        return {"messages": [AIMessage(content=f"已向 {state['to']} 转账 {state['amount']} 元。")]}
    return {"messages": [AIMessage(content="转账已被取消。")]}


builder = StateGraph(State)
builder.add_node("transfer", transfer_node)
builder.add_edge(START, "transfer")
builder.add_edge("transfer", END)

graph = builder.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "tx-001"}}

# 第一次 invoke：执行到 interrupt 处暂停，返回"待审批"信息
snapshot = graph.invoke(
    {"messages": [("user", "给供应商转账 5000 元")], "to": "supplier", "amount": 5000},
    config=config,
)
print(snapshot)   # 注意：这一步不会真正转账，只是"停在审批点"

# 人类批准后，用 Command(resume=...) 继续执行
final = graph.invoke(Command(resume="approve"), config=config)
print(final["messages"][-1].content)   # "已向 supplier 转账 5000 元。"

10.3.2 拦截每一次工具调用

上面的例子是"手动在某节点 interrupt"。更常用的是：每次模型想调用工具，都先暂停给人类看，尤其是高风险工具。用 create_react_agent(..., interrupt_before=["tools"])：

from langgraph.prebuilt import create_react_agent

# 模型每次请求调工具前都中断
agent = create_react_agent(model, tools, interrupt_before=["tools"])

config = {"configurable": {"thread_id": "tx-002"}}
agent.invoke({"messages": [("user", "帮我退订 CA1501 的票")]}, config=config)
# ↑ 这里执行会停在"即将调用工具"之前

# 人类检查后决定放行还是驳回
# agent.invoke(Command(resume="continue"), config=config)   # 放行
# agent.invoke(Command(resume="reject"), config=config)     # 驳回

生产提示：interrupt 的内容要能被人看懂（谁、做什么、多少钱），审批界面直接展示它。审批动作最好走独立接口并记录操作人（审计）。

10.4 多轮对话：thread_id 已经是会话记忆

配上 Checkpointer 后，同一个 thread_id 的多轮调用天然共享历史（第 9.6 节讲过）。对 Agent 同样适用：

agent = create_react_agent(model, tools, checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "user-1001"}}

agent.invoke({"messages": [("user", "记住，我的航班号是 CA1501")]}, config=config)
agent.invoke({"messages": [("user", "查一下我的航班状态")]}, config=config)
# 第二句不需要再报航班号——模型从历史里读到了

这就是"短期对话记忆"的完整实现。长期记忆（跨会话、向量化存储）见第 11 章。

10.5 多 Agent 协作：Supervisor 模式

需求复杂时，让一个 Agent 干所有事会越来越不可控。更稳的做法是分工：一个 Supervisor（主管） 负责任务分派，几个 Worker（工人） 各管一块（如"航班查询"、"退改签政策"、"客诉升级"）。

架构：

                ┌────────── supervisor ──────────┐
                │  (模型：决定把任务分给谁/直接回答)  │
                └──────┬──────┬──────┬──────────┘
                       │      │      │
                   flight   policy  escalate
                 (航班查询) (政策问答) (转人工)
                       │      │      │
                └──────┴──────┴──────┘
                      回到 supervisor 继续(循环)

"""Supervisor 多智能体：主管分派 + 三个工人 + 循环。"""
import os
from typing import Annotated, Literal, TypedDict

from dotenv import load_dotenv
from langchain_core.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


class State(TypedDict):
    messages: Annotated[list, add_messages]
    worker: str          # 记录当前分派给谁


# ---------- 三个 Worker 节点（示例：每类一个专门 Prompt）----------
FLIGHT_SYS = "你是航班查询专家，只回答航班状态/票价相关问题。"
POLICY_SYS = "你是退改签政策专家，只回答退票改签政策问题。"
ESCALATE_SYS = "你是升级专员，生成一条转人工工单摘要。"


def make_worker(sys_prompt: str):
    def worker(state: State) -> dict:
        reply = model.invoke([("system", sys_prompt), *state["messages"]])
        return {"messages": [AIMessage(content=reply.content)]}
    return worker


# ---------- Supervisor 节点：决定分派或结束 ----------
def supervisor_node(state: State) -> dict:
    """让模型用结构化输出决定下一步：派给哪个 worker，还是结束。"""
    class NextStep(TypedDict):
        action: Literal["flight", "policy", "escalate", "finish"]

    resp = model.with_structured_output(NextStep).invoke(
        [("system", "根据用户问题选择处理方：flight/policy/escalate；如果已有足够信息回答则 finish。"),
         *state["messages"]]
    )
    return {"worker": resp["action"]}


def route_supervisor(state: State) -> str:
    return state["worker"]


builder = StateGraph(State)
builder.add_node("supervisor", supervisor_node)
builder.add_node("flight", make_worker(FLIGHT_SYS))
builder.add_node("policy", make_worker(POLICY_SYS))
builder.add_node("escalate", make_worker(ESCALATE_SYS))

builder.add_edge(START, "supervisor")
# 分派：flight/policy/escalate 执行完都回到 supervisor（循环：让主管看结果再决定）
builder.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {"flight": "flight", "policy": "policy", "escalate": "escalate", "finish": END},
)
builder.add_edge("flight", "supervisor")
builder.add_edge("policy", "supervisor")
builder.add_edge("escalate", "supervisor")

graph = builder.compile()

result = graph.invoke(
    {"messages": [HumanMessage(content="航班起飞前4小时退票扣多少钱？")]}
)
print(result["messages"][-1].content)

工作过程：

supervisor 判断问题属于"政策" → 派给 policy；
policy 用专门 Prompt 回答；
结果回到 supervisor，主管认为可以回答了 → finish → END。

要点：Worker 之间互不可见，只有 Supervisor 掌握全局——这既是优点（隔离、可各自调优）也是缺点（主管成为瓶颈与单点）。多 Agent 模式的选择权衡见第 11 章。

10.6 综合实战：客服 + 工具 + 人工审批

把本章组件拼成一个较完整的场景：用户要求退票 → Agent 查询订单 → 计算出退款金额 → 人工确认 → 执行。

"""综合：RAG 政策问答 + 工具 + 人工审批 的 Agent 图（骨架示例）。"""
# 1) 查询工具（模拟读订单）
@tool
def get_booking(booking_no: str) -> str:
    """查询订单信息。"""
    return '{"booking_no": "BK001", "flight": "CA1501", "fare": 1000, "state": "已出票"}'


# 2) 退票工具（高风险，模型可以调用但会被 interrupt 拦下）
@tool
def refund_ticket(booking_no: str) -> str:
    """执行退票。"""
    return f"订单 {booking_no} 已退票"


tools = [get_booking, refund_ticket]

# 3) 用 create_react_agent + interrupt_before 拦住所有工具调用
agent = create_react_agent(model, tools, interrupt_before=["tools"])

config = {"configurable": {"thread_id": "refund-001"}}
agent.invoke(
    {"messages": [("user", "帮我退掉订单 BK001 的票")]}, config=config
)
# ↑ 停在这里：模型想先查订单(get_booking)，等人确认

# 人工确认放行（真实系统里这里是审批接口，并记录操作人）
agent.invoke(Command(resume="continue"), config=config)
# 模型继续：查到订单 → 调用 refund_ticket → 又会暂停等确认

agent.invoke(Command(resume="continue"), config=config)
print("退票流程完成（每一步都经过人工确认）")

生产系统会在此基础上叠加：风险分级（只对退款/外呼这类敏感工具 interrupt，查询类不拦）、审批超时（一定时间未批准自动驳回）、审计日志（谁在什么时候批准了什么）。

10.7 小结

工具循环：agent ⇄ tools 两条边就实现了第 2 章的手写循环，tools_condition 负责判断出口；
快速上手：create_react_agent 内置了这套图，还带异常处理与步数保护；
人工介入：interrupt() 挂起图等人类输入，Command(resume=...) 恢复——敏感操作必用；
会话记忆：Checkpointer + thread_id 已解决多轮对话；
多 Agent：Supervisor 模式把"一个什么都干"拆成"主管分派 + 专人专岗"，更可控；
至此，Agent 的控制流问题已经解决。下一章上升到工程原理层面——记忆、规划、反思与设计模式选型：第 11 章。

练习

给 10.1 的图加第三个工具 get_weather，验证模型能一次调用多个工具、按序执行、汇总回答。
把 10.3 的审批改成"金额超过 1000 必须审批，否则放行"——在 transfer_node 里用 if state["amount"] > 1000 决定是否 interrupt。
在 10.5 的 Supervisor 里给 escalate 工人返回内容加一段"工单号"，并让 Supervisor 看到后自动 finish。
用 stream(stream_mode="updates") 观察 10.1 一次问答的每一步，画出实际的执行路径。

参考答案

<details>
<summary>练习1（加 get_weather，多工具调用）</summary>

@tool
def get_weather(city: str) -> str:
    """查询城市天气。"""
    ...

tools = [get_flight_status, get_flight_price, get_weather]

graph = StateGraph(MessagesState)
graph.add_node("agent", create_tool_calling_agent(llm, tools))
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition)
graph.add_edge("tools", "agent")

问"CA1501 什么状态？明天上海天气如何？"时，模型一轮可同时输出 两个 tool_calls，ToolNode 按序执行并把两个结果一并回填，模型再汇总回答。验证点：

模型并行声明多个工具调用（同一个 AIMessage 里多个 tool_calls）；
ToolNode 全部执行完才回到 agent（顺序执行，不阻塞模型决策）；
若中间有依赖（如先查航班再查该航班的票价），模型会分两轮调用——无依赖并行、有依赖串行，观察它是否自动做到。
</details>

<details>
<summary>练习2（金额分级审批）</summary>

在 transfer_node 里按金额决定是否暂停：

def transfer_node(state: State) -> State:
    amount = state["amount"]
    if amount > 1000:
        user_confirm = interrupt({"action": "转账", "amount": amount,
                                  "msg": f"金额 {amount} 元，需人工审批"})
        if user_confirm != "approve":
            return {"result": "审批未通过，转账取消"}
    # 放行：执行真实转账逻辑
    return {"result": f"转账 {amount} 元成功"}

# 调用侧
result = agent.invoke({"messages": [("human", "转 2000 给张三")]}, config=config)
# 2000 > 1000 → interrupt 挂起，等人工输入

agent.invoke(Command(resume="approve"), config=config)   # 继续
agent.invoke(Command(resume="reject"), config=config)    # 或拒绝

要点：

分级规则写在节点代码里（if amount > 1000），而不是交给模型判断——审批策略是硬规则，必须程序化；
interrupt 只是"挂起 + 等人给输入"，恢复后从断点继续，State 不丢；
生产扩展：分级阈值放配置中心、interrupt 的 payload 里带上审批所需全部字段（操作人、金额、去向），配合第 10.6 节的审批超时与审计。
</details>

<details>
<summary>练习3（escalate 工单号 + Supervisor 自动收尾）</summary>

工人 escalate 返回带工单号的结构化结果，Supervisor 识别后 finish：

def escalate_worker(state: State) -> State:
    ticket_no = f"ESC-{uuid4().hex[:8]}"
    return {"messages": [{"role": "assistant",
                          "content": f"已转人工处理，工单号：{ticket_no}。"}]}

def supervisor(state: State) -> str:
    last = state["messages"][-1].content
    if "工单号" in last:                # 收到升级信号 → 本轮结束
        return "finish"
    # 其余逻辑：按意图分派给 worker 或答复
    return route_worker(state)

用条件边接上：

graph.add_conditional_edges("supervisor", supervisor,
                            {"worker": "escalate", "finish": END})

验证路径：用户说"我投诉，要找人工" → Supervisor 派给 escalate → 返回含"工单号"的文案 → Supervisor 下一轮看到信号直接 finish。效果：升级有凭证（工单号可追踪）、流程有终点（不会让 Supervisor 反复追问）——多 Agent 里给工人一个"明确结束信号"是避免死循环的常用约定。
</details>

<details>
<summary>练习4（stream 观察真实执行路径）</summary>

events = graph.stream(
    {"messages": [("human", "CA1501 什么状态？明早北京天气？")]},
    stream_mode="updates",
)
for step in events:
    print(step)

一次问答的典型事件序列：

{'agent': {'messages': [AIMessage(tool_calls=[get_flight_status, get_weather])]}}
{'tools': {'messages': [ToolMessage(...), ToolMessage(...)]}}
{'agent': {'messages': [AIMessage(content='航班…；北京天气…')]}}

把事件画成执行路径：agent → tools（并行两个工具）→ agent → END。对照 10.1 的图检查：工具结果是否都正确回填、是否有多余的重复调用（模型又调了一次已查过的工具）、是否有超出 max_iterations 的兜底触发。这张轨迹图就是生产排障的核心工具——线上每个异常回答，第一步就是拉出这条执行路径看哪一步走错了（第 12 章 tracing 的落地形态）。
</details>

▕┃