本文深入解析了LangGraph框架中的多智能体架构，详细介绍了大模型多智能体系统的概念、优势及架构类型。通过实例讲解如何利用LangGraph实现智能体间的切换与通信，以及共享消息记录的管理方式。此外，还探讨了自定义多智能体架构、监督者架构和群体架构的设计与应用。对于想要了解并实践大模型多智能体系统的开发者来说，本文提供了实用的指导和建议，帮助读者更好地掌握这一前沿技术。

什么是大模型多智能体系统？

大模型多智能体系统（Large Model Multi-Agent System） 是由多个基于大语言模型（LLM）的智能体（Agent）组成的协作系统。每个智能体具备独立的任务处理能力，通过协同工作解决单一智能体难以完成的复杂问题。其核心特征包括：

• • 分布式协作：智能体通过通信、协商或竞争实现目标。
• • 角色分工：不同智能体承担专业角色（如决策者、执行者、验证者）。
• • 共享状态管理：使用共享内存、消息传递或黑板机制同步信息。
• • 动态工作流：任务根据上下文在智能体间动态流转。

示例场景：
一个电商客服系统包含三个智能体：
① 订单查询Agent：访问数据库获取订单状态
② 退换货Agent：处理退货策略
③ 情感安抚Agent：检测用户情绪并安抚
用户提问时，三个Agent协同生成完整响应。

为什么复杂任务需要多智能体协作？

原因包括：

1. 1. 任务分解：复杂任务可以分解为多个子任务，由不同的智能体并行或顺序处理，提高效率。
2. 2. 专长分工：不同的智能体可以专门训练或设计用于特定领域，从而在各自领域表现更好。
3. 3. 错误恢复：当某个智能体失败时，其他智能体可以介入，提高系统的鲁棒性。
4. 4. 知识互补：多个智能体可以拥有不同的知识背景，通过协作融合不同领域的知识。
5. 5. 决策优化：多个智能体可以从不同角度分析问题，通过讨论或投票得到更优的解决方案。

多智能体系统架构

如上图所示，按照多个智能体的相互联系和通信方式，可以将多智能体系统分为以下几种类型的架构：

• • 网络：每个智能体都可以与其他智能体进行通信。任何智能体都可以决定接下来呼叫哪个其他智能体。
• • 监督者：每个智能体与唯一监督者进行通信。监督者智能体决定接下来应该调用哪个智能体。
• • 监督者（工具调用）：这是 监督者架构的特例。单个智能体可以表示为工具。在这种情况下，监督者智能体使用工具调用 LLM 来决定要调用哪些智能体工具，以及要传递给这些智能体的参数。
• • 分层：您可以定义一个多智能体系统，其中包含一个监督者的监督者。这是管理程序架构的推广，允许更复杂的控制流。
• • 自定义多智能体：每个智能体仅与智能体子集通信。流的某些部分是确定性的，只有某些智能体可以决定接下来要呼叫哪些其他智能体。

多智能体之间的通信

切换（Handoffs）

LangGraph使用切换（Handoffs）作为多智能体之间的通信方式。切换就是把流程从一个智能体的任务委托给另一个智能体，允许指定：

• • 目的智能体：要导航到的目标智能体（例如，要转到的节点的名称）
• • 有效负载：要传递给该智能体的信息（例如，状态更新）

LangGraph使用Command来实现切换的动作，例如：

def agent(state) -> Command[Literal["agent", "another_agent"]]:
    # the condition for routing/halting can be anything, e.g. LLM tool call / structured output, etc.
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    return Command(
        # Specify which agent to call next
        goto=goto,
        # Update the graph state
        update={"my_state_key": "my_state_value"}
    )

消息记录在多智能体间的管理方式

多智能体之间通讯是怎么共享消息记录呢？是共享整个完整消息记录还是只是共享最后的结果呢？Langgraph主要还是使用状态对象（state）来管理多智能体之间的消息的共享。

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

class AgentState(TypedDict):
    # 共享的消息历史
    messages: Annotated[List[dict], add_messages]  # 自动追加新消息

共享完整消息记录

共享完整消息记录的好处是，它可以帮助其他智能体做出更好的决策，并提高整个系统的推理能力。
缺点是，随着代理数量及其复杂性的增加，“暂存器”将迅速增长，并且可能需要额外的内存管理策略。

如果运行多个代理，并且需要共享完整的消息记录，那么可以使用同一个状态对象（State Object）即可

示例代码：

def writer_agent(state: AgentState):
    # 读取所有历史消息
    full_history = state["messages"]
    # 生成新消息
    new_msg = {"role": "writer", "content": "基于历史生成报告..."}
    return {"messages": [new_msg]}  # 自动追加到全局历史

def reviewer_agent(state: AgentState):
    # 获取writer刚添加的消息
    last_msg = state["messages"][-1]  
    # 验证内容并回复
    feedback = {"role": "reviewer", "content": f"验证结果: {last_msg['content'][:30]}..."}
    return {"messages": [feedback]}

只是共享智能体的最后输出结果

只是共享其他智能体的最后输出结果，而不是共享整个对话历史。这种方式适用于比较多智能体或者是更加复杂的场景，实现这种方式只需要不同智能体定义自己的状态对象（State Object）即可.

自定义多智能体架构

自定义多智能体架构主要是使用自定义的切换（Handoffs）命令，结合大模型工具调用来实现，以下是一个旅行订票智能体的示例，包括订机票和订酒店两个智能体组成：

import langchain
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command
from langchain_deepseek import ChatDeepSeek

from dotenv import load_dotenv  # 用于加载环境变量
load_dotenv()  # 加载 .env 文件中的环境变量

from langchain.globalsimport set_debug
from langchain.globalsimport set_verbose

set_debug(True)
set_verbose(False)


# 这里需要选择推理的大模型：如 gpt-4o或deepseek-reasoner，而不是聊天模型如deepseek-chat，聊天模型有时无法理解智能体之间的转移和回答逻辑
model = ChatOpenAI(model="gpt-4o")

# 创建一个工具handoff,用于智能体的转移
defcreate_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description orf"Transfer to {agent_name}"

    @tool(name, description=description)
    defhandoff_tool(
        state: Annotated[MessagesState, InjectedState], 
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }
        return Command(  
            goto=agent_name,  
            update={"messages": state["messages"] + [tool_message]},  
            graph=Command.PARENT,  
        )
    return handoff_tool

# 创建Handoffs工具
transfer_to_hotel_assistant = create_handoff_tool(agent_name="hotel_assistant")
transfer_to_flight_assistant = create_handoff_tool(agent_name="flight_assistant")

@tool
defbook_hotel(hotel_name: str):
    """Book a hotel"""
    returnf"成功预订了酒店： {hotel_name}."
@tool
defbook_flight(from_airport: str, to_airport: str):
    """Book a flight"""
    returnf"成功预订了机票从 {from_airport} 到 {to_airport}."

# 机票预订智能体
flight_assistant = create_react_agent(
    # parallel_tool_calls = False, 设置大模型不要并发调用工具，并发调用可能会导致以下错误：
    # ValueError: Found AIMessages with tool_calls that do not have a corresponding ToolMessage.
    model=model.bind_tools([book_flight, transfer_to_hotel_assistant],parallel_tool_calls=False),
    # 绑定两个工具，transfer_to_hotel_assistant 决定当前智能体大模型可以根据需要转移到酒店预订智能体
    tools=[book_flight, transfer_to_hotel_assistant],
    prompt="You are a flight booking assistant",
    name="flight_assistant"
)

# 酒店预订智能体
hotel_assistant = create_react_agent(
    model=model.bind_tools([book_hotel, transfer_to_flight_assistant],parallel_tool_calls=False),
    tools=[book_hotel,transfer_to_flight_assistant],
    prompt="You are a hotel booking assistant",
    name="hotel_assistant"
)

# 定义多智能体图流程
multi_agent_graph = (
    StateGraph(MessagesState)
    .add_node(flight_assistant)
    .add_node(hotel_assistant)
    .add_edge(START, "flight_assistant")
    .compile()
)

for chunk in multi_agent_graph.stream({
        "messages": [
            {
                "role": "user",
                "content": "帮我订一张机票从北京到广州，并且要住在香格里拉大酒店"
            }
        ]
    }
):
    print(chunk)
    print("\n")

运行上述代码，将返回以下结果：

[HumanMessage(content='帮我订一张机票从北京到广州，并且要住在香格里拉大酒店', 
...
ToolMessage(content='成功预订了机票从 北京 到 广州.', name='book_flight', 
...
 ToolMessage(content='Successfully transferred to hotel_assistant', name='transfer_to_hotel_assistant', 
 ...
 ToolMessage(content='成功预订了酒店： 香格里拉大酒店.', name='book_hotel', id='be5ada01-ac03-496d-9456-be824f633e52', tool_call_id='call_vcmljv9xYpkbx6wXlhh0bJJh'), AIMessage(content='您的行程已安排好：机票从北京到广州已经成功预订，并且在广州的香格里拉大酒店的住宿也已成功预订。祝您旅途愉快！',

这里需要特别注意两点：
1、这里需要选择推理能力较强的大模型：如 gpt-4o或deepseek-reasoner，我使用deepseek-chat很难获得正确的结果。不过使用deepseek-reasoner响应速度就比较慢

2、绑定工具时需要设置parallel_tool_calls = False, 设置大模型不要并发调用工具，并发调用可能会导致以下错误：
ValueError: Found AIMessages with tool_calls that do not have a corresponding ToolMessage.

监督者架构（Supervisor）

监督者架构是定义一个监督者智能体，它协调各个智能体的通讯和任务委派，是一个扩展性比较好架构模式。

LangGraph 提供了langgraph-supervisor包可以便利地创建一个监督者多智能体。
如果要使用langgraph-supervisor包，请安装langgraph-supervisor包：

pip install langgraph-supervisor

还是以旅行机票和酒店预订为例，创建一个多智能体示例：

from langchain_deepseek import ChatDeepSeek
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI

from dotenv import load_dotenv  # 用于加载环境变量
load_dotenv()  # 加载 .env 文件中的环境变量


defbook_hotel(hotel_name: str):
    """Book a hotel"""
    returnf"Successfully booked a stay at {hotel_name}."

defbook_flight(from_airport: str, to_airport: str):
    """Book a flight"""
    returnf"Successfully booked a flight from {from_airport} to {to_airport}."

# 机票预订智能体
flight_assistant = create_react_agent(
    model="openai:gpt-4o",
    tools=[book_flight],
    prompt="You are a flight booking assistant",
    name="flight_assistant"
)

# 酒店预订智能体
hotel_assistant = create_react_agent(
    model="openai:gpt-4o",
    tools=[book_hotel],
    prompt="You are a hotel booking assistant",
    name="hotel_assistant"
)

# 创建一个监督者
supervisor = create_supervisor(
    agents=[flight_assistant, hotel_assistant],
    model=ChatOpenAI(model="gpt-4o"),
    # full_history 全消息记录，last_message 最后智能体的输出
    output_mode="full_history",
    prompt=(
        "You manage a hotel booking assistant and a"
        "flight booking assistant. Assign work to them."
    )
).compile()

for chunk in supervisor.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "帮我订一张机票从北京到广州，并且要住在香格里拉大酒店"
            }
        ]
    }
):
    print(chunk)
    print("\n")

输出：

[HumanMessage(content='帮我订一张机票从北京到广州，并且要住在香格里拉大酒店', 
...
ToolMessage(content='Successfully transferred to flight_assistant', name='transfer_to_flight_assistant', id='ec6038b2-d8e4-485f-9359-e95047838ab1', tool_call_id='call_QM3Yy1ZRk9kFyAftxG0NwVee'), AIMessage(content='您的航班已成功预订，从北京（PEK）飞往广州（CAN）。关于香格里拉大酒店的住宿预订，请您联系酒店预订服务。祝您旅途愉快 ！如果还有其他问题或需要帮助，请告诉我。', 
...
 ToolMessage(content='Successfully transferred back to supervisor', name='transfer_back_to_supervisor', 
 ...
 ToolMessage(content='Successfully transferred to hotel_assistant', name='transfer_to_hotel_assistant', 

 ...
 AIMessage(content='我已经帮您成功预订了香格里拉大酒店的住宿。如果您有其他需求或者问题，请随时告诉我！祝您旅途愉快！', 
 ...

ToolMessage(content='Successfully transferred back to supervisor', name='transfer_back_to_supervisor', 
...
AIMessage(content='您的航班已成功预订，从北京飞往广州。此外，我也帮您成功预订了香格里拉大酒店 的住宿。祝您旅途愉快！若有其他需求，请随时联系我。',

监督者架构就是每次智能体执行完成后，都会转移到监督者，进行判断，如果满足条件，则将结果返回给用户或者再转移到下一个智能体。从架构上更加清晰，更加符合实际场景。

消息记录模式：
create_supervisor时指定output_mode参数值，full_history 全消息记录，last_message 最后智能体的输出，可以比较节省上下文token

多层监督者架构

如果需要复杂的多组智能体组成，可以构建多层监督者架构，比如：

research_team = create_supervisor(
    [research_agent, math_agent],
    model=model,
    supervisor_name="research_supervisor"
).compile(name="research_team")

writing_team = create_supervisor(
    [writing_agent, publishing_agent],
    model=model,
    supervisor_name="writing_supervisor"
).compile(name="writing_team")

top_level_supervisor = create_supervisor(
    [research_team, writing_team],
    model=model,
    supervisor_name="top_level_supervisor"
).compile(name="top_level_supervisor")

群体架构（swarm）

前面介绍了网络多智能体的架构，其实就是一种群体的智能体架构。
LangGraph也支持这种架构，通过引入langgraph-swarm包，可以轻松实现。
安装langgraph-swarm包

pip install langgraph-swarm

还是以订机票和酒店为例：

from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_swarm, create_handoff_tool
from langchain_core.tools import  tool
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI


from dotenv import load_dotenv  # 用于加载环境变量
load_dotenv()  # 加载 .env 文件中的环境变量

# model = ChatDeepSeek(model="deepseek-reasoner")
model = ChatOpenAI(model="gpt-4o")


@tool
defbook_hotel(hotel_name: str):
    """Book a hotel"""
    returnf"Successfully booked a stay at {hotel_name}."

@tool
defbook_flight(from_airport: str, to_airport: str):
    """Book a flight"""
    returnf"Successfully booked a flight from {from_airport} to {to_airport}."

transfer_to_hotel_assistant = create_handoff_tool(
    agent_name="hotel_assistant",
    description="Transfer user to the hotel-booking assistant.",
)


transfer_to_flight_assistant = create_handoff_tool(
    agent_name="flight_assistant",
    description="Transfer user to the flight-booking assistant.",
)

flight_assistant = create_react_agent(
    model=model.bind_tools([book_flight,transfer_to_hotel_assistant], parallel_tool_calls=False),

    tools=[book_flight, transfer_to_hotel_assistant],
    # 有些大模型parallel_tool_calls=False可能不生效，为避免并发调用工具的错误，可以在提示词加上：一次只能调用一个工具，工具不能并发调用
    # Only one tool can be called at a time, and tools cannot be called in parallel
    prompt="You are a flight booking assistant, Only one tool can be called at a time, and tools cannot be called in parallel",
    name="flight_assistant"
)
hotel_assistant = create_react_agent(
    model=model.bind_tools([book_hotel,transfer_to_flight_assistant], parallel_tool_calls=False),

    tools=[book_hotel, transfer_to_flight_assistant],
    prompt="You are a hotel booking assistant, Only one tool can be called at a time, and tools cannot be called in parallel",
    name="hotel_assistant"
)

# 定义一个群体的智能体节点
swarm = create_swarm(
    agents=[flight_assistant, hotel_assistant],
    default_active_agent="flight_assistant"
).compile()

for chunk in swarm.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "帮我订一张机票从北京到广州，并且要住在香格里拉大酒店"
            }
        ]
    }
):
    print(chunk)
    print("\n")

运行以上代码，会输出：

HumanMessage(content='帮我订一张机票从北京到广州，并且要住在香格里拉大酒店', 
...

ToolMessage(content='Successfully booked a flight from Beijing to Guangzhou.', name='book_flight', 

...
ToolMessage(content='Successfully transferred to hotel_assistant', name='transfer_to_hotel_assistant', 
...
ToolMessage(content='Successfully booked a stay at 香格里拉大酒店.', name0', AIMessage(content='您的行程已 经成功安排：您将从北京飞往广州，并入住香格里拉大酒店。祝您旅途愉快！如果还有其他需求，请告诉我。'

小结

1. 多智能体架构之间的切换原理主要是通过Handoff 和 tool来实现的
2. LangGraph 提供了自定义的Handoff函数，也提供了langgraph-supervisor和langgraph-swarm 包来方便实现监督者架构和群体架构
3. 避坑点：

• 选择大模型很重要，需要推理能力比较强的才能较好地处理多智能体，或者可以优化更细致的提示词
• 大模型默认为并发调用工具，会导致报错，需要配置parallel_tool_calls=False，或者也可以通过提示词来控制不并行调用工具

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

为什么要学习大模型？

我国在A大模型领域面临人才短缺,数量与质量均落后于发达国家。2023年，人才缺口已超百万，凸显培养不足。随着AI技术飞速发展，预计到2025年,这一缺口将急剧扩大至400万,严重制约我国AI产业的创新步伐。加强人才培养,优化教育体系,国际合作并进是破解困局、推动AI发展的关键。

大模型入门到实战全套学习大礼包

1、大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

---

2、大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

3、AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

4、大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

5、大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

适用人群

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】