LangGraph：把 Agent 循环一步步改成状态图

这篇不是 LangGraph API 速查表，而是一堂迁移练习。默认你已经完成了 裸 API 手写 Agent，知道 messages、tool_calls 和最大轮数分别在做什么。现在我们不换任务，只把同一个循环一点点改写成图。

贯穿全文的任务是“企业 IT 服务台助手”:

• 用户问公司制度，助手查知识库后回答，并带上文档标题。
• 用户问工单，助手调用只读的工单查询工具。
• 用户要关闭工单，助手先准备动作，必须等人工确认后才能执行。

你可以把代码片段复制到自己的练习文件中，每完成一步就运行一次。

1. 先准备一个最小练习环境

在你自己的虚拟环境里安装 LangGraph，以及一个能实现 LangChain BaseChatModel 接口的模型适配器。下面以 OpenAI 适配器为例，换成其他供应商时，图的代码不用跟着改。

pip install -U langgraph langchain-openai

设置模型所需的环境变量。没有 API Key 时，也可以先把 model 换成你自己的 mock 模型，先验证图的流转。

export OPENAI_API_KEY="你的 key"
export OPENAI_MODEL="gpt-4o-mini"

运行观察：安装后执行 python -c "import langgraph; print(langgraph.__version__)"，记下版本号。教程中的类名来自当前 LangGraph Python API；版本升级后先对照官方变更记录检查导入路径。

常见错误：

• ModuleNotFoundError: langgraph：确认安装命令运行在当前虚拟环境，而不是系统 Python。
• 模型适配器报认证错误：先单独调用一次模型，确认 Key、模型名和网络都正常，再排查图。

想一想：为什么先验证模型调用，再开始写图？如果模型本身没通，图节点报错时你很难判断是编排问题还是配置问题。

2. 先把裸循环里的数据写成 State

Raw API 里，所有数据都藏在一个不断增长的 messages 列表和几个局部变量里。LangGraph 的第一步，是把“节点之间需要传递什么”写成明确的状态契约。

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages


class ServiceDeskState(TypedDict, total=False):
    messages: Annotated[list, add_messages]
    route: Literal["ticket", "knowledge"]
    turn_count: int
    last_error: str
    pending_action: dict
    approval: Literal["pending", "approved", "rejected"]

逐行看这段定义：

1. TypedDict 只描述数据形状，不会替你做业务校验。
2. messages 使用 add_messages reducer，节点返回新消息时会追加，而不是把整个列表意外覆盖。
3. turn_count 是循环保险丝，后面由代码递增，不能让模型自己决定。
4. pending_action 保存待确认的副作用动作，例如关闭哪个工单；它不是“已经执行成功”。
5. approval 用有限值表达审批状态，避免出现 "yes"、"ok"、True 混用。

运行观察：这一步还没有真正运行图。先在练习文件里打印 ServiceDeskState.__annotations__，确认字段名和你计划的节点一致。

常见错误：

• 把 API Key、完整网页正文、数据库连接对象放进 State。State 可能被 checkpoint 持久化，敏感信息和大对象都不应放进去。
• 所有节点都能改 messages 以外的字段。给每个字段指定一个主要写入者，后面排错会轻松很多。

想一想：pending_action 为什么要和 approval 分开？如果只存一个 approved=True，进程恢复时你还能准确知道批准的是哪个工单、什么参数吗？

3. 把“调模型”单独变成一个节点

节点就是接收 State、返回部分更新的 Python 函数。先只迁移 Raw API 中的模型调用，不急着接工具。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)


def call_model(state: ServiceDeskState):
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "turn_count": state.get("turn_count", 0) + 1,
    }

这里的对应关系很直接：model.invoke(...) 就是 Raw API 的一次模型请求；返回字典就是“本轮对 State 的更新”。不要在节点里直接修改 state["messages"].append(...)，让 reducer 统一合并更安全。

运行观察：暂时手动构造一份状态调用节点：

state = {"messages": [("user", "公司报销发票丢了怎么办?")], "turn_count": 0}
updated = call_model(state)
print(updated["messages"][-1].content)
print(updated["turn_count"])

你应该看到一条模型回复和 turn_count == 1。如果这里就失败，先修复模型适配器，不要继续加节点。

常见错误：

• messages 里混入 Chat Completions 的普通字典，但当前适配器要求 LangChain 消息对象。统一使用一种消息表示，必要时在入口处转换。
• 节点没有返回字典，或者返回了完整 State。返回“本次改变的字段”更容易看懂，也避免覆盖别的节点结果。

想一想：如果模型节点每次都把 turn_count 重置为 1，图还能可靠地限制循环吗？计数器应该由谁负责增加？

4. 再写一个工具节点

先使用只读的工单查询，避免一开始就把审批和副作用混进来。工具函数仍然是你自己写的业务代码，LangGraph 不会替你访问数据库。

def search_ticket(ticket_id: str) -> dict:
    """根据工单编号查询本地工单记录。"""
    fake_tickets = {
        "INC-1001": {"status": "处理中", "owner": "张工", "summary": "VPN 无法连接"},
    }
    return fake_tickets.get(ticket_id, {"error": "工单不存在"})

为了让模型知道如何调用它，给模型绑定工具。绑定只是在请求中声明 schema，真正执行仍由你的节点完成。

tools = [search_ticket]
model_with_tools = model.bind_tools(tools)


def call_model_with_tools(state: ServiceDeskState):
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "turn_count": state.get("turn_count", 0) + 1,
    }

工具节点读取上一条 AI 消息的 tool_calls，逐个执行，再把 ToolMessage 放回消息流。

from langchain_core.messages import ToolMessage


def run_tools(state: ServiceDeskState):
    last_message = state["messages"][-1]
    results = []
    for call in last_message.tool_calls:
        try:
            result = search_ticket(**call["args"])
        except (TypeError, KeyError) as exc:
            result = {"error": "参数错误", "detail": str(exc)}
        results.append(
            ToolMessage(content=str(result), tool_call_id=call["id"])
        )
    return {"messages": results}

逐行看重点：

• tool_calls 可能一次有多个调用，所以不能只取第一个。
• call["id"] 必须原样放入 tool_call_id，模型才能把结果对应回请求。
• 业务异常要转成结构化结果回传；参数校验和权限校验仍然是你的责任。

运行观察：给 call_model_with_tools 一个明确的输入，例如“查询 INC-1001 当前状态”，打印最后一条消息的 tool_calls。接着手动调用 run_tools，确认得到 ToolMessage，并且 ID 能对应。

常见错误：

• last_message 没有 tool_calls 就进入工具节点。条件边必须保证只有存在调用时才路由到这里，节点内部也应做防御式判断。
• 直接把异常对象、数据库连接或超长结果塞进消息。只回传模型下一步需要的字段和可读错误。

想一想：如果查询工单的结果有 10 万字，模型真的需要全部内容吗？先在工具节点裁剪成状态、负责人、摘要和更新时间，通常更稳。

5. 用固定边把节点串起来

现在把两个节点接成最小图。固定边表达“工具执行完一定回到模型”，这就是 Raw API for 循环中最稳定的一段。

from langgraph.graph import StateGraph, START, END

builder = StateGraph(ServiceDeskState)
builder.add_node("model", call_model_with_tools)
builder.add_node("tools", run_tools)
builder.add_edge(START, "model")
builder.add_edge("tools", "model")

add_node 把函数登记为节点；第一条边规定入口；第二条边规定工具完成后回到模型。此时还没有“结束”这条边，所以先别急着编译。

此时图还缺一个出口：模型没有工具调用时应该结束，不能永远在 model -> tools -> model 之间打转。

运行观察：先不要编译，画一下你现在的路径：START -> model -> ?，你会发现问号就是下一步要写的条件路由。

常见错误：节点名称是字符串，函数名称是 Python 对象；add_edge("tools", "model") 写反会导致工具执行后重新走工具，或者图无法编译。

想一想：固定边适合“工具执行后回模型”，那“模型有时要调用工具、有时直接回答”应该用固定边还是条件边？

6. 用条件边替代裸循环里的 if

把 Raw API 里的 if msg.tool_calls 单独提成路由函数。路由只负责选择下一节点，不执行副作用。

def route_after_model(state: ServiceDeskState):
    last_message = state["messages"][-1]
    if getattr(last_message, "tool_calls", None):
        return "tools"
    return END


builder.add_conditional_edges(
    "model",
    route_after_model,
    {"tools": "tools", END: END},
)

这条条件边要替换第 5 步中“等待补出口”的部分；后面加入循环上限时，也是在这里替换路由函数，而不是再注册一条同名条件边。

逐行看：

1. getattr 让路由同时兼容普通文本消息和 AI 消息。
2. 返回值是逻辑标签，不是节点函数；映射表负责把标签映射到真实节点。
3. END 是图的终止标记，不要写成普通字符串节点。

运行观察：编译并运行第一版图：

graph = builder.compile()
result = graph.invoke({"messages": [("user", "查询 INC-1001")]})
print(result["messages"][-1].content)

分别测试“查询工单”和“今天天气不错吗”两类输入。前者应经过 model -> tools -> model，后者应直接从 model 到 END。

常见错误：

• 路由函数返回了 "end"，却没有在映射表里声明，编译或运行时会报未知分支。
• 用模型文字“我想查工单”判断是否调用工具。应该看结构化的 tool_calls，不要解析自然语言。

想一想：路由函数里能不能顺便调用工单服务？最好不要。把“决定去哪”和“真正执行”分开，测试时才能单独验证路由。

7. 给循环加硬上限和失败出口

Raw API 里用 for range(max_turns) 限制循环；图里没有天然的 for，所以把计数放进 State，再加一个明确的超限路由。

MAX_TURNS = 6


def route_with_limit(state: ServiceDeskState):
    if state.get("turn_count", 0) >= MAX_TURNS:
        return "failed"
    last_message = state["messages"][-1]
    return "tools" if getattr(last_message, "tool_calls", None) else END


def fail_after_limit(state: ServiceDeskState):
    return {
        "last_error": "达到最大执行轮数，已停止继续调用模型或工具。",
        "messages": [("assistant", "这个请求处理步骤太多，我先停下来，请缩小问题范围后再试。")],
    }


builder.add_node("failed", fail_after_limit)
builder.add_conditional_edges(
    "model",
    route_with_limit,
    {"tools": "tools", "failed": "failed", END: END},
)
builder.add_edge("failed", END)

这段代码要替换第 6 步的 route_after_model 条件边。新增的 failed 是真正的终止节点，builder.add_edge("failed", END) 让失败消息写入后结束运行。

这里有两个容易混淆的点：计数是在模型节点返回时增加的；失败节点只负责给出可理解的结果，真正的告警、重试和人工介入仍由业务系统决定。

运行观察：临时把 MAX_TURNS 改成 1，输入一个必然触发工具的请求。你应该看到失败消息，而不是程序无限运行。

常见错误：

• 把上限判断只写在工具节点。模型连续返回无效调用时，可能根本到不了工具节点。
• 失败节点又连回 model，结果只是把“失败出口”变成了另一条死循环。

想一想：除了超限，还有哪些情况应该走失败出口？例如工具超时、权限拒绝、知识库没有证据。先列出错误类型，再决定哪些可重试、哪些应直接结束。

8. 增加“资料为空”和“工具失败”的分支

条件边不只判断有没有工具调用，也可以根据工具结果中的业务状态决定下一步。先让工具统一返回一个 ok 字段，路由就不会依赖错误字符串。

def normalize_ticket_result(result: dict) -> dict:
    if result.get("error"):
        return {"ok": False, "error": result["error"]}
    return {"ok": True, "ticket": result}

工具节点里调用 normalize_ticket_result，并把结果序列化后放入 ToolMessage。把第 4 步中的 result = search_ticket(...) 替换成下面两行：

payload = normalize_ticket_result(result)
results.append(ToolMessage(
    content=json.dumps(payload, ensure_ascii=False),
    tool_call_id=call["id"],
))

记得在文件顶部增加 import json。这样路由可以解析结构化数据，而不是靠查找某段文字。随后增加一个错误出口：

def route_after_tools(state: ServiceDeskState):
    last_tool_message = state["messages"][-1]
    payload = json.loads(last_tool_message.content)
    if payload.get("ok") is False:
        return "tool_failed"
    return "model"


def explain_tool_failure(state: ServiceDeskState):
    return {
        "last_error": "工单服务返回失败",
        "messages": [("assistant", "我暂时查不到这张工单。请确认工单号，或稍后再试。")],
    }


builder.add_node("tool_failed", explain_tool_failure)
builder.add_conditional_edges(
    "tools",
    route_after_tools,
    {"model": "model", "tool_failed": "tool_failed"},
)
builder.add_edge("tool_failed", END)

route_after_tools 返回 "model" 时沿成功边回到模型，返回 "tool_failed" 时进入说明节点；因此要先删除第 5 步的 builder.add_edge("tools", "model")，否则成功时会同时触发固定边和条件边。

逐行看：json.loads 把工具消息还原成字典，payload.get("ok") is False 只识别明确的失败值，其他异常值不会被误判成成功。路由函数仍然不负责重试，重试次数应由独立字段和明确的边控制。

这里的关键是让工具结果有稳定 schema；如果工具很多，可以把结果类型和错误码再细分。生产代码仍要捕获 JSON 解析异常，避免一条损坏的工具消息让整个图崩掉。

运行观察：输入一个不存在的工单号，确认流程走到 tool_failed，不会让模型编造工单信息。

常见错误：把“查不到资料”当成普通成功结果继续生成答案。没有证据时应该明确告诉用户没有查到，并保留内部错误码。

想一想：哪些错误适合让模型改参数再试一次？哪些错误（越权、服务明确拒绝）必须直接结束？把重试次数设成代码常量，而不是让模型决定。

9. 在副作用工具前暂停，加入 checkpoint 和人工确认

查询是只读操作；关闭工单会改变系统状态，不能因为模型发出了 close_ticket 就直接执行。这里用一个很短的“准备动作”节点，把动作写入 State，然后在执行前中断。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt


def prepare_close_ticket(state: ServiceDeskState):
    action = {"name": "close_ticket", "ticket_id": "INC-1001", "reason": "用户确认问题已解决"}
    return {"pending_action": action, "approval": "pending"}


def approve_close_ticket(state: ServiceDeskState):
    decision = interrupt({
        "question": "是否关闭工单?",
        "action": state["pending_action"],
    })
    if decision != "approved":
        return {
            "approval": "rejected",
            "messages": [("assistant", "已取消关闭工单。")],
        }
    return {"approval": "approved"}

interrupt(...) 会暂停当前运行；它不是让模型“自己批准”，而是把待确认信息交给外部 UI、人工审核系统或命令行。编译图时挂上 checkpoint：

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "ticket-INC-1001"}}

# input_state 是前面步骤准备好的用户输入，例如 {"messages": [("user", "关闭 INC-1001")]}
first = graph.invoke(input_state, config)
# 人工确认后恢复同一条 thread
second = graph.invoke(Command(resume="approved"), config)

逐行看：

• MemorySaver 适合本地练习；生产环境应换成持久化 checkpoint 实现。
• thread_id 是找回同一次运行的关键，不能每次恢复都随机生成。
• 恢复时传 Command(resume=...)，不会重新从 START 执行。
• 真正执行 close_ticket 的节点应放在审批节点之后，并再次做权限、幂等和参数校验。

运行观察：第一次 invoke 应停在 interrupt，程序不会关闭工单。用相同 thread_id 恢复并批准后，才进入后续执行节点；恢复并传 rejected 则走取消出口。

常见错误：

• 没有配置 checkpointer 就使用 interrupt，暂停后无法可靠恢复。
• 恢复时换了 thread_id，结果像启动了一个新任务。
• 把“批准”写进 prompt 让模型决定。审批必须来自外部可信身份，模型只能提出建议。
• 恢复接口没有幂等键，用户重复点击批准可能造成重复关闭。副作用工具要在业务层做幂等保护。

想一想：如果审批人拒绝后允许修改原因再提交，你会把图接回 prepare_close_ticket，还是新建一个“补充信息”节点？先画出两条路径，再决定 State 里要保存哪些字段。

10. 把图编译成可观察的练习对象

到这里，你已经把 Raw API 的循环拆成了几类明确部件：

Raw API 里的代码	LangGraph 里的部件	仍由你负责的事情
messages 列表	State 的 messages 字段和 reducer	上下文裁剪、敏感信息处理、消息格式转换
for max_turns	turn_count + 超限条件边	上限数值、成本预算、超限后的用户体验
if tool_calls	条件路由函数	路由规则、默认分支、错误分类
run_tool()	工具节点	参数校验、权限、超时、幂等和真实服务调用
手工暂停	checkpoint + interrupt	审批人身份、审计、持久化和恢复接口

可以用 stream 观察每个节点的更新，帮助自己理解“图到底走了哪条路”：

for update in graph.stream(input_state, config, stream_mode="updates"):
    print(update)

stream_mode="updates" 表示按节点更新返回事件；如果要看 token 级文本流，应选择适合的流模式并单独处理消息事件。先把节点名和状态更新打印出来，再考虑接到 UI。

运行观察：一次查询请求至少会看到 model，有工具时还会看到 tools；超限或工具失败时，应出现对应失败节点。记录节点名、耗时、工具名和错误码即可，不要把密钥和完整个人数据原样写入日志。

常见错误：把 stream 的每个事件都当成最终文本直接展示。节点更新、工具事件和文本 token 是不同层次的数据，先区分事件类型，再决定展示给用户的内容。

想一想：如果线上只记录最后答案，出了“模型重复调用工具”的问题，你还能还原路径吗？至少保留 run_id、thread_id、节点名、调用次数、耗时和脱敏后的工具参数。

11. 再加一个业务路由，而不是继续堆 if

服务台助手通常有两条入口：制度问答走知识库，工单请求走工具。可以在模型前增加一个轻量路由节点，但路由规则最好是结构化的，别让模型自由拼接节点名。

def classify_request(state: ServiceDeskState):
    text = state["messages"][0].content
    if "工单" in text or text.startswith("INC-"):
        return {"route": "ticket"}
    return {"route": "knowledge"}


def route_request(state: ServiceDeskState):
    return state.get("route", "knowledge")


builder.add_node("classify", classify_request)
# 如果把本节加入前面的 builder，请先删除第 5 步的 builder.add_edge(START, "model")。
# START 只能保留一个入口，否则同一次运行会同时走两条路径。
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_request,
    {"ticket": "model", "knowledge": "model"},
)

classify_request 只写入 route，route_request 只读取并返回有限标签，映射表再把标签指向节点。把入口改成分类节点后，前面的 START -> model 必须删除，避免一次运行出现两个入口。

目前 ticket 和 knowledge 都指向 model。你可以让它们分别进入不同节点，再汇合到统一回答节点。如果分类错误，应该有默认安全路径和可观测指标，而不是静默丢请求。

运行观察：准备“如何申请 VPN 权限”和“查询 INC-1001”两条输入，记录它们各自走过的节点。比较图结构和 Raw API 中嵌套 if 的可读性。

常见错误：路由键写成了多个近义词（ticket、tickets、workorder），但映射表只声明一个。路由值应使用有限集合，并在入口处校验。

想一想：当规则已经能用关键词或正则判断时，为什么还要额外调用一次模型做分类？把每个模型调用都换算成延迟、成本和失败概率。

12. 你应该带走的判断标准

LangGraph 帮你省下的是编排样板：State 合并、节点连接、条件边、暂停恢复和运行轨迹都有统一的表达方式。它没有替你完成下面这些设计：

• 工具是否真的需要，参数和权限边界是什么。
• State 哪些字段该持久化，哪些内容必须脱敏或裁剪。
• 什么时候重试，什么时候向用户返回失败。
• 如何防止循环、重复副作用和成本失控。
• 人工审批由谁做、凭什么身份做、如何审计。

如果你的流程只有“模型调用一次，最多调一个工具”，Raw API 往往更清楚；当流程开始出现分支、回退、暂停、恢复和多个状态写入者时，图的价值才会超过它的学习成本。判断依据是控制流复杂度，不是“用了框架就更专业”。

课后练习

1. 把 search_ticket 换成一个知识库检索工具，规定“没有证据时必须走 tool_failed”。
2. 给工具节点增加超时和一次重试，超过次数后进入失败出口；确认副作用工具不会自动重试。
3. 为 pending_action 增加操作者、理由和过期时间，模拟审批超时后的取消路径。
4. 用同一组输入分别运行 Raw API 版和 LangGraph 版，比较最终答案、节点路径、工具调用次数、总耗时和失败原因。

每做完一个练习，都先写下“我新增了哪个 State 字段、哪个节点、哪条边，以及为什么”，再运行代码。这样你是在设计工作流，而不是把循环换成另一套 API 名字。