LangGraph 多智能体编排：像画流程图一样编排 AI Agent

dev.to · AI#LangChain#Agent#开源

Editor brief · 编辑速览

LangGraph 是 LangChain 开源的多智能体编排框架，本质是一个通用图执行引擎，节点可挂载 AI 模型、工具或人机交互，边描述流转逻辑；文章以凌晨告警排查为案例，从图设计、代码实现、执行模型到上下文传递完整演示，并对比 Dify 与 Claude Code 的动态工作流。

文章思维导图

Knowledge map

拖拽缩放

Full translation

完整中文译文

一个 AI 助手在很多团队中已经是标配：提一个问题，得到一个答案，就像在和一个资深工程师聊天。但一旦你想让多个 AI 协同工作——一个查指标，一个翻日志，一个得出结论——局面就崩溃了：谁交接给谁？它们能并行运行吗？中间要不要停下来问你？如果中途崩溃了，前面的工作是不是都白费了？

LangGraph 就是为了回答这些问题而生的。它是 LangChain 的开源多智能体编排框架（MIT 协议）。但本质要先说清楚：它首先是一个通用的图执行引擎——节点、边、条件边、并行执行、检查点、等待人工介入，这些都是通用的图能力，和 AI 本身没有固有联系；只不过恰好一个节点可以塞进去一个 AI，而一旦塞进去了，你就成了"编排多个 AI 智能体"。核心简单说就是：你画出图，它执行图。

本文用一个具体案例——凌晨 2 点的告警："order-service 错误率飙升"——来从头到尾解释它：先看图，再看代码，再看执行模型和上下文传递，最后和 Dify 以及 Claude Code 的动态工作流做个对比。

案例：凌晨 2 点，order-service 错误率飙升

值班手机响了。告警显示 order-service 错误率刚刚达到 12%，你的 AI 助手需要自主完成整个调查流程。下面这张图就是它将执行的完整逻辑：

图 1 · 深夜调查：一张图调度节点、边、条件边、Send、interrupt 和 checkpoint。

一次完整运行大致是这样的：AI 先诊断，发现情况严重，于是分发 12 个并行子任务逐一检查实例；检查完成后，总结"这 3 个有问题"；然后停下来问你是否要重启——你批准后，它才执行修复，最后复查并报告。

这张图几乎用到了 LangGraph 的每一个核心能力：节点（每个方框）、边（箭头）、条件边（serious?）、Send（批量检查）、interrupt（等待你的审批）、checkpoint（持久化每一步）。下面用代码来画它。

用代码画这张图

构建一个图，把每个方框注册为节点，把箭头连成边：

from langgraph.graph import StateGraph, START, END

def diagnose(state):   # 1 diagnose: check error rate, pull Trace, return severity
    error_rate = query_error_rate(state["service"])
    return {"abnormal": error_rate > 5}

def fan_out(state):    # 3 batch check: Send dispatches 12 subtasks at once
    return [Send("check_host", {"host": h}) for h in state["hosts"]]

def check_host(state): # 4 check one instance (each Send is its own small task)
    return {"results": [f"{state['host']}: ok"]}

def ask_human(state):  # 5 await your approval
    decision = interrupt("3 bad instances found — auto-restart?")
    return {"decision": decision}

def fix(state):        # 6 execute the fix
    restart(state["hosts"])
    return {}

def report(state):     # 7 produce the report
    return {"report": "handled"}

g = StateGraph(dict)
g.add_node("diagnose", diagnose)          # register nodes
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")             # start -> diagnose
g.add_conditional_edges("diagnose", route)# 2 conditional edge: serious? -> fan_out or report
g.add_conditional_edges("fan_out", fan_out)  # 3 Send fan-out
g.add_edge("check_host", "ask_human")     # checked -> await approval
g.add_edge("ask_human", "fix")            # approved -> fix
g.add_edge("fix", "report")               # fixed -> report
g.add_edge("report", END)

graph = g.compile(checkpointer=InMemorySaver())   # compile + enable persistence

几个关键点，逐一说清楚：

1 · 节点就是一个普通函数。 diagnose、check_host、fix 都是普通的 Python 函数。是否调用 LLM 完全由函数体决定——diagnose 可以调模型，check_host 也可以是纯计算。

2 · 条件边决定走哪条路。 route 返回 "fan_out" 或 "report"，图就走到对应的节点：

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

3 · Send = 批量分发。 fan_out 返回 12 个 Send("check_host", {...})，在同一个 superstep 内并行执行，结果自动合并：

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

图 2 · Send 放大：一个 superstep 内分发 N 个独立子任务，然后合并。

4 · interrupt = 停下来等人工。 在节点内调用 interrupt()，图就会暂停，把问题呈现给你；你批准并携带答案恢复后，图从检查点继续执行：

decision = interrupt("3 bad instances found — auto-restart?")
# after you approve:
graph.invoke(Command(resume="yes"),
             {"configurable": {"thread_id": "t1"}})

5 · checkpoint = 持久化每一步。 图在每一步之后存储进度（thread_id 充当工单号）。如果中途崩溃了，用同一个 thread_id 重新调用，它会从断点恢复——官方称之为持久化执行（durable execution）。