在构建 AI 应用时，很多人最早接触的是单链式 Agent，但当任务变复杂，就会遇到LangGraph 这类”图编排”框架。它把多智能体协作拆解成一个有向状态图：每个节点是一个函数或大模型调用，边决定下一步去哪，条件边还能让流程自动循环、分支或暂停等待人工。相比把全部逻辑写进一条 ReAct 链，用图来描述可以让”自我纠错、人工审核、多角色分工”变得清晰且可维护。下面我从概念到代码，带你把 LangGraph 真正用起来。
一、为什么需要图编排
我们之前在《手写 ReAct Agent》里用”思考-行动-观察”循环解决过工具调用问题。那种单链结构胜在简单，但一旦需要”生成→校验→不达标则重生成””先调研再撰写””关键步骤人工确认”，硬编码的循环就会迅速失控。LangGraph 的核心价值，正是把这些控制流显式建模成图。
图由三类要素组成：State（在节点间传递的数据）、Node（一个处理步骤，可以是普通函数，也可以是一个封装好的 Agent）、Edge（流转方向，含普通边与按条件分支的 Conditional Edge）。当流程需要记住历史、支持中断恢复时，再挂一个 Checkpointer 即可。
举个直观的例子：把一次”写周报”的任务想成图，比把它塞进一个超长提示词要清楚得多——收集数据、起草、主管审核、退回修改，每一步都是图上的一个节点，箭头写明谁接谁。一旦某一步要插入”人工确认”，只需在对应节点前加一条中断边，而不必重写整段逻辑。
| 要素 | 作用 |
|---|---|
| State | 在节点间传递、累积的数据（如对话消息列表） |
| Node | 一个处理步骤：普通函数或预置 Agent |
| Edge | 控制流转方向，连接两个节点 |
| Conditional Edge | 按返回值分支或循环，实现自我纠错 |
| Checkpointer | 持久化状态，支持断点续跑与人工介入 |
二、最小可运行：带自我纠错的图
先看一个能体现”图优势”的例子：生成答案后交给校验节点，不达标就回到生成节点重来，达标才结束。这种”生成-校验”循环用单链要写一堆 if/while，用图只需一条条件边。
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
question: str
answer: str
attempts: int
def generate(state: State) -> dict:
# 调用 LLM 生成答案（此处用伪代码表示实际调用）
reply = llm(state["question"])
return {"answer": reply, "attempts": state["attempts"] + 1}
def validate(state: State) -> str:
# 校验答案是否合规，返回分支名
return "ok" if check(state["answer"]) else "retry"
builder = StateGraph(State)
builder.add_node("generate", generate)
builder.add_node("validate", validate)
builder.add_edge(START, "generate")
builder.add_edge("generate", "validate")
builder.add_conditional_edges(
"validate", validate,
{"ok": END, "retry": "generate"} # 不达标则回到 generate
)
graph = builder.compile()
注意 validate 返回的是字符串分支名，add_conditional_edges 据此决定走向。正是这一行，让你无需手写 while 循环就能表达”最多重试 N 次”的控制流——只要在 generate 里对 attempts 做上限判断即可。
三、多智能体协作：Supervisor 模式
真实业务里，一个 Agent 往往干不完所有事。更常见的做法是” supervisor（主管）+ 多个 worker（专员）”：主管读上下文决定下一步交给谁，各专员执行完再把结果交回主管，直到可以收尾。下面用 create_react_agent 快速封装两个专员。
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import create_react_agent
# 两个专员：一个负责联网调研，一个负责成稿
research = create_react_agent(model, tools=[web_search],
prompt="你是调研助手，只做资料收集")
writer = create_react_agent(model, tools=[],
prompt="你是撰写助手，负责把资料整理成稿")
def supervisor(state: MessagesState) -> str:
last = state["messages"][-1].content
if "需要调研" in last: return "research"
if "可以成稿" in last: return "writer"
return "finish"
g = StateGraph(MessagesState)
g.add_node("supervisor", supervisor)
g.add_node("research", research)
g.add_node("writer", writer)
g.add_edge(START, "supervisor")
g.add_conditional_edges("supervisor", lambda s: s["next"],
{"research": "research", "writer": "writer", "finish": END})
g.add_edge("research", "supervisor")
g.add_edge("writer", "supervisor")
app = g.compile()
这里每个专员本身也是带工具的 Agent（Function Calling 在底层驱动），而 LangGraph 负责”谁先谁后、何时汇合”。比起把多角色逻辑揉进一个超长 prompt，图结构让职责边界和流转一目了然。
主管模式的妙处在于可扩展：你随时能再挂一个”审核专员”，让成稿在对外发布前过一遍事实核查；节点之间是松耦合的，替换或升级某一个专员并不会牵动整条链路。这也是它区别于”把所有指令写进一个系统提示词”的根本原因——能力边界被显式拆分成了可测试的单元。
四、持久化与人工介入
复杂流程常常要”跑一半暂停，等人确认再继续”。LangGraph 用 Checkpointer 把每一步状态存下来，配合 interrupt_before 就能在指定节点前中断。这是单链脚本最难补的能力。
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = graph.compile(checkpointer=memory,
interrupt_before=["validate"]) # 校验前暂停等人审
cfg = {"configurable": {"thread_id": "task-1"}}
app.invoke({"question": "写一份迁移方案"}, cfg)
# 此时流程停在 validate 前，可人工查看中间结果
app.invoke(None, cfg) # 人工确认后继续
thread_id 把整条对话/任务状态串起来，刷新页面、服务重启都不会丢上下文。做客服、审批、长流程自动化时，这个能力几乎是刚需。
五、接上统一网关与对外服务
生产环境里，模型调用最好走一层网关，方便换模型、做配额和审计。我们之前搭过 One API 统一网关，只要把 base_url 指过去即可，代码无需改动模型名以外的任何逻辑。
import os
os.environ["OPENAI_API_KEY"] = "sk-xxxx" # 网关颁发的密钥
os.environ["OPENAI_BASE_URL"] = "https://gw.example.com/v1"
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="qwen2.5-72b-instruct", temperature=0)
如果想把编排好的 Agent 暴露成 HTTP 接口，套一层 FastAPI 就能对外提供服务，再用 Nginx 反向代理 收口域名与证书。本地开发用 VS Code 远程连容器，体验最顺。
from fastapi import FastAPI
api = FastAPI()
@api.post("/agent/run")
async def run(payload: dict):
result = app.invoke({"question": payload["q"]})
return {"answer": result["answer"]}
六、LangGraph 与 Dify、单链 Agent 怎么选
| 方式 | 适合场景 | 灵活性 | 上手成本 |
|---|---|---|---|
| 单链 Agent（ReAct） | 简单问答、单次工具调用 | 低 | 低 |
| LangGraph 图编排 | 多步、需循环/分支/人工介入 | 高 | 中 |
| Dify 可视化 | 业务快速搭建、少写代码 | 中 | 低 |
如果团队不想写代码、要快速搭一个知识库问答，Dify + Ollama 本地 RAG 更合适；如果要精确控制每一步、做多智能体编排和人工审批，LangGraph 是更稳的选择；二者甚至能共存——Dify 负责某个业务节点，LangGraph 负责把多个节点串起来。
七、常见坑与排障
1. State 字段忘了用 Annotated 累积。 消息类状态要用 Annotated[list, operator.add]，否则节点返回会覆盖而不是追加历史，导致上下文丢失。
2. 条件边返回值与映射不一致。 add_conditional_edges 的映射字典必须覆盖函数所有可能返回值，漏掉一个就会在运行时报”no route”错误。
3. 忘了传 thread_id。 用了 Checkpointer 却没在调用时带 configurable.thread_id，状态不会持久化，中断恢复也就失效。
4. 循环没有出口。 条件边指向自身节点时务必设置上限或终止条件，否则会出现无限循环把额度刷光。
八、小结
LangGraph 把”智能体如何协作”从隐含在 prompt 和循环里的黑盒，变成一张可读、可调试、可中断的状态图。它不一定是每个项目的首选——简单任务用单链或 Dify 更快——但一旦你的 AI 应用需要多角色分工、自我纠错或人工把关，图编排就是那把最合适的锤子。
落地建议很明确：先用最小图验证”生成-校验”闭环，再叠 supervisor 做多角色分工，最后用 checkpointer 补上中断恢复。模型层统一走网关、接入层用 FastAPI 加反向代理收口，整套链路就能从本地原型平滑过渡到线上服务。记住，工具是为问题服务的——只有当任务真正需要多步编排与人工把关时，才值得引入图的复杂度；否则，一条 ReAct 链或一个 Dify 工作流反而更轻、更省心。
