LangGraph实战：用图状态机编排多智能体工作流

1. 为什么需要一张"图"来编排多智能体：失控场景与 StateGraph 的解法

1.1 智能体协作最常掉的三个坑

如果只用一段话概括我最早的踩坑经历，那就是：当工作流里超过两个智能体角色时，"先A再B，然后把结果拼起来"这种流水线式的写法，会迅速变成一坨需要反复调试的意大利面。我最初搭过一个"需求拆解→写作→检查→重写"的流程，代码写得很痛快，一个函数接一个函数，每个函数内部再调模型，像搭积木一样。结果上线后才发现三个问题。

第一个问题是分支逻辑无处安放。检查节点不满意，我要重新触发写作节点，这时候函数调用链已经走完了，我只能在外层强行写一个 while 循环包住整段流水线。一旦循环里再套循环，代码的可读性直接归零。第二个问题是状态丢失。A节点改了某个字段，B节点要用，C节点又改了，整段逻辑里根本没有一个清晰的地方能看到"当前到底进行到哪一步、手上握着哪些数据"。第三个问题是重试和持久化极其难做。任何一次中间调用出错，我只能从头开始跑，刚才各节点生成过的中间结果全部作废。

后来我接触了 LangGraph 这个框架，它把多智能体工作流抽象成一张有向图。图上的节点是你要执行的逻辑，可以是调用大模型、工具函数或者纯Python计算；边才是控制流，决定了节点之间谁先谁后、何时跳过、何时回头。这种思维切换之后，上面三个问题都有了比较标准的解法：分支用条件边表达，循环用图里的回边表达，状态由一个统一的State对象负责传递。这篇文章我就从最基础的东西讲起，带你把第一个能用、能扩展的多智能体工作流跑起来。

1.2 图、状态与边的本质：更像状态机而不是流水线

很多人第一次看到"图结构"这个词会觉得是高深理论。其实你可以把它理解成一张路线图：你从某个入口进入（对应 START 节点），沿途经过一系列站点（对应普通节点），最终走到终点（对应 END 节点）。关键区别在于，路线不是只能一条道走到黑，你可以根据当前状态选择走左边还是走右边，甚至可以绕回刚才的站点再来一遍。

LangGraph 的核心构成是三个东西：节点（Node）、边（Edge）和状态（State）。节点是最小执行单元，每个节点都是一个函数，输入当前的整个State，输出一个字典，字典里的内容会被合并回State。边定义了执行顺序，有普通边（无条件直接跳转）和条件边（根据当前State的内容决定跳往哪个节点）。State则是整张图的"公共黑板"，所有节点都能读、都能写，工作流的上下文不再散落在各个局部变量里。

我看到过一种很贴切的类比：LangGraph 本质上不是一个现代化Pipeline工具，而是一个状态机框架。Pipeline的思路是"数据流经过管道"，每一步只关心上一管道的输出；状态机的思路是"系统处于某个状态，根据事件转移到另一个状态"。多智能体协作天然需要状态机，因为智能体之间经常需要来回协商、反复修正，而不是一锤子买卖。用图结构表达这种控制流，比用嵌套条件语句表达要直观得多。

我把几种常见编排方式的差异放在下面的表里，方便你建立直觉：

我个人认为，只要你遇到至少两个智能体角色之间需要"多轮往返"，就值得用图状态机来替代手写的调用链。这不是过度设计，恰恰是在控制复杂度。

2. 搭第一个两节点流水线：从安装到编译调用

2.1 环境准备

LangGraph 是一个 Python 库，官方文档要求 Python 3.9 及以上版本。安装只需要一条命令：

bash复制pip install langgraph

如果你只是在标准库之外额外装它，依赖会比较克制，不会拖上一大堆重量级库。为了后面调试方便，我建议同时装一个 langgraph-cli 或者直接在代码里用 print 输出状态，二选一即可。

提示：如果你在某个内置了 Python 的沙盒环境里运行，先确认 python --version 能达到 3.9。某些旧版本发行版的默认 Python 还是 3.8，这会让你在启动编译时遇到奇怪的类型报错。

装完之后可以快速验证版本：

python复制import langgraph
print(langgraph.__version__)

我实测下来，0.2 之后的版本 API 变化比较大，尤其是 START、END 这类常量的引入方式。为了你之后对照阅读官方文档方便，这篇文章里我会使用较新的写法，也就是：

python复制from langgraph.graph import StateGraph, START, END

2.2 State 与 Node 的最小写法

现在我们来搭第一个最小工作流：一个生成标题的节点，再接一个校验标题长度的节点。这两个节点之间没有分支，纯粹是顺序执行，但足以让你理解"状态如何流入和流出"。

先定义状态。状态必须是一个类型字典，平时最顺手的方式是用 TypedDict 来定义：

python复制from typing import TypedDict

class DemoState(TypedDict):
    topic: str          # 输入主题
    title: str          # 生成结果
    is_valid: bool      # 校验结果

这个状态字典里应该有三个字段，代表工作流从开始到结束需要用到的所有数据。注意，节点写返回字典时只能返回这些字段的子集，框架会自动合并，不会抹掉没返回的其他字段。

接着定义两个节点函数。节点函数的签名是固定的：接收一个 state 参数，返回一个字典。

python复制def generate_title_node(state: DemoState) -> dict:
    # 假装调了一次大模型，这里返回一个固定标题
    title = f"关于{state['topic']}的一篇深度解读"
    return {"title": title}

def validate_title_node(state: DemoState) -> dict:
    title = state["title"]
    is_valid = len(title) > 5
    return {"is_valid": is_valid}

这两个函数都不访问局域网以外的服务，纯粹做演示。你会注意到，第二个节点并没有手动接收第一个节点的返回值，而是直接从 state 里读取 title 字段，这就是"公共黑板"的含义。

最后是构建和编译：

python复制builder = StateGraph(DemoState)
builder.add_node("generate_title", generate_title_node)
builder.add_node("validate_title", validate_title_node)

builder.add_edge(START, "generate_title")
builder.add_edge("generate_title", "validate_title")
builder.add_edge("validate_title", END)

graph = builder.compile()

2.3 编译、调用与查看内部连接

compile() 会把我们声明好的节点和边"冻结"成一张可执行的图。调用它也非常直接，用 invoke 传入初始状态：

python复制result = graph.invoke({"topic": "LangGraph入门"})
print(result)

输出结果会是一个字典，包含工作流跑完之后的所有状态字段：

python复制{
  "topic": "LangGraph入门",
  "title": "关于LangGraph入门的一篇深度解读",
  "is_valid": True
}

这里有个容易忽略但很关键的点：invoke 返回的是整张图执行完的最终状态，而不是某个节点的返回值。这也是图状态机和普通函数调用的最大区别——你拿回的不是某一环的输出，而是全局快照。

你可以用一行代码查看整张图的拓扑结构：

python复制graph.get_graph().print_ascii()

执行后终端会打印：

code复制       +-----------+      
       | __start__ |      
       +-----------+      
          *        
          *        
          *        
  +-----------------+     
  | generate_title  |     
  +-----------------+     
          *        
          *        
          *        
  +-----------------+     
  | validate_title  |     
  +-----------------+     
          *        
          *        
          *        
       +-------+            
       |  end  |            
       +-------+            

虽然这个例子里只有两个节点，但它已经把最重要的心智模型立起来了：所有节点共享一份状态，所有节点都有明确的执行路径。后面我们再往里面加分支、加循环、加更多智能体，逻辑骨架都是这一套。

3. 状态对象才是真正的"数据库"：多智能体共享上下文的底层逻辑

3.1 自定义 State 与 reducer 规则

一旦你想让多个节点之间真正协作，状态的设计就成了决定成败的关键。 TypedDict 里的每个字段都只是静态约束，真正有意思的是 Annotated 泛型加 reducer 函数。这个机制决定了一个字段在被多个节点写入时，是"覆盖"还是"合并"再写回。

我先解释一下默认行为。如果你在状态里写：

python复制class WorkflowState(TypedDict):
    draft: str

那么任何节点返回包含 draft 的字典，都会直接覆盖原来的值。这在"一个人写完另一个人继续改"的场景下没有问题。但在多智能体场景里，某条消息可能由多个节点分别生成，直接覆盖会导致信息丢失。此时就需要给字段配置 reducer。

reducer 的规则其实很简单：新值和旧值同时进入一个函数，函数返回什么东西，最终状态里就存什么东西。LangGraph 预置了好几个实用 reducer，其中一个最常用的是 add_messages，用于把多个节点的输出累积成一个消息列表。

看下面这个例子：

python复制from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
    final_summary: str

这里的 messages 字段会走 add_messages 这个 reducer：如果有节点返回 {"messages": [某条消息]}，框架不会覆盖整个列表，而是把新消息追加到已有列表后面。这个机制正是多智能体共享"聊天记录"的基础。

3.2 为什么不能简单用函数返回值拼内存

有人可能会问：我直接用全局变量或者在节点之间靠函数返回值传递上下文不是更简单吗？为什么非要状态对象？

回答这个问题需要先意识到：多智能体工作流不是一次函数调用，它是一段可能包含分支、循环、异常回退的执行过程。普通函数返回值只能向后传，不能向前传；循环场景里你需要的是"上一次迭代的产物能作为下一次迭代的输入"，这靠参数拼接会很别扭，靠状态对象就自然得多。

更重要的是，状态对象让"可观测"成为可能。你可以随时打印 state，看到当前所有智能体都知道什么信息；也可以随时把一个中间状态插回去，单独测试某个节点。全局变量能读能写，但无法回答"这个值是在哪一步被谁改的"。状态对象带着明确的 Schema，每一个字段的写入路径都清楚，排查问题的时候优势极其明显。

另外，LangGraph 的状态对象不只是内存里的一个字典。配合检查点机制（后面会详细讲），它还能被序列化、持久化，支持断点续跑。这是普通局部变量完全做不到的。简单说：状态对象既是工作流的内存，也是工作流的"存档系统"。

3.3 消息列表的增量更新：add_messages

我们用一个小例子看看 add_messages 实际怎么工作。假设你有两个节点，一个负责生成分析观点，一个负责总结观点。

python复制from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, HumanMessage

class AnalysisState(TypedDict):
    messages: Annotated[list, add_messages]

def analyst_node(state: AnalysisState) -> dict:
    # 假装这个节点是"分析智能体"
    return {"messages": [AIMessage(content="我的分析结果是：成本偏高")]}

def summarizer_node(state: AnalysisState) -> dict:
    # 集市智能体可以读取全部历史消息
    all_text = "\n".join(m.content for m in state["messages"])
    return {"messages": [AIMessage(content=f"总结后的一句话：{all_text}")]}

构建这张图的时候，messages 字段虽然被两个节点都写了，但不会相互覆盖。analyst_node 先追加一条 AIMessage，summarizer_node 读取到的 state["messages"] 里已经包含了这条记录，再追加自己的输出。最终状态里会有两条消息，顺序也是明确的。

add_messages 背后对同一消息 ID 的去重逻辑也值得一提：如果你返回的消息里带了一个已经存在的 ID，它会把旧消息替换掉而不是追加。这个特性在重试场景里非常有用——重试节点可以基于相同消息 ID 更新结果，而不会在历史记录里堆积冗余内容。

到这里你应该已经感受到，状态字段的"合并策略"才是 LangGraph 表达业务规则的真正抓手。你定义状态时，其实就是定义多智能体之间怎么交换信息；这个环节想得越细，后面写节点和边就越顺手。

4. 条件边与循环：让工作流学会"推荐"和"重跑"

4.1 路由函数怎么设计

顺序执行只能解决最基础的工作流。真实的多智能体场景里，节点经常要根据当前状态决定下一步去哪。这时候就要用条件边。

条件边的写法分两步：先写一个路由函数，返回一个字符串（或字符串列表），表示下一步要去哪个节点；再用 add_conditional_edges 把它接到某个节点后面。

举个具体例子：一个节点负责判断用户的反馈是否通过，如果通过就进入发布节点，不通过就进入改写节点。

python复制from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class ReviewState(TypedDict):
    feedback_ok: bool
    article: str

def draft_node(state: ReviewState) -> dict:
    return {"article": "这是一篇初稿"}

def route_review(state: ReviewState) -> str:
    if state["feedback_ok"]:
        return "publish"
    return "rewrite"

def publish_node(state: ReviewState) -> dict:
    return {"article": state["article"] + "（已发布）"}

def rewrite_node(state: ReviewState) -> dict:
    return {"article": state["article"] + "（已改写）"}

关键在下面这段构建代码，注意 add_conditional_edges 的第三个参数是一个映射表，把路由函数的返回值映射到真实节点名：

python复制builder = StateGraph(ReviewState)
builder.add_node("draft", draft_node)
builder.add_node("publish", publish_node)
builder.add_node("rewrite", rewrite_node)

builder.add_edge(START, "draft")
builder.add_conditional_edges(
    "draft",
    route_review,
    {
        "publish": "publish",
        "rewrite": "rewrite",
    },
)
builder.add_edge("publish", END)
builder.add_edge("rewrite", END)

调用时传入不同的 feedback_ok 值，你会看到 article 走完全不同的路径。这就像高速公路的转向匝道，转向原因清清楚楚写在一个路由函数里，不用再去翻主流程代码。

4.2 用循环实现失败重试与人工兜底

条件边加图结构之后，一个很容易被忽略的威力是：你可以让边回到之前的节点，形成循环。这对智能体协作太重要了，因为大模型生成内容天然有"不稳定"的属性，一个节点给出的结果未必合格，必须允许反复修正。

设想一个"生成标题→检查标题是否合格→不合格则回到生成节点再来一次"的流程。这个流程用普通函数写，就是一个 while 循环加 break 条件；用 LangGraph 写，则是给生成节点添加一条指向自己的回边，同时给检查节点加条件边：

python复制class TitleState(TypedDict):
    title: str
    attempts: int
    valid: bool

def generate_title(state: TitleState) -> dict:
    return {"title": f"第{state['attempts'] + 1}次尝试生成的标题"}

def check_title(state: TitleState) -> str:
    valid = len(state["title"]) > 5 and state["attempts"] < 2
    if valid:
        return "accept"
    return "retry"

def mark_attempt(state: TitleState) -> dict:
    return {"attempts": state["attempts"] + 1}

builder = StateGraph(TitleState)
builder.add_node("generate", generate_title)
builder.add_node("check", check_title)
builder.add_node("mark", mark_attempt)

builder.add_edge(START, "generate")
builder.add_edge("generate", "check")
builder.add_conditional_edges(
    "check",
    check_title,
    {"accept": END, "retry": "mark"},
)
builder.add_edge("mark", "generate")  # 回边，形成循环

这里 mark_attempt 先递增尝试次数，然后无条件跳回 generate，于是"检查不过就重新生成"的循环就成型了。再看一眼那条 builder.add_edge("mark", "generate")，它就是一个普通的无条件边，但在图中它引入了一条回路。靠这条回路，图从一个单向前进结构变成了真正有状态的循环结构。

4.3 设置最大迭代次数避免死循环

图上的循环虽然强大，但也带来了一个现实风险：一旦路由函数判断失误，某个节点反复触发回边，工作流就会无限循环。LangGraph 默认有递归限制，超过后会抛出错误，但你最好自己显式控制上限。

配置方式有两种。一种是在 invoke 时传入配置：

python复制graph.invoke(
    {"title": "", "attempts": 0, "valid": False},
    config={"recursion_limit": 10},
)

另一种是在节点内部做好防抖逻辑：比如给状态加一个 attempts 字段，路由函数一看到尝试次数超过N就强制走 accept 路径。我个人倾向于两种都做，因为外部限制是兜底，内部字段是业务语义，两者各司其职。

关于循环还有一个很实用的小技巧：你可以打印路由函数收到的状态，快速判断为什么某条路径被触发。比如下面的代码：

python复制def check_title(state: TitleState) -> str:
    print(f"[DEBUG] attempts={state['attempts']}, title={state['title']}")
    ...

这个调试习惯能帮你省下大量排查时间，尤其当节点数变多之后。

5. 一个完整的多智能体小项目：需求拆解、审查与执行三个节点的协作

5.1 三个节点的职责划分

讲了这么多基础概念，是时候拼一个稍微完整的例子了。我们做一个"需求拆解→方案审查→方案执行"的三节点工作流，模拟三个不同角色在不同阶段的协作。

第一个节点是需求拆解节点，负责把用户的一句需求拆成可执行的任务列表；第二个节点是审查节点，负责检查任务列表是否完整；第三个节点是执行节点，负责把审查通过后的任务列表转成最终输出。三个节点的职责一定要分开，避免一个函数里同时做拆解和审查，否则后面加路由条件时会很痛苦。

定义状态：

python复制from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class ProjectState(TypedDict):
    requirement: str
    tasks: list
    review_passed: bool
    final_plan: str

5.2 状态流转与条件出口

需求拆解节点拿 requirement，产出 tasks。审查节点看 tasks 里有没有足够的关键词和要素，决定 review_passed 是真是假。如果审查通过，执行节点把 tasks 拼成 final_plan；如果没通过，工作流回到需求拆解节点重新生成任务列表。

注意这里我故意不加人人都会加的 attempts 字段，而是假设审查节点内部会“严苛一点”，第一次大概率不通过，从而让你能直观看到回环效果。

节点函数如下：

python复制def decompose_node(state: ProjectState) -> dict:
    requirement = state["requirement"]
    # 模拟第一次拆解时漏掉了一个要素
    if state.get("tasks") is None:
        tasks = [f"理解需求：{requirement}", "实现主要功能"]
    else:
        tasks = [f"理解需求：{requirement}", "实现主要功能", "补充测试与文档"]
    return {"tasks": tasks}

def review_node(state: ProjectState) -> dict:
    tasks = state["tasks"]
    passed = len(tasks) >= 3  # 至少有三个任务才算合格
    return {"review_passed": passed}

def execute_node(state: ProjectState) -> dict:
    plan = "\n".join(f"- {t}" for t in state["tasks"])
    return {"final_plan": plan}

5.3 完整运行代码与输出分析

构建图，以及路由函数：

python复制def route_review(state: ProjectState) -> str:
    return "execute" if state["review_passed"] else "decompose"

builder = StateGraph(ProjectState)
builder.add_node("decompose", decompose_node)
builder.add_node("review", review_node)
builder.add_node("execute", execute_node)

builder.add_edge(START, "decompose")
builder.add_edge("decompose", "review")
builder.add_conditional_edges(
    "review",
    route_review,
    {"execute": "execute", "decompose": "decompose"},
)
builder.add_edge("execute", END)

graph = builder.compile()

调用一个初始状态，注意 tasks 字段先留空：

python复制result = graph.invoke(
    {
        "requirement": "开发一个带登录和日志统计的小系统",
        "tasks": [],
        "review_passed": False,
        "final_plan": "",
    }
)
print(result["tasks"])
print(result["final_plan"])

第一次运行时，decompose_node 只生成两个元素，审查失败，于是回环重新执行一次。第二次 decompose_node 发现 tasks 不是空，补上了"补充测试与文档"这一项，审查通过，最终 execute 节点输出完整方案。

这个例子虽然逻辑简单，但它完整地展示了条件路由、回环和状态共享三个核心机制的协作方式。你可以把它当作以后搭建复杂多智能体系统的基本骨架：不同角色都做成独立节点，通过共享状态和条件边决定彼此的协作顺序。

6. 进阶配置与踩坑实录：检查点、并发限制与调试技巧

6.1 检查点机制：把对话记忆落到硬盘

前面的例子跑完后，状态就留在内存里，进程一结束就消失。真实项目里，多智能体工作流往往需要跨请求保持会话状态，比如用户中途退出，下次回来还要能从上次中断的地方继续。LangGraph 提供检查点（checkpointer）机制来解决这个问题。

用起来很简单，先创建一个检查点存储器，编译时传进去：

python复制from langgraph.checkpoint.memory import InMemorySaver

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

config = {"configurable": {"thread_id": "demo-project-001"}}
result1 = graph.invoke(
    {"requirement": "开发一个带登录和日志统计的小系统", ...},
    config=config,
)

同一个 thread_id 的多次调用会共享同一份状态。也就是说，第一次调用如果中途报错，你不用重新把整张图跑一遍，只要在同样的 thread_id 下再次 invoke，LangGraph 会从检查点恢复。这是做长对话、断点调试的利器。

生产环境里 InMemorySaver 只适合测试，它毕竟只存在内存中。更常见的做法是用持久化存储，比如本地文件系统或数据库实现，不过迁移逻辑并不复杂，核心接口一致，换掉存储器即可。

6.2 并发节点与资源竞争的教训

LangGraph 支持在一个节点之后连接多条边，引入并行路径。听起来很爽，但这里有一个容易犯迷糊的坑：并行节点同时读写同一个状态字段时，最终结果按顺序最后一次写入生效，而节点的调度顺序有不确定性，于是会产生可怕的“有时A赢有时B赢”的随机行为。

我在一个模拟项目里踩过这个坑：两个并行节点同时往 report_result 字段写最终报告，结果每次输出的风格都不一样，原因就是谁后写完谁覆盖了。后来我把两个节点的输出分别放进 report_part_a 和 report_part_b，再用一个汇总节点合并，问题立刻消失。

我的建议是：如果你没有明确需要真并行，宁愿让节点串行执行，状态一致性更好；如果确实需要并行，一定要为每个节点分配独立的字段，再由一个下游节点统一合并。

6.3 两条很实用的调试经验

最后分享两个实打实的调试心得，都是常规文档里不会仔细写的。

第一个是"空跑"调试法。在没有任何外部服务依赖的时候，先用返回固定值的节点把整张图跑通，确认拓扑逻辑无误，再接真实的大模型调用和工具函数。很多新手一上来就把节点函数写得特别复杂，结果一旦报错，根本分不清是图的连线错了还是节点内部业务错了。先空跑，能帮你把"编排问题"和"业务问题"分离开。

第二个是用 ASCII 图辅助找路由问题。graph.get_graph().print_ascii() 会输出一张可读性很好的结构图。你只要看一眼这张图，就能立刻发现"诶，为什么这个节点没有连到 END？""为什么这里有条我意料之外的回边？"。比反复读代码高效得多。

另外多说一句，LangGraph 的报错信息不算特别友好，很多错误只说"node not found"或者"cannot merge"这类笼统提示。遇到这种报错，第一时间去核对节点名拼写和状态字段类型，不要先怀疑框架本身出了什么问题。大多数情况就是节点名写错了，或者某个字段更新时 reducer 和实际数据类型不匹配。

就我个人经验来说，图状态机这个模型对多智能体协作非常自然。最开始你需要刻意转变思维，把"下一步做什么"的决策从函数代码里抽离出来，变成节点和边的显式关系。一旦适应了这种表达方式，再回头去看那些嵌套循环和层层条件判断的旧代码，会有一种"原来可以这样清爽"的感慨。希望这篇文章能帮你把第一个工作流顺利跑起来，也欢迎你按这个骨架去扩充自己的真实业务场景。