LangGraph：用一张图，编排一群 AI 助手

一个 AI 帮手，已经成了不少团队的标配：问一句、答一句，像跟个熟手聊天。可一旦想让好几个 AI 一起干活——一个查指标、一个翻日志、一个出结论——立刻就乱套了：谁来接力？能不能并行？中途要不要停下来问你？干到一半崩了，前面算不算白干？

LangGraph 就是来回答这些问题的。它是 LangChain 开源的多智能体编排框架（MIT 协议）。但先把本质说在前头：它首先是一个通用图执行引擎——节点、边、条件边、并行、存档、等人，都是图的通用能力，跟 AI 没有必然关系；只不过节点恰好可以装 AI，装上就是「编排调度多个 AI Agent」了。核心就是：你画图，它跑图。

这篇文章只用一个案例——凌晨告警「订单服务错误率飙升」——把它从头到尾讲透：先看图、再给代码，然后深入执行机制与上下文传递，最后和 Dify、Claude Code 动态工作流对比。

值班手机响了。告警说订单服务错误率飙到 12%，你的 AI 助手要自己走完整个排查流程。这张图就是它要执行的全部逻辑：

一次跑完大概是这样的：AI 先诊断，发现确实严重，就扇出 12 个并行子任务挨个查实例；查完汇总出「哪 3 台有问题」；然后停下来问你要不要重启——你批了，它才执行修复，最后复查 + 出报告。

这张图里，几乎用到了 LangGraph 的全部核心能力：节点（每个方框）、边（箭头）、条件边（严重？）、Send（批量排查）、interrupt（等你批准）、checkpoint（每一步落盘）。下面用代码把它画出来。

先建一张图，把每个方框注册成节点、箭头连成边：

from langgraph.graph import StateGraph, START, END

def diagnose(state):   # ① 诊断：查错误率、拉 Trace，返回严不严重
    error_rate = query_error_rate(state["service"])
    return {"abnormal": error_rate > 5}

def fan_out(state):    # ③ 批量排查：Send 一次派 12 份子任务
    return [Send("check_host", {"host": h}) for h in state["hosts"]]

def check_host(state): # ④ 查单台实例（每份 Send 一个独立小任务）
    return {"results": [f"{state['host']}: 正常"]}

def ask_human(state):  # ⑤ 等你批准
    decision = interrupt("发现 3 台异常，是否自动重启？")
    return {"decision": decision}

def fix(state):        # ⑥ 执行修复
    restart(state["hosts"])
    return {}

def report(state):     # ⑦ 出报告
    return {"report": "已处理"}

g = StateGraph(dict)
g.add_node("diagnose", diagnose)          # 注册节点
g.add_node("fan_out", fan_out)
g.add_node("check_host", check_host)
g.add_node("ask_human", ask_human)
g.add_node("fix", fix)
g.add_node("report", report)

g.add_edge(START, "diagnose")             # 起点 → 诊断
g.add_conditional_edges("diagnose", route)# ② 条件边：严重？→ 排查 or 出报告
g.add_conditional_edges("fan_out", fan_out)  # ③ Send fan-out
g.add_edge("check_host", "ask_human")     # 查完 → 等批准
g.add_edge("ask_human", "fix")            # 批准 → 修复
g.add_edge("fix", "report")               # 修复 → 出报告
g.add_edge("report", END)

graph = g.compile(checkpointer=InMemorySaver())   # 编译 + 开持久化

逐个看几个关键点：

① 节点就是一个普通函数。diagnose、check_host、fix 都是普通 Python 函数。要不要用 LLM，完全由函数内部决定——diagnose 里可以调模型，check_host 可以是纯计算。

② 条件边决定走哪条路。route 返回 "fan_out" 还是 "report"，图就自动走到对应节点：

def route(state):
    return "fan_out" if state.get("abnormal") else "report"

③ Send = 批量派活。fan_out 返回 12 个 Send("check_host", {...})，同一超步并行执行，结果自动合并：

def fan_out(state):
    return [Send("check_host", {"host": h}) for h in state["hosts"]]

④ interrupt = 停下来等人工。节点里调 interrupt()，图就停住把问题抛给你；你批准后带答案 resume，图从 checkpoint 恢复接着跑：

decision = interrupt("发现 3 台异常，是否自动重启？")
# 你批准后：
graph.invoke(Command(resume="yes"),
             {"configurable": {"thread_id": "t1"}})

⑤ checkpoint = 每一步落盘。图每跑完一步就存一次进度（thread_id 当工单号）。中途崩了，用同一个 thread_id 重新调用，从断点续跑——官方叫 durable execution。

跑图不是一口气从头窜到尾，而是一轮一轮往前走。官方叫超步，一轮就干四件事：

节点之间不传话。大家都对着同一块白板：本轮只读板上已有的字，写下的更新先放一边，这一轮全部跑完，白板才刷新。源码注释就一句：第 N 步写的，第 N+1 步才看得见。

拿图 1 的值班案例走一遍，就是这个节奏：

聊天记录也走同一套：把 messages 放进白板，下一步才看得见。

Dify、Claude Code 都常被叫「多 Agent」，但和 LangGraph 一比，差别一张表就够。

Dify：可视化平台 vs 代码库。 Dify 是网页画布拖节点，LangGraph 是 Python 代码画图。

Claude Code 动态工作流：AI 现写一份 LangGraph。 你只说要做什么，它在后台自动拆任务、编几十到几百个 agent 并行跑。

一句话记：Dify 给你开好的车，Claude Code 是 AI 帮你现写编排，LangGraph 给你发动机和图纸——自由度最大，也最费功夫。

我们开源的 DataBuff（AI 原生 APM，GitHub：github.com/databufflabs/databuff）也做了多 Agent 协同。你只对一个入口说话，AI 大脑把活派给问数、巡检、运维、答疑等专家并行去查，再汇总成带证据链的结论：

LangGraph 的核心目的不是「给你现成的多 Agent 方案」，而是给「需要长期运行、有状态、要持久化、要人机回路的 Agent 流程」提供一个可控的运行时而定——它把「并行、存档、等人、恢复」这些底层脏活做成原语，让你只操心业务流程本身。

Takeaway：把案例里那 6 个函数和图结构抄到本地跑一遍，你就知道多智能体编排是怎么一回事了。