概念

MultiAgent系统是什么意思

• 一个Agent通常是一个 LLM 驱动的系统，用来决定流程控制（比如调用哪一个工具、执行什么操作）。
• 当系统变得复杂，例如：
  • 一个Agent要管理太多工具，难以选用最优工具。
  • 上下文／记忆变得太大，单一Agent难以处理。
  • 不同任务需要不同专长（如“规划者”、“研究者”、“数学专家”）时。
• 因此，MultiAgent系统就是将大的任务拆分为多个专门的Agent，协作完成。
• 使用MultiAgent系统可以带来：模块化（更易开发、测试、维护）、专长化（每个Agent聚焦某一领域）、更好控制Agent之间的通讯与流程。

主要模式 & 架构

架构类型

• 网络型（Network）：每个Agent可能与其它任意Agent通信，任意Agent可决定调用哪个Agent。
• 监督型（Supervisor）：存在一个“主管”Agent，负责决策调用哪个子Agent。
• 分层型（Hierarchical）：一层主管下还有下一级主管－Agent结构，更复杂的控制流程。

主要运作模式

设计时需要考虑的事项

• Agent职责清晰：为每个Agent定义明确任务／专长。
• 流程控制：决策哪个Agent什么时候被调用、何时交接，是设计中关键。
• 上下文管理：Agent要看到多少历史、哪些状态？输出给下一个Agent又包括什么？
• 模块化与可维护性：每个Agent最好可以单独开发、调试、优化。
• 工具分配：每个Agent可以拥有自己特定的工具集合，避免一个Agent用太多不同领域工具导致混乱。

编码示例

下面代码 仅仅是简单的一个流程示例，现将下面的代码跑通，后续可以在上面进行扩展，比如上下文记忆，工具记忆，响应结果格式化等操作。

langchain==1.0.3

langgraph==1.0.2

Tool Calling 模式（主管调用子代理作为工具）

“总入口Agent（Supervisor）”，负责回答用户的问题。
系统中有两个子Agent：

• MathAgent：负责计算问题
• InfoAgent：负责解释概念

主Agent判断问题类型，决定调用哪个子Agent。

代码

from langchain.tools import tool
from langchain.agents import create_agent
from langgraph.config import get_stream_writer
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

llm = ChatOpenAI(
        temperature=0.75,
        model="qwen-max",
        api_key=SecretStr(DASHSCOPE_API_KEY),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )

# ================== 两个子Agent初步定义下 ==================
math_agent = create_agent(
    model=llm,
    system_prompt="你是数学专家，只负责进行数学计算并返回答案。"
)

info_agent = create_agent(
    model=llm,
    system_prompt="你是知识解释专家，负责解释概念和原理。"
)

# ================== 子Agent包装成工具 ==================
@tool("math_agent", description="用于数学计算")
def call_math_agent(query: str):
    writer = get_stream_writer()
    writer(f"\n 调用[math_agent] 开始计算：{query}")

    final_result = ""
    # 工具子代理流式调用，使用 custom 输出进度
    for mode, chunk in math_agent.stream(
        {"messages": [{"role": "user", "content": query}]},
        stream_mode=["messages", "custom"]
    ):
        # writer(chunk)
        # print("=====>>>>>>>>",chunk)
        # final_result += str(chunk)

        # 试试看工具的结果 流式 输出到主agent
        # 试试看工具的结果 流式 输出到主agent

        if mode == "messages":
            if chunk[0].content:
                final_result += chunk[0].content
                # print("====>",chunk[0])
                writer(chunk[0].content.strip())

    writer(f"\n 使用[math_agent] 计算完成。")
    return final_result or "数学计算完成"

@tool("info_agent", description="用于解释问题或提供知识说明")
def call_info_agent(query: str):
    writer = get_stream_writer()
    writer(f"调用[info_agent] 开始解释：{query}")

    final_result = ""
    # 工具子代理流式调用
    for chunk in info_agent.stream(
        {"messages": [{"role": "user", "content": query}]},
        stream_mode="custom"
    ):
        writer(chunk)
        print("=====>>>>>>>>",chunk)
        final_result += str(chunk)

    writer(f"[info_agent] 解释完成。")
    return final_result or "解释完成"

# ================== 主代理 Supervisor ==================
supervisor = create_agent(
    model=llm,
    system_prompt=(
        "你是总管代理，负责判断用户问题类型。"
        "数学计算用 math_agent，知识解释用 info_agent，不需要工具就直接回答。"
        "请始终用自然口语回答。"
    ),
    tools=[call_math_agent, call_info_agent],
)

# ================== 对话循环 + 双层流式输出 ==================
def chat():
    print("=== 多智能体流式对话系统（Tool Calling 模式）===")
    print("输入 'exit' 退出\n")

    messages = []

    while True:
        user_input = input("你：")

        messages.append({"role": "user", "content": user_input})

        if user_input.strip().lower() in ["exit", "quit"]:
            break

        print("智能体：", end="", flush=True)

        # 使用多个流模式同时监听主 Agent token + 工具自定义流
        assistant_reply = ""
        for mode, chunk in supervisor.stream(
            {"messages": messages},
            stream_mode=["messages", "custom"],# 见 https://docs.langchain.com/oss/python/langgraph/streaming#supported-stream-modes
        ):
            if mode == "messages":
                # 主 Agent token 流逐字打印
                if chunk[0].content:
                    assistant_reply += chunk[0].content
                    print(chunk[0].content, end="", flush=True)
            elif mode == "custom":
                # 工具 Agent 自定义流打印
                print(chunk, end="", flush=True)
                assistant_reply += str(chunk)
        messages.append({"role": "assistant", "content": assistant_reply})
        print("\n" + "-"*50)

if __name__ == "__main__":
    chat()

调试示例

=== 多智能体流式对话系统（Tool Calling 模式）===
输入 'exit' 退出

你：你是谁
智能体：我是你的助手，你可以叫我总管代理。我来这里是帮助你解决问题的，无论是数学计算、知识解释还是其他方面的问题，都可以找我哦！有什么我可以帮到你的？
--------------------------------------------------
你：你能帮我做啥
智能体：当然可以！我能帮你做的事情还挺多的：

- 如果你有数学题目需要解答，比如计算、方程求解之类的，我可以通过math_agent来帮助你。
- 如果你想了解一些知识性的内容，比如科学原理、历史事件、概念解释等，我可以用info_agent给你提供详细的解释。
- 对于一些简单的问题或者日常咨询，如果不需要特别的工具，我就可以直接回答你。

所以，不管你需要哪种帮助，尽管告诉我吧！
--------------------------------------------------
你：帮我计算 11*999-555
智能体：
 调用[math_agent] 开始计算：11*999-55511*999 -555 = 10434
 使用[math_agent] 计算完成。11*999 - 555 = 10434计算结果是 10434。这个计算里我们先做了乘法，然后进行了减法。如果还有其他数学题或者其他问题需要帮忙，随时告诉我！
--------------------------------------------------
你：帮我解释一下 什么是 multiagent 协作
智能体：调用[info_agent] 开始解释：什么是 multiagent 协作[info_agent] 解释完成。解释完成Multiagent协作是指在多智能体系统中，多个独立的智能体通过相互之间的沟通与合作，共同完成一个或一系列任务的过程。这里的智能体可以是软件程序、机器人或者是能够自主决策和行动的实体。

在Multiagent协作中，每个智能体都有自己的目标和能力，它们需要通过协调彼此的行为来达到整个系统的最优效果。这种协作方式广泛应用于各种领域，比如自动化生产、交通管理、游戏AI等场景中，它能够帮助解决复杂问题并提高效率。

如果你对这个话题感兴趣，想要了解更多细节或者有其他相关的问题，随时欢迎提问！
--------------------------------------------------
你：

Handoffs 模式（代理之间交接控制权）

三个不同的 Agent：

对话中会有接力逻辑：

• 用户开始和 Planner 对话；
• 当 Planner 需要订票 → 把控制权交给 FlightAgent；
• FlightAgent 订完票后 → 接回到 Planner；
• Planner 再交给 HotelAgent；
• 全程用户感觉是自然对话，但实际上是多个 Agent 接力进行。

代码

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

llm = ChatOpenAI(
        temperature=0.75,
        model="qwen-max",
        api_key=SecretStr(DASHSCOPE_API_KEY),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )

# === 定义各个智能体 ===
planner_agent = create_agent(
    model=llm,
    system_prompt=(
        "你是一个旅行规划师，根据用户需求制定旅行路线和时间表。"
        "如果用户的意愿是订票相关服务，你将返回且只能返回：“正在将任务转交订票服务”；"
        "如果用的意愿是寻找住宿地点，你将返回且只能返回：“正在将任务转交住宿服务”；"
    )
)

flight_agent = create_agent(
    model=llm,
    system_prompt=(
        "你是航班专家，可以帮助用户预订或推荐航班。"
        "在用与户确认航班机票等服务完成后，你将返回且只能返回：“感谢使用订票服务，正在转接回旅行规划师”；"
    )
)

hotel_agent = create_agent(
    model=llm,
    system_prompt=(
        "你是酒店专家，可以帮助用户寻找或预订住宿。"
        "在与用户确认酒店预订服务完成后，你将返回且只能返回：“感谢使用住宿服务，正在转接回旅行规划师”；"
    )
)

# === 初始化一个开始的智能体 ===
active_agent = planner_agent
messages = []


def chat():
    global active_agent

    print("=== 多智能体接力系统（旅行规划 Handoffs 模式） ===")
    print("输入 'exit' 退出\n")

    old_active_agent = active_agent

    while True:
        user_input = input("\n你：")
        if user_input.strip().lower() in ["exit", "quit"]:
            break

        messages.append({"role": "user", "content": user_input})

        print("\n智能体：", end="", flush=True)
        reply = ""

        for token, _ in active_agent.stream({"messages": messages}, stream_mode="messages"):
            if token.content:
                print(token.content, end="", flush=True)
                reply += token.content

        print("\n" + "-" * 50)
        messages.append({"role": "assistant", "content": reply})

        # === 交接控制权 ===
        if "正在将任务转交订票服务" in reply:
            print("转交控制权 ====>>>> 控制权 → FlightAgent（航班专家）")
            active_agent = flight_agent
        elif "正在将任务转交住宿服务" in reply:
            print("转交控制权 ====>>>> 控制权 → HotelAgent（酒店专家）")
            active_agent = hotel_agent
        elif "正在转接回旅行规划师" in reply:
            print("转交控制权 ====>>>> 控制权 → PlannerAgent（旅行规划师）")
            active_agent = planner_agent

        if old_active_agent != active_agent:
            reply = ""
            print(f"转交后 ==> 接收到任务开始处理==========>>>>>>>>>>")
            messages.append({"role": "user", "content": user_input})
            for token, _ in active_agent.stream({"messages": messages}, stream_mode="messages"):
                if token.content:
                    print(token.content, end="", flush=True)
                    reply += token.content
            old_active_agent = active_agent
            messages.append({"role": "assistant", "content": reply})


if __name__ == "__main__":
    chat()

调试示例

=== 多智能体接力系统（旅行规划 Handoffs 模式） ===
输入 'exit' 退出


你：你是谁，你能帮我做啥

智能体：我是你的旅行规划助手，可以帮助你制定旅行路线和时间表。如果你需要订票或寻找住宿地点，我也可以帮你转接到相应的服务。告诉我你的旅行目的地和偏好，我们就可以开始规划了！
--------------------------------------------------

你：我希望订一张去四川的机票，从南京出发，2025-11-06 出发 2人

智能体：正在将任务转交订票服务
--------------------------------------------------
转交控制权 ====>>>> 控制权 → FlightAgent（航班专家）
转交后 ==> 接收到任务开始处理==========>>>>>>>>>>
好的，我来帮您查找从南京到四川的航班。请问您有具体的到达城市吗？比如成都、绵阳等。另外，请问您对航班时间（如早上、下午或晚上）有任何偏好吗？还有，您需要经济舱还是商务舱呢？提供更多细节可以帮助我为您找到最合适的航班。
你：经济舱即可

智能体：好的，我将为您查找2025年11月6日从南京出发到成都的经济舱航班。以下是几个推荐的航班选项：

1. **航班号：MU8973**
   - **航空公司：东方航空**
   - **出发时间：08:00**
   - **到达时间：10:30**
   - **价格：¥1,200/人**

2. **航班号：CA4567**
   - **航空公司：中国国航**
   - **出发时间：10:30**
   - **到达时间：13:00**
   - **价格：¥1,300/人**

3. **航班号：3U8567**
   - **航空公司：四川航空**
   - **出发时间：14:00**
   - **到达时间：16:30**
   - **价格：¥1,100/人**

4. **航班号：CZ3456**
   - **航空公司：南方航空**
   - **出发时间：16:00**
   - **到达时间：18:30**
   - **价格：¥1,250/人**

请问您对以上哪个航班感兴趣？或者您有其他特殊需求吗？

如果您确认了航班，我将为您预订两张机票。
--------------------------------------------------

你：好的我知道了，我自己购买即可，谢谢

智能体：感谢使用订票服务，正在转接回旅行规划师
--------------------------------------------------
转交控制权 ====>>>> 控制权 → PlannerAgent（旅行规划师）
转交后 ==> 接收到任务开始处理==========>>>>>>>>>>
好的，如果您需要任何其他帮助，比如制定旅行路线、安排景点游览或者寻找当地的美食和活动，请随时告诉我。祝您旅途愉快！
你：