Skip to main content
Version: Next

LangGraph 与多智能体

LangGraph 是当前 Agent 编排的主流方案,面试重点四个概念 + 两个机制:State/Node/Edge/Compile、Reducer 消息合并、interrupt 人机协同、什么时候拆多智能体。

四个核心概念​

State(状态) —— 共享的数据结构(TypedDict/Pydantic),全图流转
Node(节点) —— 一个函数:收 state,干活,返回要更新的字段
Edge(边) —— 节点间的跳转:固定边 / 条件边(按 state 判断)
Compile(编译) —— graph.compile() 产出可执行的 CompiledGraph

节点只返回"要改的字段",不返回整个 state:

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

class State(TypedDict):
messages: Annotated[list, add_messages] # 关键:指定 reducer
done: bool

def chatbot(state: State):
# 只返回要更新的字段,其余字段原样保留
return {"messages": [model.invoke(state["messages"])]}

def route(state: State) -> str:
last = state["messages"][-1]
return "tools" if getattr(last, "tool_calls", None) else END

graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_conditional_edges("chatbot", route)
app = graph.compile()

Reducer:消息为什么会被吞​

State 字段更新有两种策略:

  • 默认:后写覆盖——节点返回 {"messages": [新消息]},旧 messages 整条被覆盖掉,历史就丢了。
  • Reducer:Annotated[list, add_messages]——返回的值追加到现有列表后面。

这就是"消息必须 add_messages,否则前一条被吞"的原因:chatbot 节点写了一轮新消息,tools 节点再写工具结果,默认覆盖策略下每一步都把前一步冲掉,最后 state 里只剩最后一条。多轮 Agent 的 state 定义里,messages 字段永远要挂 add_messages。

条件边 + 回边 = 循环,必须有刹车​

Agent 的本质循环是:chatbot →(有 tool_calls)→ tools →(回边)→ chatbot →…… 条件边加回边构成了有环的图,这正是不用 LCEL 而用 LangGraph 的原因(LCEL 是单向无环管道)。

环必须有刹车,否则模型反复调工具就是死循环烧钱:

  • 业务层:单工具调用次数上限、重复调用检测;
  • 框架层:recursion_limit(编译后调用时传入,超限直接抛错兜底)。

checkpointer、thread_id 与记忆​

checkpointer + thread_id 才有记忆:

from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(checkpointer=MemorySaver())

# 同一 thread_id 的多次调用共享检查点 = 同一会话
app.invoke({"messages": [...]}, config={"configurable": {"thread_id": "user-42-session-1"}})

checkpointer 在每一步(超步)后把 state 存下来;thread_id 标识一条会话线程——多用户、多会话隔离全靠它。不配 checkpointer,图每次执行都是全新的,没有记忆。

interrupt:人机协同与恢复​

危险操作前暂停等人工审批:

  • interrupt() 必须在节点内部调用(不是图外面)——它抛出一个特殊信号,框架把当前 state 存入 checkpointer 后停机。
  • 恢复时必须用同一个 thread_id:框架靠它找到"停在哪一步、state 是什么",从断点继续。
  • 超步数恢复的坑:如果中断发生在"模型已输出 tool_calls、工具还没执行"之间,恢复前必须把悬空的 tool_calls 补上对应的 ToolMessage(哪怕内容是"已人工拒绝"),否则 messages 里有一条没有结果的工具调用,下次调模型直接报 400。

多智能体:什么时候拆​

职责、工具、提示词都不同才拆。只差提示词的两个"Agent"其实是同一个 Agent 的两个 prompt,拆开只增加复杂度。合理拆分的例子:客服系统里"查订单 Agent"(订单查询工具)、"退款 Agent"(退款+风控工具)、"售前 Agent"(商品知识库)——职责边界清晰、工具集不重叠。

Supervisor 模式:主管节点只做路由——读用户请求和对话历史,决定转给哪个工人。关键细节:要告诉 Supervisor"谁已经干过"(在 state 里维护已完成节点列表),否则主管可能把同一个问题反复派给同一个工人,转圈出不去。

用户 → Supervisor ─┬→ 查订单 Agent ─┐
├→ 退款 Agent ─┼→ 汇总回 Supervisor → 结束
└→ 售前 Agent ─┘

LCEL 与 LangGraph 的本质区别​

维度LCELLangGraph
结构单向无环管道(a 串联 b 串联 c)有环的状态图
状态数据一路向后流,不回传共享 State,节点读写
循环不支持条件边 + 回边
持久化/中断无checkpointer + interrupt
适合简单链式任务(检索→生成)Agent 循环、审批、多智能体

一句话:需要"绕回来"和"停下来",就上 LangGraph。

高频面试题​

问题答题要点
LangGraph 四个概念?State/Node/Edge/Compile;节点只返回要改的字段
消息为什么会被吞?默认更新是后写覆盖;messages 字段必须挂 add_messages reducer 才是追加
条件边 + 回边是什么?怎么防死循环?构成有环图;业务层次数上限 + 重复检测,框架层 recursion_limit
checkpointer 和 thread_id 干什么用?每超步存 state,thread_id 隔离会话;两者配合才有跨请求记忆
interrupt 怎么恢复?为什么必须同一个 thread_id?节点内调 interrupt,框架存档停机;同 thread_id 才能定位断点与 state;恢复前补齐悬空 tool_calls
什么时候该拆多智能体?职责、工具、提示词都不同才拆;Supervisor 只做路由且要知道"谁干过"
LCEL 和 LangGraph 的区别?无环管道 vs 有环状态图;要循环、暂停审批、持久化就上 LangGraph