AI Agent工作流编排实战：LangGraph多智能体协作与状态管理方案

AI Agent工作流编排为什么需要LangGraph

构建AI Agent应用时，单次调用大模型往往只能完成一个简单任务。真实业务场景里的智能客服、自动化运维、数据分析助手，都需要把多个LLM调用、工具调用、人工确认串起来形成一条完整工作流。LangGraph把Agent的每一步建模成图节点，节点之间用边连接，状态在节点之间显式传递，解决了传统Chain模式难以处理循环、分支和人工介入的问题。

LangGraph的状态管理基于TypedDict定义schema，每个节点函数接收并返回状态字典。这种设计让多智能体协作变得可控：每个智能体是图中的一个节点，共享一个状态对象，通过条件边决定下一步由谁执行。

LangGraph多智能体协作的图结构与状态定义

先定义一个简单的多智能体状态。以“工单分类-分派-生成回复”为例：

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

class AgentState(TypedDict):
    ticket: str          # 原始工单
    category: str        # 分类结果
    assigned_to: str     # 分派对象
    reply: str           # 最终回复
    history: Annotated[list, operator.add]  # 累积消息

graph = StateGraph(AgentState)

Annotated配合operator.add实现消息列表的累加合并，这在多轮对话场景里很常用。LangGraph会按节点声明的返回键自动合并状态，不需要手动管理上下文拼接。

LangGraph条件路由与循环控制实现

分类Agent输出结果后，根据category决定走哪个分支，这是条件边的典型用法：

def classify(state: AgentState) -> AgentState:
    # 调用分类模型，返回 {"category": "bug", ...}
    return {"category": "bug"}

def route_after_classify(state: AgentState) -> str:
    return state["category"]  # 返回边名

def handle_bug(state: AgentState) -> AgentState:
    return {"assigned_to": "bug-team"}

def handle_feature(state: AgentState) -> AgentState:
    return {"assigned_to": "feature-team"}

graph.add_node("classify", classify)
graph.add_node("bug", handle_bug)
graph.add_node("feature", handle_feature)
graph.add_edge("classify", "bug")   # 兜底
graph.add_conditional_edges(
    "classify", route_after_classify,
    {"bug": "bug", "feature": "feature"}
)

路由函数返回的字符串要跟add_conditional_edges里传入的映射键一一对应。循环场景（比如模型输出格式不合法需要重试）也是条件边返回起点节点实现，配合最大重试次数避免死循环。

LangGraph人工审批与中断恢复机制

涉及外部动作的Agent流程需要人工确认，LangGraph用interrupt_before/interrupt_after把执行停在某个节点之前，通过Command恢复执行：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

graph = StateGraph(AgentState)
# 在send节点前中断，等待人工审批
compiled = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["send"]
)

# 首次执行会停在send前
result = compiled.invoke(
    {"int": "账号无法登录"},
    config={"configurable": {"thread_id": "tk-001"}}
)
# 人工审批通过后恢复
compiled.invoke(
    Command(resume={"approved": True}),
    config={"configurable": {"thread_id": "tk-001"}}
)

checkpointer必须配置，否则中断后状态无法恢复。生产环境用Postgres或Redis checkpoint存储，避免MemorySaver在进程重启后丢失上下文。

LangGraph多智能体协作的失败排查方法

多智能体联调时常见三类问题：状态键名不匹配导致数据丢失、条件边映射键写错导致路由落空、中断后resume传参和节点预期不符。排查时把LangGraph的debug模式打开，逐节点打印state变更；或在节点函数里记录输入输出hash，对比两次运行的差异。状态对象里尽量用不可变字段减少并发写入冲突，可变列表用Annotated加operator.add显式声明合并策略。