Multi-Agent Workflows梳理与实战

发表于 更新于 分类于 阅读次数：

前言

代码见learn-rag-langchain/multi-agent at main · zxj-2023/learn-rag-langchain

什么是多智能体

当我们谈论”多智能体”时，我们指的是由llm驱动的多个独立的agent以特定方式连接在一起。

每个agent可以拥有自己的提示、LLM、工具和其他自定义代码，以最佳方式与其他智能体协作。

这种思维方式非常适合用图来表示，就像 langgraph 所提供的那样。在这种方法中，每个智能体都是图中的一个节点，而它们之间的连接则表示为一条边。控制流由边管理，它们通过向图的状态中添加信息来进行通信。

多智能体架构梳理

langgraph给我们提供了几种多智能体架构

image-20250818163359517

Network: 每个智能体可以与其他所有智能体通信。任何智能体都可以决定下一步调用哪个其他智能体。

Multi-agent network

Supervisor：每个智能体与一个单一的监督者智能体通信。监督者智能体决定下一步应该调用哪个智能体。

代理监督者 — Agent Supervisor

Hierarchical: 你可以定义一个具有监督者监督者的多代理系统。这是监督者架构的泛化，并允许更复杂的控制流程。

层级代理团队 — Hierarchical Agent Teams

Custom multi-agent workflow: 每个代理只与代理子集通信。流程的部分是确定的，只有一些代理可以决定下一步调用哪些其他代理。

Agent Supervisor

在本教程中，你将构建一个包含两个代理的监督者系统——一个研究专家和一个数学专家。

环境

1
pip install -U langgraph langgraph-supervisor langchain-tavily "langchain[openai]"

1. 创建工作代理

首先，让我们创建我们的专业工作代理——研究代理和数学代理：

研究代理

对于网络搜索，我们将使用 TavilySearch 工具来自 langchain-tavily :

1
2
3
4
5
6
from langchain_tavily import TavilySearch

web_search = TavilySearch(max_results=3,tavily_api_key="tvly-dev-")
web_search_results = web_search.invoke("南京在哪")

print(web_search_results["results"][0]["content"])

为了创建单个工作代理，我们将使用 LangGraph 的预构建代理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm=ChatOpenAI(
    model="qwen3-235b-a22b-thinking-2507",
    api_key="sk-",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

research_agent = create_react_agent(
    model=llm,
    tools=[web_search],
    prompt=(
        "你是一个研究代理。\n\n指令：\n- 仅协助与研究相关的任务，不得进行任何数学计算\n- 完成任务后，直接向主管回复\n- 仅回复你的工作结果，不得包含任何其他文字。"
    ),
    name="research_agent",
)

让我们运行代理来验证它的行为是否符合预期。我们将使用 pretty_print_messages 辅助工具来美观地渲染流式代理输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from langchain_core.messages import convert_to_messages


def pretty_print_message(message, indent=False):
    """
    美化打印单条消息
    
    Args:
        message: 要打印的消息对象
        indent: 是否需要缩进打印
    """
    # 将消息转换为美观的HTML格式表示
    pretty_message = message.pretty_repr(html=True)
    if not indent:
        # 如果不需要缩进，直接打印
        print(pretty_message)
        return

    # 如果需要缩进，为每一行添加制表符前缀
    indented = "\n".join("\t" + c for c in pretty_message.split("\n"))
    print(indented)


def pretty_print_messages(update, last_message=False):
    """
    美化打印消息更新
    
    Args:
        update: 包含消息更新的数据结构
        last_message: 是否只打印最后一条消息
    """
    is_subgraph = False  # 标记是否为子图更新
    
    # 检查更新是否为元组格式（包含命名空间信息）
    if isinstance(update, tuple):
        ns, update = update
        # 如果命名空间为空，跳过父图更新的打印
        if len(ns) == 0:
            return

        # 提取图ID并打印子图更新信息
        graph_id = ns[-1].split(":")[0]
        print(f"来自子图 {graph_id} 的更新:")
        print("\n")
        is_subgraph = True

    # 遍历每个节点的更新
    for node_name, node_update in update.items():
        # 构造更新标签
        update_label = f"来自节点 {node_name} 的更新:"
        if is_subgraph:
            # 如果是子图，添加缩进
            update_label = "\t" + update_label

        print(update_label)
        print("\n")

        # 将节点更新中的消息转换为消息对象列表
        messages = convert_to_messages(node_update["messages"])
        # 如果只要求最后一条消息，则截取最后一条
        if last_message:
            messages = messages[-1:]

        # 打印每条消息
        for m in messages:
            pretty_print_message(m, indent=is_subgraph)
        print("\n")
1
2
3
4
for chunk in research_agent.stream(
    {"messages": [{"role": "user", "content": "南京在哪?"}]}
):
    pretty_print_messages(chunk)
数学代理

对于数学代理工具，我们将使用纯 Python 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def add(a: float, b: float):
    """将两个数字相加。"""
    return a + b


def multiply(a: float, b: float):
    """将两个数字相乘。"""
    return a * b


def divide(a: float, b: float):
    """将两个数字相除。"""
    return a / b


math_agent = create_react_agent(
    model=llm,
    tools=[add, multiply, divide],
    prompt=(
        "你是一个数学代理。\n\n"
        "指令：\n"
        "- 仅协助处理数学相关任务\n"
        "- 完成任务后，直接回复给主管\n"
        "- 仅回复你的工作结果，不要包含任何其他文字。"
    ),
    name="math_agent",
)

让我们运行数学代理：

1
2
3
4
for chunk in math_agent.stream(
    {"messages": [{"role": "user", "content": "what's (3 + 5) x 7"}]}
):
    pretty_print_messages(chunk)

2.创建监督者 langgraph-supervisor

为了实现我们的多智能体系统，我们将使用预构建的 langgraph-supervisor 库中的 create_supervisor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langgraph_supervisor import create_supervisor
from langchain.chat_models import init_chat_model

supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, math_agent],
    prompt=(
        "你是一个管理两个代理的主管：\n"
        "- 一个研究代理。将研究相关任务分配给这个代理\n"
        "- 一个数学代理。将数学相关任务分配给这个代理\n"
        "一次只分配工作给一个代理，不要并行调用代理。\n"
        "不要自己做任何工作。"
    ),
    add_handoff_back_messages=True,
    output_mode="full_history",
).compile()
1
2
3
from IPython.display import display, Image

display(Image(supervisor.get_graph().draw_mermaid_png()))
image-20250819113009108

现在让我们用一个需要两个代理的查询来运行它：

研究代理将查找必要的 GDP 信息；数学代理将执行除法以找到纽约州 GDP 的百分比，如所请求

3.从头创建监督者

现在让我们从头实现这个多智能体系统。我们需要：

  1. 设置主管如何与各个代理进行沟通
  2. 创建监督代理
  3. 将监督代理和工作代理组合成一个多代理图。
设置代理通信

我们需要定义一种方式，让监督代理能够与工作代理进行通信。在多代理架构中，实现这一功能的一种常见方法是使用handoffs，即一个代理将控制权交给另一个代理。交接允许你指定：

  • destination:要转移到的目标代理
  • payload:要传递给该智能体的信息

我们将通过handoff tools（转接工具）实现转接，并将这些工具交给监督代理：当监督代理调用这些工具时，它将控制权转交给工作代理，并将完整消息历史传递给该代理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command


def create_handoff_tool(*, agent_name: str, description: str | None = None):
    """
    创建一个“交接”工具函数，用于在 LangGraph 的 Supervisor-Worker 架构中
    把当前对话状态移交给指定名称的子 Agent。

    参数
    ----
    agent_name : str
        目标子 Agent 的名称，必须与 Supervisor 图中注册的节点名一致。
    description : str | None
        工具的描述文本。如果为 None，则使用默认描述 "Ask {agent_name} for help."。

    返回
    ----
    handoff_tool : Callable
        一个已用 @tool 装饰的函数，可直接注入到 Supervisor 的工具列表。
    """
    # 动态生成工具名，例如 agent_name="math_agent" -> "transfer_to_math_agent"
    name = f"transfer_to_{agent_name}"

    # 如果调用者没有提供描述，则使用默认描述
    description = description or f"Ask {agent_name} for help."

    # 用 LangGraph 的 @tool 装饰器注册工具
    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        """
        实际执行交接逻辑的工具函数。

        参数
        ----
        state : MessagesState
            当前对话状态，由 LangGraph 注入。
        tool_call_id : str
            本次工具调用的唯一 ID，由 LangGraph 注入。

        返回
        ----
        Command
            一个 LangGraph Command 对象，告诉框架：
            - goto=agent_name        : 跳转到哪个子 Agent
            - update                 : 更新后的状态
            - graph=Command.PARENT   : 在父图（Supervisor）作用域内执行
        """
        # 构造一条工具消息，记录交接动作
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }

        # 使用 Command 把对话状态连同新消息一起发送到目标 Agent
        return Command(
            goto=agent_name,
            update={**state, "messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )

    # 返回已装饰的工具函数，供 Supervisor 添加进 tools 列表
    return handoff_tool


# 创建研究代理的交接工具
assign_to_research_agent = create_handoff_tool(
    agent_name="research_agent",
    description="Assign task to a researcher agent.",
)

# 创建数学代理的交接工具
assign_to_math_agent = create_handoff_tool(
    agent_name="math_agent",
    description="Assign task to a math agent.",
)
创建监督代理

然后，我们使用刚刚定义的交接工具来创建监督代理。我们将使用预构建的 create_react_agent :

1
2
3
4
5
6
7
8
9
10
11
12
supervisor_agent = create_react_agent(
    model=llm,
    tools=[assign_to_research_agent, assign_to_math_agent],
    prompt=(
        "你是一个管理两个代理的主管：\n"
        "- 一个研究代理。将研究相关任务分配给这个代理\n"
        "- 一个数学代理。将数学相关任务分配给这个代理\n"
        "一次只分配工作给一个代理，不要并行调用代理。\n"
        "不要自己做任何工作。"
    ),
    name="supervisor",
)
创建多智能体图

将这些内容整合起来，让我们为我们的整体多代理系统创建一个图。我们将添加监督代理和各个代理作为子图节点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langgraph.graph import END

# 定义多代理主管图
supervisor = (
    StateGraph(MessagesState)
    # 注意：`destinations` 仅用于可视化，不影响运行时行为
    .add_node(supervisor_agent, destinations=("research_agent", "math_agent", END))
    .add_node(research_agent)
    .add_node(math_agent)
    .add_edge(START, "supervisor")
    # 总是返回到主管
    .add_edge("research_agent", "supervisor")
    .add_edge("math_agent", "supervisor")
    .compile()
)

在这个代码中，去 research_agent 和 math_agent 的条件边是通过工具调用实现的，而不是显式的条件边。

工作机制：

  1. 工具作为交接手段：

    • assign_to_research_agent 和 assign_to_math_agent 这两个工具被添加到 supervisor_agent 中
    • 当 supervisor_agent 决定需要某个代理帮助时，它会调用相应的工具

  2. 工具内部实现交接：

    1
    2
    3
    4
    5
    6
    def handoff_tool(...) -> Command:
        return Command(
            goto=agent_name,  # 这里指定了要跳转到哪个代理
            update={...},
            graph=Command.PARENT,
        )

  3. 隐式的条件边：

    • 当 supervisor_agent 调用 assign_to_research_agent 工具时 → 自动跳转到 research_agent
    • 当 supervisor_agent 调用 assign_to_math_agent 工具时 → 自动跳转到 math_agent

什么是 Command 机制

Command 机制是 LangGraph 提供的一种显式控制流程跳转的方式。它允许工具或节点直接指定下一步要执行什么操作，而不需要通过传统的条件边路由。

Command 的核心概念

1
2
3
4
5
6
7
from langgraph.types import Command

Command(
    goto=agent_name,           # 要跳转到的目标节点
    update=state_update,       # 要更新的状态
    graph=Command.PARENT      # 在哪个图中执行（父图/子图）
)

请注意，我们已经从工作代理添加了明确的边回到主管——这意味着它们保证会将控制权返回给主管。如果你希望代理直接响应用户（即，将系统转变为路由器），你可以移除这些边。

Multi-agent network

一个单一智能体通常可以使用单个领域内的一小批工具来有效运作，但即使使用像 gpt-4 这样强大的模型，使用多个工具时也可能效果不佳。

处理复杂任务的一种方法是采用“分而治之”的方法：为每个任务或领域创建一个专门的智能体，并将任务路由到正确的“专家”。这是一个多智能体网络架构的例子。

image-20250819150039818

这个多agent架构，就像多个agent进行讨论，所以也叫Multi Agent Collaboration，但是给我的感觉，比较混乱，agent直接的路由很难去定义，agent一多就搞不清楚了，所以这里也不实战了。

Hierarchical Agent Teams

对于某些应用，如果工作按层次分布，系统可能会更有效。你可以通过组合不同的子图，并创建一个顶层监督者以及中层监督者来实现这一点。

image-20250819151813264

使用预设的supervisor构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# 1. 定义研究团队的代理
@tool
def web_search(query: str) -> str:
    """执行网络搜索"""
    return f"搜索结果：关于'{query}'的最新信息..."

@tool
def analyze_data(data: str) -> str:
    """分析数据"""
    return f"数据分析结果：{data}的趋势显示..."

research_agent = create_react_agent(
    model=llm,
    tools=[web_search, analyze_data],
    prompt="你是一个研究专家，负责进行网络搜索和数据分析。",
    name="research_specialist"
)

# 2. 定义数学团队的代理
@tool
def calculate_statistics(numbers: list[float]) -> str:
    """计算统计值"""
    if not numbers:
        return "错误：数据列表为空"
    avg = sum(numbers) / len(numbers)
    return f"统计结果：平均值={avg:.2f}，数据点数量={len(numbers)}"

@tool
def solve_equation(equation: str) -> str:
    """解方程"""
    return f"方程 {equation} 的解为：x = 42"

math_agent = create_react_agent(
    model=llm,
    tools=[calculate_statistics, solve_equation],
    prompt="你是一个数学专家，负责统计计算和方程求解。",
    name="math_specialist"
)

# 3. 创建研究团队主管
research_supervisor = create_supervisor(
    model=llm,
    agents=[research_agent],
    prompt=(
        "你是研究团队的主管。\n"
        "你的团队有一个研究专家，负责网络搜索和数据分析。\n"
        "根据任务需求，将工作分配给研究专家。\n"
        "等待专家完成任务后，总结结果并报告给上级主管。"
    ),
    name="research_supervisor"
).compile(name="research_supervisor")

# 4. 创建数学团队主管
math_supervisor = create_supervisor(
    model=llm,
    agents=[math_agent],
    prompt=(
        "你是数学团队的主管。\n"
        "你的团队有一个数学专家，负责统计计算和方程求解。\n"
        "根据任务需求，将工作分配给数学专家。\n"
        "等待专家完成任务后，总结结果并报告给上级主管。"
    ),
    name="math_supervisor"
).compile(name="math_supervisor")

# 5. 创建顶层主管
top_supervisor = create_supervisor(
    model=llm,
    agents=[research_supervisor, math_supervisor],
    prompt=(
        "你是顶层主管，管理两个专业团队：\n"
        "- 研究团队：负责市场调研、数据分析等任务\n"
        "- 数学团队：负责统计计算、方程求解等任务\n"
        "根据任务的性质，将工作分配给相应的团队主管。\n"
        "等待团队完成任务后，整合所有结果并给出最终报告。"
    ),
    name="top_supervisor"
).compile(name="top_supervisor")

参考资料

LangGraph：多智能体工作流 — LangGraph: Multi-Agent Workflows

Overview