纲要

• 单智能体的局限与多智能体的兴起
• 多智能体架构类型
  • 网状结构
  • 监管者模式
  • 智能体作为工具
  • 分级架构
  • 自定义工作流
• 多智能体的优势与挑战
• LangGraph：多智能体编排框架
• 实战：用 LangGraph 构建监管者多智能体系统
  • 项目结构
  • 依赖安装
  • 状态定义
  • 工人节点实现
  • 监管者节点实现
  • 条件路由
  • 图构建与编译
  • 完整可运行代码
• API 速览
• Demo 示例
• 参考文档
• 总结

从单兵作战到群体智慧

在单智能体系统中，一个模型实例挂载多个工具，通过记忆机制和推理循环完成用户请求。这种“一个大脑加一组工具”的架构能够解决大量常规任务，但当任务复杂度上升到需要并行处理、多领域知识协作或结果相互校验时，单智能体的局限性便逐渐显现。

多智能体系统（Multi-Agent System, MAS）通过引入多个专业化的智能体，让它们各司其职、相互通信，从而实现更强的鲁棒性、可扩展性和输出质量。这一架构范式将AI应用从“单兵作战”升级为“特种小队”协同。

多智能体架构一览

多智能体系统的组织方式直接影响其能力边界与实现复杂度。常见的架构模式包括以下几种：

架构类型	特点	典型场景
网状结构	智能体之间双向通信，所有节点平等决策	分布式协商、去中心化协作
监管者模式	一个管理者负责拆分任务、分派给工人，并汇总结果	客服系统、复杂工作流
智能体作为工具	将多个智能体封装为工具，由主模型调用	过渡方案、简单扩展
分级架构	多层管理，上级监管下级，模拟组织层级	大型项目、多阶段任务
自定义工作流	通过图结构自由组合节点与边，混合确定性与智能体逻辑	需要灵活控制的复杂场景

以监管者模式（Supervisor Pattern）为例，其结构类似于一个团队：监管者接收用户需求，将任务拆分后分派给不同的工人智能体，工人完成任务后返回结果，监管者整合后输出最终答案。这种模式结构清晰，适合作为多智能体开发的入门实践。

LangGraph：多智能体编排框架

LangGraph 是 LangChain 生态中用于构建有状态、多步骤智能体工作流的底层编排框架。其核心设计理念是用图（Graph）来建模工作流：每个节点可以是一个智能体、一个工具或一段处理逻辑，边定义了数据流转方向，包括条件边（Conditional Edges）。通过组合节点和边，开发者可以快速构建上述任意一种多智能体架构。

LangGraph 的低层级原语（low-level primitives）提供了构建完全自定义智能体所需的灵活性，支持单智能体、多智能体、分级架构等多种控制流。LangGraph 自 0.2 版本起成为 LangChain 官方推荐的智能体构建方式，0.3 版本进一步将预置组件拆分到独立的 langgraph-prebuilt 包中。

实战：用 LangGraph 构建监管者多智能体系统

本节实现一个简单的监管者系统：监管者分析用户问题，决定调用“搜索工人”还是“写作工人”，最终整合输出。

项目结构

multi_agent_demo/
├── main.py
├── .env
└── requirements.txt

依赖安装

pip install langgraph langchain-openai langchain-core python-dotenv

状态定义

LangGraph 的核心概念之一是共享状态（Shared State）。所有节点通过读写同一个状态对象进行通信。状态中的每个字段可以标注 reducer 函数，用于合并来自多个节点的更新。

import os
from typing import TypedDict, Annotated, Sequence, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END, add_messages

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next_worker: Literal["search", "writer", "supervisor", "end"]

add_messages 是 langgraph 内置的 reducer，用于将新消息追加到消息列表中，而非覆盖。

工人节点实现

每个工人节点接收当前状态，执行特定任务后返回更新后的状态。

def search_worker(state: AgentState) -> dict:
    """搜索工人：模拟网络搜索，实际可接入 SerpAPI 等搜索工具"""
    query = state["messages"][-1].content
    result = f"搜索结果：关于「{query}」的最新信息显示，多智能体架构正成为AI应用的主流范式。"
    return {"messages": [AIMessage(content=result)], "next_worker": "supervisor"}

def writer_worker(state: AgentState) -> dict:
    """写作工人：根据已有素材生成最终回答"""
    context = state["messages"][-1].content
    response = llm.invoke(f"根据以下素材，写一段专业的技术总结：\n{context}")
    return {"messages": [response], "next_worker": "end"}

监管者节点实现

监管者分析当前用户消息，决定下一步调用哪个工人。

def supervisor(state: AgentState) -> dict:
    """监管者：分析用户意图，决定下一个调用的工人"""
    user_msg = state["messages"][-1].content
    if "搜索" in user_msg or "最新" in user_msg:
        next_worker = "search"
    else:
        next_worker = "writer"
    return {"next_worker": next_worker}

条件路由

条件边（Conditional Edge）根据状态中的 next_worker 字段动态决定路由目标。

def route(state: AgentState) -> Literal["search", "writer", "end"]:
    return state["next_worker"]

图构建与编译

workflow = StateGraph(AgentState)

workflow.add_node("supervisor", supervisor)
workflow.add_node("search", search_worker)
workflow.add_node("writer", writer_worker)

workflow.set_entry_point("supervisor")

workflow.add_conditional_edges(
    "supervisor",
    route,
    {
        "search": "search",
        "writer": "writer",
        "end": END,
    }
)

workflow.add_edge("search", "writer")
workflow.add_edge("writer", END)

app = workflow.compile()

StateGraph 是构建器类，必须调用 .compile() 后生成可执行的图实例，支持 invoke()、stream()、astream() 等方法。

完整可运行代码

import os
from typing import TypedDict, Annotated, Sequence, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END, add_messages

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next_worker: Literal["search", "writer", "supervisor", "end"]

def search_worker(state: AgentState) -> dict:
    query = state["messages"][-1].content
    result = f"搜索结果：关于「{query}」的最新信息显示，多智能体架构正成为AI应用的主流范式。"
    return {"messages": [AIMessage(content=result)], "next_worker": "supervisor"}

def writer_worker(state: AgentState) -> dict:
    context = state["messages"][-1].content
    response = llm.invoke(f"根据以下素材，写一段专业的技术总结：\n{context}")
    return {"messages": [response], "next_worker": "end"}

def supervisor(state: AgentState) -> dict:
    user_msg = state["messages"][-1].content
    if "搜索" in user_msg or "最新" in user_msg:
        next_worker = "search"
    else:
        next_worker = "writer"
    return {"next_worker": next_worker}

def route(state: AgentState) -> Literal["search", "writer", "end"]:
    return state["next_worker"]

workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("search", search_worker)
workflow.add_node("writer", writer_worker)
workflow.set_entry_point("supervisor")

workflow.add_conditional_edges(
    "supervisor",
    route,
    {
        "search": "search",
        "writer": "writer",
        "end": END,
    }
)
workflow.add_edge("search", "writer")
workflow.add_edge("writer", END)

app = workflow.compile()

if __name__ == "__main__":
    inputs = {"messages": [HumanMessage(content="搜索最新的多智能体发展趋势")]}
    for output in app.stream(inputs):
        for key, value in output.items():
            if "messages" in value and value["messages"]:
                last_msg = value["messages"][-1]
                print(f"节点 '{key}': {last_msg.content}")

    print("\n---\n")

    inputs2 = {"messages": [HumanMessage(content="写一段关于人工智能的介绍")]}
    for output in app.stream(inputs2):
        for key, value in output.items():
            if "messages" in value and value["messages"]:
                last_msg = value["messages"][-1]
                print(f"节点 '{key}': {last_msg.content}")

运行说明

1. 在项目根目录创建 .env 文件，配置 OPENAI_API_KEY=your_api_key。
2. 执行 python main.py。
3. 观察控制台输出：监管者根据输入内容自动决定调用搜索工人还是直接进入写作工人，搜索结果会传递给写作工人，最终生成回答。

API 速览

StateGraph

所属库：langgraph.graph

描述：StateGraph 是 LangGraph 的主要图构建器类，用于构建有状态的、多参与者（multi-actor）应用。节点通过读写共享状态进行通信。

方法签名：

class StateGraph(StateSchemaType, input_schema=None, output_schema=None)

核心方法：

• add_node(name: str, action: Callable)：向图中添加节点
• add_edge(start: str, end: str)：添加普通边
• add_conditional_edges(source: str, condition: Callable, path_map: dict)：添加条件边
• set_entry_point(node: str)：设置入口节点
• compile()：编译图，返回可执行的 CompiledStateGraph 实例

add_messages

所属库：langgraph.graph

描述：内置的 reducer 函数，用于将新消息追加到消息列表中，而非覆盖。

用法：

from langgraph.graph import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

create_react_agent（参考）

所属库：langgraph.prebuilt（0.3 版本后移至 langgraph-prebuilt 包）

描述：创建 ReAct 风格的工具调用智能体的工厂函数。

注意：该函数已被标记为弃用（deprecated），官方推荐使用 LangChain 的 create_agent 替代。

Demo 示例

运行说明

1. 创建项目目录 multi_agent_demo/，将上述完整代码保存为 main.py。
2. 创建 .env 文件，填入 OPENAI_API_KEY。
3. 创建 requirements.txt：

   langgraph>=0.2.0
   langchain-openai>=0.2.0
   langchain-core>=0.2.0
   python-dotenv>=1.0.0
   

4. 执行 pip install -r requirements.txt。
5. 运行 python main.py。

代码说明

• 状态管理：AgentState 定义了所有节点共享的数据结构，messages 使用 add_messages reducer 实现消息累积。
• 监管者路由：supervisor 节点通过简单的关键词匹配决定路由目标，实际生产环境可替换为基于 LLM 的意图分类。
• 条件边：add_conditional_edges 根据 next_worker 字段动态选择下一个节点。
• 流式输出：使用 app.stream() 而非 invoke()，可实时观察每个节点的中间输出。

技术点总结

• StateGraph 的声明式图构建 API
• 共享状态与 reducer 机制
• 条件边实现动态路由
• 多智能体协作的监管者模式

参考文档

官方文档

• LangGraph 官方文档
• LangGraph Python API 参考
• LangGraph 多智能体概念
• LangGraph 多智能体教程
• LangGraph Supervisor 参考

参考链接

• LangChain 多智能体指南
• LangGraph GitHub 仓库
• LangGraph 0.3 发布说明
• LangGraph v1 迁移指南

总结

多智能体架构为复杂 AI 应用提供了强大的组织能力，LangGraph 则是实现这些架构的核心编排框架。本文系统介绍了五种常见的多智能体架构模式，并通过一个监管者系统的完整代码示例，展示了如何使用 StateGraph 构建有状态的多智能体协作工作流。

核心技术要点包括：共享状态与 reducer 机制、条件边实现动态路由、以及节点化的智能体封装。掌握这些基础后，开发者可以轻松扩展到分级架构、评估-修正循环、群体协作（Swarm）等更高级的设计模式。