大模型Agent编排框架实战：LangGraph状态图与多智能体协作流程设计

大模型Agent编排框架是构建复杂AI应用的核心基础设施。LangGraph作为LangChain团队推出的状态图编排框架，通过有向图模型管理Agent的执行流程，支持条件路由、循环回退、人工介入等高级模式，解决了单链Agent在面对多步骤、多分支任务时的脆弱性问题。LangGraph的核心思想是将Agent执行过程建模为状态机，每个节点是一个函数或LLM调用，边定义状态转移逻辑，从而实现对复杂工作流的精确控制。

LangGraph核心概念与状态图模型

LangGraph的基础数据结构是StateGraph，其工作原理是维护一个全局状态字典（TypedDict），每个节点函数接收当前状态作为输入，返回状态更新部分，框架自动合并到全局状态中。这种设计使节点之间解耦，同时共享上下文信息。

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

class AgentState(TypedDict):
    messages: Annotated[List, add_messages]
    next_agent: str
    research_results: str
    final_answer: str

def researcher_node(state: AgentState) -> dict:
    messages = state["messages"]
    # 调用检索工具获取上下文
    result = retrieval_tool.invoke(messages[-1].content)
    return {"research_results": result, "next_agent": "writer"}

def writer_node(state: AgentState) -> dict:
    context = state["research_results"]
    answer = llm.invoke(f"基于以下资料回答问题：{context}")
    return {"final_answer": answer.content}

# 构建状态图
graph = StateGraph(AgentState)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", END)

app = graph.compile()

StateGraph的add_messages注解器实现了消息列表的追加语义，新消息不会覆盖旧消息，而是追加到列表末尾。这个细节在多轮对话场景中至关重要。

单Agent状态图构建与条件路由

实际业务中，Agent需要根据中间结果决定下一步走向。LangGraph通过conditional_edges实现条件路由，路由函数返回目标节点名称，框架据此跳转。

def route_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

def call_model(state: AgentState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def call_tools(state: AgentState) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        result = tool_map[tc["name"]].invoke(tc["args"])
        results.append({"role": "tool", "content": result, "tool_call_id": tc["id"]})
    return {"messages": results}

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)
graph.set_entry_point("agent")
graph.add_conditional_edges(
    "agent",
    route_tools,
    {"tools": "tools", END: END}
)
graph.add_edge("tools", "agent")  # 工具执行后回到Agent
app = graph.compile()

上述代码实现了经典的ReAct循环：Agent调用模型→判断是否需要工具→执行工具→将结果反馈给模型。conditional_edges的第三个参数是路由映射表，显式声明可能的跳转目标，避免隐式行为。

多智能体协作模式：Supervisor与Hierarchical架构

当任务复杂度超出单Agent能力时，LangGraph支持多智能体协作。Supervisor模式由一个中央调度Agent决定将任务分发给哪个子Agent，子Agent执行完毕后返回结果给Supervisor，由Supervisor决定是否继续分发或结束。

from langgraph.graph import StateGraph, END

class MultiAgentState(TypedDict):
    messages: Annotated[List, add_messages]
    next: str

def supervisor(state: MultiAgentState) -> dict:
    # Supervisor LLM决定下一个执行的Agent
    response = supervisor_llm.invoke(
        f"对话历史：{state['messages']}
选择下一个Agent：researcher / coder / FINISH"
    )
    next_agent = parse_decision(response)
    return {"next": next_agent}

def researcher(state: MultiAgentState) -> dict:
    result = research_agent.invoke(state["messages"])
    return {"messages": [result], "next": "supervisor"}

def coder(state: MultiAgentState) -> dict:
    result = code_agent.invoke(state["messages"])
    return {"messages": [result], "next": "supervisor"}

def route_supervisor(state: MultiAgentState) -> str:
    next_agent = state["next"]
    if next_agent == "FINISH":
        return END
    return next_agent

graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route_supervisor,
    {"researcher": "researcher", "coder": "coder", END: END})
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
app = graph.compile()

Supervisor模式的优点是中心化决策，易于调试和追踪。Hierarchical模式在此基础上允许子Agent内部也包含Supervisor，形成多层委托结构，适用于超大规模任务分解。

LangGraph持久化与人工介入节点

LangGraph通过Checkpointer实现状态持久化，每个节点执行后自动保存快照。配合thread_id参数，可以恢复任意历史状态的执行。人工介入（Human-in-the-Loop）通过interrupt_before或interrupt_after实现，框架在指定节点前后暂停执行，等待外部输入后继续。

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

# 在"review"节点前暂停，等待人工审核
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)

# 首次执行，会在review节点前暂停
config = {"configurable": {"thread_id": "session-001"}}
result = app.invoke(initial_state, config=config)

# 人工审核后，传入审核结果继续执行
result = app.invoke(
    {"messages": [HumanMessage(content="审核通过，继续执行")]},
    config=config
)

MemorySaver将状态保存在内存中，适合开发环境。生产环境应使用SqliteSaver或PostgresSaver，确保持久化数据不丢失。thread_id的设计建议使用业务ID，如订单号或工单号，便于跨会话追踪。

生产环境部署与性能调优

LangGraph应用部署到生产环境时，需要关注几个关键配置项。并发控制通过max_concurrency参数限制同时处理的请求数，防止LLM API限流。超时控制通过recursion_limit限制图的最大执行步数，防止Agent陷入无限循环。

app = graph.compile(
    checkpointer=PostgresSaver(conn_string),
    interrupt_before=["human_review"]
)

# 流式输出，逐步返回结果
for event in app.stream(
    initial_state,
    config={"configurable": {"thread_id": "task-123"},
            "recursion_limit": 50},
    stream_mode="values"
):
    if "final_answer" in event:
        print(event["final_answer"])

stream_mode参数支持values（返回完整状态）、updates（返回增量更新）和messages（返回LLM消息流）三种模式。对于聊天场景，使用messages模式可以实现打字机效果，用户体验更好。

LangGraph在0.2版本后引入了Subgraph概念，允许将一个编译好的Graph作为另一个Graph的节点使用，实现图嵌套。这对于构建大型Agent系统尤为重要，可以将不同团队的Agent模块独立开发和测试，再通过Subgraph组合部署。

监控方面，LangGraph原生支持LangSmith追踪，开启后所有节点执行、LLM调用、工具调用的输入输出都会自动上报。通过LangSmith的Dashboard可以查看每步执行耗时、Token消耗、错误堆栈，快速定位性能瓶颈。