多智能体概述

本文档介绍如何在同一系统中定义、协作和运行多个智能体（Agent），多 Agent 系统的设计核心在于角色分工与通信机制。

通过合理的协作策略，可将复杂任务拆解为可控的子任务，从而提高整体稳定性与可扩展性。

多 Agents 系统构建

多智能体系统通常由若干具备独立职责的 Agent 组成。主 Agent 的常见类型包括：

自主决策 Agent：基于大语言模型的通用智能体，支持自然语言理解、推理与生成，可配置工具、记忆、知识库等功能
工作流 Agent：针对特定执行模式优化的 Agent，包括顺序执行、并行执行、循环执行等模式，适用于不同的任务编排场景

自主决策 Agent

定义：自主决策 Agent 是 VeADK 的核心智能体，基于大语言模型构建，支持完整的智能对话、工具调用、记忆管理等功能。它可以根据配置自动调用子 Agent 或工具来完成复杂任务。

样例：下面给出一个生活建议 Agent 来演示说明自主决策 Agent 的主要交互方法。其中，主 Agent 负责接收用户输入，子 Agents 分别负责获取天气及给出穿衣建议。自主决策型 Agent 调用示意图如下所示：主 Agent 根据用户提示场景，首先调用了天气 Agent 来获取天气情况，随后调用了穿衣建议 Agent 来给出穿衣建议。

运行前，通过环境变量提供模型 API Key（VeADK 默认读取 MODEL_AGENT_API_KEY）：

export MODEL_AGENT_API_KEY="<你的方舟 API Key>"

llm_agent.py

import asyncio

from veadk import Agent, Runner
from veadk.tools.demo_tools import get_city_weather

weather_reporter = Agent(
    name="weather_reporter",
    description="A weather reporter agent to report the weather.",
    tools=[get_city_weather],
)

suggester = Agent(
    name="suggester",
    description="A suggester agent that can give some clothing suggestions according to a city's weather.",
    instruction=(
        "Provide clothing suggestions based on weather temperature: "
        "wear a coat when below 15°C, long sleeves when 15-25°C, "
        "short sleeves when above 25°C."
    ),
)

root_agent = Agent(
    name="planner",
    description="A planner that can generate a suggestion according to a city's weather.",
    instruction=(
        "Invoke weather reporter agent first to get the weather, "
        "then invoke suggester agent to get the suggestion. "
        "Return the final response to user."
    ),
    sub_agents=[weather_reporter, suggester],
)

if __name__ == "__main__":
    runner = Runner(root_agent)
    response = asyncio.run(runner.run("北京穿衣建议"))
    print(response)

运行结果：

运行结果

工作流 Agent

定义：工作流（Workflow）Agent 是负责调度、调用与结果整合的控制型智能体。它自身不具备语言模型能力，而是通过用户定义调用子 Agent 工作流类别，调用子基础 Agents 来完成整体任务。在 VeADK 中，工作流型 Agent 主要分为三类：

顺序型 Agent（Sequential Agent）：串行依次执行的多个智能体的流程
循环型 Agent（Loop Agent）：循环执行多个智能体的流程，直到满足某个特定条件退出
并行型 Agent（Parallel Agent）：可并行执行的多个智能体的流程

以下内容分别说明这三类多 Agent 主要构建方式。

顺序型 Agent

样例：下面给出一个文案生成 Agent 来演示顺序型 Agent 的执行流程。我们实现 3 个 Agents 来执行“打招呼—再见”工作流。调用示意图如下所示，SequentialAgent 按照用户定义的顺序，依次调用 greeting_agent 和 goodbye_agent，每个子 Agent 完成自己的任务后，将结果传递给下一个 Agent。

examples/agent/agents/seq_agent.py

import asyncio

from veadk import Agent, Runner
from veadk.agents.sequential_agent import SequentialAgent

greeting_agent = Agent(
    name="greeting_agent",
    description="A friendly agent that greets the user.",
    instruction="Greet the user warmly.",
)

goodbye_agent = Agent(
    name="goodbye_agent",
    description="A polite agent that says goodbye to the user.",
    instruction="Say goodbye to the user politely.",
)

root_agent = SequentialAgent(sub_agents=[greeting_agent, goodbye_agent])

if __name__ == "__main__":
    runner = Runner(root_agent)
    response = asyncio.run(runner.run("你好"))
    print(response)

运行结果：

运行结果

循环型 Agent

定义：循环型 Agent 会重复执行子 Agents，直到满足特定条件才退出循环。适用于需要迭代优化、多轮对话或条件判断的复杂任务。

样例：下面给出一个诗歌创作 Agent 来演示循环型 Agent 的执行流程。我们实现 2 个 Agents 来执行“计划-执行”循环工作流，并且设置最大 3 次循环次数：

计划 Agent：根据用户目标制定下一步行动计划
执行 Agent：执行计划并检查结果，完成后调用退出函数

循环型 Agent 调用示意图如下所示，LoopAgent 会循环调用 planner_agent 和 executor_agent，直到 executor_agent 检测到任务完成并调用退出函数。

examples/agent/agents/loop_agent.py

import asyncio

from google.adk.tools.tool_context import ToolContext
from veadk import Agent, Runner
from veadk.agents.loop_agent import LoopAgent


def exit_loop(tool_context: ToolContext):
    print(f"  [Tool Call] exit_loop triggered by {tool_context.agent_name}")
    tool_context.actions.escalate = True
    return {}


planner_agent = Agent(
    name="planner_agent",
    description="Decomposes a complex task into smaller actionable steps.",
    instruction=(
        "Given the user's goal and current progress, decide the NEXT step to take. You don't need to execute the step, just describe it clearly. "
        "If all steps are done, respond with 'TASK COMPLETE'."
    ),
)

executor_agent = Agent(
    name="executor_agent",
    description="Executes a given step and returns the result.",
    instruction="Execute the provided step and describe what was done or what result was obtained. If you received 'TASK COMPLETE', you must call the 'exit_loop' function. Do not output any text.",
    tools=[exit_loop],
)

root_agent = LoopAgent(
    sub_agents=[planner_agent, executor_agent],
    max_iterations=3,  # Limit the number of loops to prevent infinite loops
)

if __name__ == "__main__":
    runner = Runner(root_agent)
    response = asyncio.run(runner.run("用中文帮我写一首三行的小诗，主题是秋天"))
    print(response)

运行结果：

运行结果

并行型 Agent

定义：并行型 Agent 会同时执行多个子 Agents，适用于可以独立处理的任务，如优缺点分析、多角度评估等场景，能够显著提高处理效率。

样例：下面给出一个优缺点分析 Agent 来演示并行型 Agent 的执行流程。我们实现 2 个 Agents 来并行分析“LLM-as-a-Judge 评测模式”：

优点分析 Agent：识别和阐述该模式的优势
缺点分析 Agent：识别和阐述该模式的劣势

调用示意图如下所示，ParallelAgent 会同时调用 pros_agent 和 cons_agent，两个子 Agent 独立运行，最后将结果汇总返回。

examples/agent/agents/parallel_agent.py

import asyncio

from veadk import Agent, Runner
from veadk.agents.parallel_agent import ParallelAgent

pros_agent = Agent(
    name="pros_agent",
    description="An expert that identifies the advantages of a topic.",
    instruction="List and explain the positive aspects or advantages of the given topic.",
)

cons_agent = Agent(
    name="cons_agent",
    description="An expert that identifies the disadvantages of a topic.",
    instruction="List and explain the negative aspects or disadvantages of the given topic.",
)

root_agent = ParallelAgent(sub_agents=[pros_agent, cons_agent])

if __name__ == "__main__":
    runner = Runner(root_agent)
    response = asyncio.run(runner.run("请分析 LLM-as-a-Judge 这种评测模式的优劣"))
    print(response)

运行结果：

运行结果

多智能体概述

多 Agents 系统构建

自主决策 Agent

工作流 Agent

顺序型 Agent

循环型 Agent

并行型 Agent

常见问题

本页导航

多智能体概述

多 Agents 系统构建

自主决策 Agent

工作流 Agent

顺序型 Agent

循环型 Agent

并行型 Agent

常见问题

多 Agent 交互常见问题及解决方案

本页导航