用 LangGraph 设计多步骤智能体：节点、状态与流程调试方法

很多智能体项目并不是“调用一次模型、返回一段文字”这么简单。比如整理一条优惠信息，可能要先提取商品和时间，再判断信息是否完整，缺少关键字段时补充处理，最后生成适合发布的结果。把这些动作全部塞进一个大函数，开始时看似省事，后续却很难定位问题。LangGraph 的价值，就在于把任务拆成节点，用状态连接节点，再用条件边决定下一步走向。

先把任务画成一张流程图

这篇文章用“整理一条优惠信息”作为例子。输入可能是一段不完整的文字，程序需要完成四件事：

1. 提取商品名称和优惠内容。




2. 检查关键字段是否齐全。




3. 如果缺少信息，进入补充节点。




4. 信息完整后，生成一段适合发布的摘要。

对应的流程可以先画成这样：

输入
  ↓
提取信息
  ↓
检查完整性
  ├── 完整 ──→ 生成摘要 ──→ 结束
  └── 缺失 ──→ 补充信息 ──→ 生成摘要 ──→ 结束

这里有三个重要判断。

第一，节点应该承担单一职责。提取节点只负责整理字段，不负责写最终文案；检查节点只负责判断是否缺失，不直接修改所有内容。职责越清楚，后面越容易替换或测试。

第二，状态是节点之间传递的共同数据。它不只是最终答案，也可以保存原始输入、中间结果、判断标记和调试信息。

第三，分支应该由状态中的明确字段驱动。例如 is_complete 表示信息是否完整，而不是让下一个节点自行猜测上一步发生了什么。

设计状态：只保存流程真正需要的数据

一个实用的状态通常包含三类信息：

• 输入和中间结果，例如原始文本、提取出的商品名称；




• 流程判断，例如信息是否完整、下一步应该走哪条路径；




• 调试记录，例如每个节点执行后的简短说明。

状态不宜无限膨胀。把所有模型原始响应、重复文本和无关配置都塞进状态，会让流程变得难读，也增加调试成本。先定义一个足够小的状态，再根据实际问题补充字段，通常更稳妥。

下面的示例使用 Python 类型标注定义状态：

from typing import TypedDict


class DealState(TypedDict, total=False):
    raw_text: str
    product: str
    discount: str
    is_complete: bool
    route: str
    summary: str
    debug: list[str]

total=False 表示部分字段可以在流程开始时暂时不存在。这样，输入阶段只需要提供 raw_text，后面的节点再逐步写入其他字段。

从节点开始搭建最小流程

为了先理解图的组织方式，下面的例子不调用模型，而是使用简单规则模拟提取和判断。这样可以把注意力放在节点、状态和分支上。之后如果需要接入大模型，只需替换某个节点内部的处理逻辑，整体图结构不必重写。

1. 编写提取节点

def extract_deal(state: DealState) -> dict:
    text = state["raw_text"]

    product = "未识别商品"
    discount = ""

    if "咖啡" in text:
        product = "咖啡"
    if "买一送一" in text:
        discount = "买一送一"
    elif "打折" in text:
        discount = "打折"

    debug = list(state.get("debug", []))
    debug.append("extract_deal：完成商品和优惠信息提取")

    return {
        "product": product,
        "discount": discount,
        "debug": debug,
    }

节点接收完整状态，返回需要更新的字段。不要在节点内部偷偷修改共享对象后再依赖副作用，这会让执行顺序和问题定位变得模糊。显式返回更新内容，更容易观察状态变化。

2. 编写检查节点

def check_deal(state: DealState) -> dict:
    product = state.get("product", "")
    discount = state.get("discount", "")

    is_complete = (
        product not in ("", "未识别商品")
        and discount != ""
    )

    debug = list(state.get("debug", []))
    debug.append(
        f"check_deal：信息{'完整' if is_complete else '不完整'}"
    )

    return {
        "is_complete": is_complete,
        "debug": debug,
    }

这里的检查规则很简单，但它体现了一个关键做法：让分支依据显式存在于状态中。实际项目中，检查内容可以换成字段校验、模型输出格式校验或人工确认标记。

3. 编写补充和生成节点

def enrich_deal(state: DealState) -> dict:
    debug = list(state.get("debug", []))
    debug.append("enrich_deal：补充缺失的优惠说明")

    return {
        "discount": state.get("discount") or "优惠信息待确认",
        "debug": debug,
    }


def write_summary(state: DealState) -> dict:
    product = state.get("product", "商品")
    discount = state.get("discount", "优惠信息待确认")

    summary = f"{product}：{discount}。具体活动条件请以实际页面或门店说明为准。"

    debug = list(state.get("debug", []))
    debug.append("write_summary：生成最终摘要")

    return {
        "summary": summary,
        "debug": debug,
    }

补充节点不一定要“猜测”缺失内容。对于真实优惠信息，无法确认的字段应该明确标记为待确认，而不是为了让文案完整而编造条件。

添加条件边，形成可执行的图

节点写好之后，再把节点连接起来：

from langgraph.graph import StateGraph, START, END


def choose_route(state: DealState) -> str:
    if state.get("is_complete", False):
        return "complete"
    return "missing"


builder = StateGraph(DealState)

builder.add_node("extract_deal", extract_deal)
builder.add_node("check_deal", check_deal)
builder.add_node("enrich_deal", enrich_deal)
builder.add_node("write_summary", write_summary)

builder.add_edge(START, "extract_deal")
builder.add_edge("extract_deal", "check_deal")

builder.add_conditional_edges(
    "check_deal",
    choose_route,
    {
        "complete": "write_summary",
        "missing": "enrich_deal",
    },
)

builder.add_edge("enrich_deal", "write_summary")
builder.add_edge("write_summary", END)

graph = builder.compile()

这段代码表达的是：

• START 指向第一个业务节点；




• extract_deal 完成后固定进入 check_deal；




• check_deal 通过 choose_route 返回路由名称；




• 路由名称再映射到不同节点；




• 两条路径最终都汇合到 write_summary；




• 最终节点连接到 END。

条件边的函数最好只负责“判断去哪儿”，不要在里面顺便修改状态。判断和数据处理分开后，流程图更容易理解，也更方便单独测试。

运行最小示例

可以准备两组输入，分别观察完整路径和缺失路径：

complete_result = graph.invoke({
    "raw_text": "咖啡今天买一送一",
    "debug": [],
})

missing_result = graph.invoke({
    "raw_text": "今天有个活动",
    "debug": [],
})

print(complete_result["summary"])
print(complete_result["debug"])

print(missing_result["summary"])
print(missing_result["debug"])

第一组输入会经过：

extract_deal → check_deal → write_summary

第二组输入会经过：

extract_deal → check_deal → enrich_deal → write_summary

这里的关键不是输出句子有多复杂，而是两次执行共享同一张图，却能根据状态进入不同路径。以后将 extract_deal 换成模型抽取节点，或将 write_summary 换成模型写作节点，流程控制仍然可以保持不变。

调试时，不要只看最终结果

智能体最难排查的问题，往往不是最终输出为空，而是中途某个节点做了错误判断。例如：

• 提取节点没有写入预期字段；




• 检查节点把缺失信息误判为完整；




• 路由函数返回了没有配置的名称；




• 补充节点覆盖了原本正确的字段；




• 生成节点读取了错误的状态键。

因此，调试记录应当伴随状态流转，而不是只在程序最后打印一句“执行失败”。

前面的 debug 字段就是一种简单做法。每个节点完成自己的工作后，追加一条短记录：

[
    "extract_deal：完成商品和优惠信息提取",
    "check_deal：信息不完整",
    "enrich_deal：补充缺失的优惠说明",
    "write_summary：生成最终摘要",
]

这能帮助你先确认“走了哪些节点”，再进一步检查“每个节点写入了什么”。如果问题更复杂，可以在调试环境中同时记录节点名称、关键字段和路由结果，但不要把完整敏感输入无差别写入日志。

用小测试验证每个节点

不要一开始就只测试整张图。节点可以分别测试，先确认局部行为，再测试分支是否正确。

例如，提取节点至少应该验证两类输入：

def test_extract_deal():
    result = extract_deal({
        "raw_text": "咖啡今天买一送一",
        "debug": [],
    })

    assert result["product"] == "咖啡"
    assert result["discount"] == "买一送一"

路由函数也可以单独验证：

def test_choose_route():
    assert choose_route({"is_complete": True}) == "complete"
    assert choose_route({"is_complete": False}) == "missing"

最后再验证整张图是否生成了预期字段：

def test_graph():
    result = graph.invoke({
        "raw_text": "咖啡今天买一送一",
        "debug": [],
    })

    assert result["summary"]
    assert result["debug"]

这种测试顺序比直接盯着最终文案更有效：先验证字段提取，再验证条件判断，最后验证节点连接。如果最终结果异常，可以较快缩小问题范围。

把这个模式迁移到真实智能体

当流程需要接入大模型时，可以保留相同的组织方式：

• 把“提取信息”节点替换为结构化信息抽取；




• 把“检查完整性”节点保留为确定性的校验逻辑；




• 把“补充信息”节点改成模型生成澄清问题，或调用已有资料；




• 把“生成摘要”节点替换为模型写作；




• 在状态中保存模型返回的结构化结果和必要的调试信息。

不要让每个节点都直接读取和修改全部状态。节点只读取完成任务所需的字段，并返回自己的更新结果。这样做既能减少节点之间的隐式耦合，也方便以后增加循环，例如“生成草稿—检查—不合格则返回修改”。

对于初学者，最稳妥的练习顺序是：先用固定规则跑通一条直线路径，再加入一个条件分支，接着记录状态变化，最后才把其中一个节点替换成模型调用。这样遇到问题时，你能区分到底是图结构、状态设计，还是模型输出导致的错误。