Agno 多智能体团队（Agent Teams）实战指南：从四种执行模式到生产级团队编排

Agno 多智能体团队（Agent Teams）实战指南：从四种执行模式到生产级团队编排

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

本篇指南聚焦 Agno 开源仓库中 cookbook/03_teams 目录下的多智能体团队（Agent Teams）完整实践体系：从 route/broadcast/tasks 等核心协调模式，到记忆、会话、流式输出、守卫（Guardrails）、多模态、人机协作与故障恢复等 27 个专题。读完你将掌握如何用Team与TeamMode构建从简单两人协作到嵌套团队、分布式 RAG 的生产级多智能体系统，并能在源码层面理解 leader 与成员之间的协调机制。

准备工作与环境要求

在运行任何 Team 示例之前，需要满足以下前提：

• 环境变量：通过direnv allow加载环境变量（例如OPENAI_API_KEY）。几乎所有示例都依赖 LLM API Key，运行前务必确认已正确注入。
• 解释器：建议使用仓库提供的虚拟环境解释器.venvs/demo/bin/python运行示例，避免依赖冲突。
• 外部服务：部分示例需要额外服务，例如 PostgreSQL、LanceDB、Infinity server 或 AgentOS 远程实例（remote instances）。具体依赖会在对应脚本的 docstring 中注明，运行前请逐一核对。

从仓库目录结构看，cookbook/03_teams 之下按主题划分为 27 个子目录（01_quickstart 至 27_result_offloading），每个目录内含对应的.py示例与配套 README/TEST_LOG，形成"快速上手 → 模式 → 纵深能力"的递进式学习路径。

快速上手：Team 的最小可用示例

01_quickstart 目录包含了团队工作流最核心的 12 个示例文件，其中 01_basic_coordination.py 是最小可用的起点：

from agno.agent import Agent from agno.models.openai import OpenAIResponses from agno.team import Team # 创建团队成员（Member） planner = Agent( name="Planner", role="You plan tasks and split work into clear, ordered steps.", model=OpenAIResponses(id="gpt-5-mini"), ) writer = Agent( name="Writer", role="You draft concise, readable summaries from the team discussion.", model=OpenAIResponses(id="gpt-5-mini"), ) # 创建团队 team = Team( model=OpenAIResponses(id="gpt-5-mini"), name="Planning Team", members=[planner, writer], instructions=[ "Coordinate with the two members to answer the user question.", "First plan the response, then generate a clear final summary.", ], markdown=True, show_members_responses=True, ) if __name__ == "__main__": team.print_response( "Create a three-step outline for launching a small coding side project.", stream=True, )

这段代码揭示了 Team 的四个关键概念：

1. 成员（members）：团队成员就是普通的Agent实例，通过members=[...]挂载到 Team 上，每个成员用role明确职责边界。
2. leader 模型：model参数配置的是 Team leader（协调者）所用的大模型，由它负责拆分任务、分派成员并汇总结果。
3. 指令（instructions）：指导 leader 如何协调成员的顶层策略，例如"先规划、再总结"。
4. 可视化：markdown=True启用 Markdown 格式输出，show_members_responses=True在终端展示每个成员的中间响应，便于调试。

同一目录下还包含：02_respond_directly_router_team.py（响应直通路由团队）、03_delegate_to_all_members.py（委派给全部成员）、04_respond_directly_with_history.py（带历史的直通响应）、05_team_history.py（团队历史）、06_history_of_members.py（成员历史）、07_share_member_interactions.py（共享成员交互）、08_concurrent_member_agents.py（并发成员）、broadcast_mode.py（广播模式）、09_caching.py（响应缓存）、nested_teams.py（嵌套团队）、task_mode.py（任务模式）。从文件顺序可以看出，官方设计了一条从"基础协调"到"路由/广播/任务/嵌套团队"的渐进学习曲线。

四种核心执行模式：TeamMode 源码解析

Team 的执行模式由TeamMode枚举控制，源码位于 libs/agno/agno/team/mode.py，定义如下：

class TeamMode(str, Enum): """Execution mode for a Team. Controls how the team leader coordinates work with member agents. """ coordinate = "coordinate" """Default supervisor pattern. Leader picks members, crafts tasks, synthesizes responses.""" route = "route" """Router pattern. Leader routes to a specialist and returns the member's response directly.""" broadcast = "broadcast" """Broadcast pattern. Leader delegates the same task to every member. ``arun`` runs the members concurrently; ``run`` runs them in sequence.""" tasks = "tasks" """Autonomous task-based execution. Leader decomposes goals into a shared task list, delegates tasks to members, and loops until all work is complete."""

四种模式的核心差异：

值得注意的是 broadcast 模式的并发行为：源码注释明确指出arun（异步）下成员并发运行，而run（同步）下成员按顺序执行。这意味着需要并行吞吐时应优先使用异步 API。

路由模式：多语言路由团队

02_respond_directly_router_team.py 演示了一个完整的路由团队：六个成员分别负责英语、西班牙语、日语、法语、德语和中文。团队通过mode=TeamMode.route声明路由模式，并用instructions约束路由逻辑：

multi_language_team = Team( name="Multi Language Team", model=OpenAIResponses(id="gpt-5-mini"), mode=TeamMode.route, members=[english_agent, spanish_agent, japanese_agent, french_agent, german_agent, chinese_agent], markdown=True, instructions=[ "You are a language router that directs questions to the appropriate language agent.", "If the user asks in a language whose agent is not a team member, respond in English with:", "'I can only answer in the following languages: ...'", "Always check the language of the user's input before routing to an agent.", ], show_members_responses=True, )

该示例同时演示了同步与异步两种调用方式：

# 同步调用 multi_language_team.print_response("How are you?", stream=True) # 异步调用 import asyncio async def run_async_router() -> None: await multi_language_team.aprint_response("你好吗？", stream=True) await multi_language_team.aprint_response("お元気ですか?", stream=True) asyncio.run(run_async_router())

路由模式下 leader 直接返回被选中成员的响应，因此适合"分诊-转派"类任务；当用户使用团队不支持的意大利语提问时，leader 会按指令回退到英语兜底提示。

广播模式：多方独立评审

broadcast_mode.py 将产品经理、工程师、设计师三个成员组成Broadcast Review Team，把"是否下月发布 beta 自动驾驶功能"这一决策同时交给三方独立评估：

broadcast_team = Team( name="Broadcast Review Team", members=[product_manager, engineer, designer], model=OpenAIResponses(id="gpt-5.2"), mode=TeamMode.broadcast, instructions=[ "Each member must independently evaluate the same request.", "Provide concise recommendations from your specialist perspective.", "Highlight tradeoffs and open risks clearly.", ], markdown=True, show_members_responses=True, )

广播模式与03_delegate_to_all_members.py（委派给全部成员）都属于"全员参与"类，区别在于 broadcast 由TeamMode直接驱动、结构更显式，适合评审、头脑风暴、多视角风险排查等场景。

任务模式：自主分解与执行

task_mode.py 演示TeamMode.tasks：leader 将"新 AI 功能上线清单"这类目标自动分解成带依赖关系的任务列表，再分派给 Researcher、Architect、Writer 三个成员并循环跟踪直到全部完成：

tasks_team = Team( name="Task Execution Team", members=[researcher, architect, writer], model=OpenAIResponses(id="gpt-5.2"), mode=TeamMode.tasks, instructions=[ "Break goals into clear tasks with dependencies before starting.", "Assign each task to the most appropriate member.", "Track task completion and surface blockers explicitly.", "Provide a final consolidated summary with completed tasks.", ], markdown=True, show_members_responses=True, )

从源码结构（mode.py）可以推断，tasks 模式是四种模式中最"自治"的一种：leader 不仅负责分派，还维护共享任务清单、检测阻塞并在全部任务完成后输出合并总结，适合项目规划、发布清单等长流程任务。

嵌套团队：团队作为成员

nested_teams.py 展示了一个高级能力——Team可以像Agent一样作为成员被挂载到另一个Team中。示例中Research Team（研究团队）和Writing Team（写作团队）作为Program Team的两个成员协同工作：

research_team = Team( name="Research Team", members=[research_agent, analysis_agent], model=OpenAIResponses(id="gpt-5.2"), instructions=[ "Collect relevant information and summarize evidence.", "Highlight key takeaways and uncertainties.", ], ) parent_team = Team( name="Program Team", members=[research_team, writing_team], # 子团队作为成员 model=OpenAIResponses(id="gpt-5.2"), instructions=[ "Coordinate nested teams to deliver a single coherent response.", "Ask Research Team for evidence first, then Writing Team for synthesis.", ], markdown=True, show_members_responses=True, )

嵌套团队的成员职责可以自上而下逐层细化，父团队只关心"先取证、再合成"的宏观编排，子团队内部自行协调细节。这为大型组织架构（例如按业务域分团队）提供了天然的建模方式。配套的 21_state 专题还专门讲解了嵌套团队之间的共享会话状态。

团队历史与会话持久化

01_quickstart/05_team_history.py 演示了如何让成员共享团队历史：为 Team 配置 SQLite 数据库（db=SqliteDb(...)），这是历史功能正常工作的硬性要求，然后通过add_team_history_to_members=True把用户与团队之间的全部交互发送给成员，并通过respond_directly=True与determine_input_for_members=False让输入直接透传给成员：

multi_lingual_q_and_a_team = Team( name="Multi Lingual Q and A Team", model=OpenAIResponses(id="gpt-5-mini"), members=[german_agent, spanish_agent], instructions=[ "You are a multi lingual Q and A team that can answer questions in English and Spanish...", ], db=SqliteDb(db_file="tmp/multi_lingual_q_and_a_team.db"), respond_directly=True, determine_input_for_members=False, # 输入直接发送给成员 add_team_history_to_members=True, # 团队历史共享给成员 ) # 使用固定 session_id 保持跨轮次记忆 session_id = f"conversation_{uuid4()}" multi_lingual_q_and_a_team.print_response( "Hallo, wie heißt du? Meine Name ist John.", stream=True, session_id=session_id, ) multi_lingual_q_and_a_team.print_response( "Cuéntame una historia de 2 oraciones usando mi nombre real.", stream=True, session_id=session_id, )

示例中第一轮用德语告诉团队"我叫 John"，第二轮用西班牙语请求"用我的真名编一个两句话的故事"——成员能成功回忆出名字，正是团队历史 + 固定session_id协同工作的效果。

围绕历史与持久化，Team还提供了丰富的底层配置（见 libs/agno/agno/team/team.py）：

• 历史注入：add_history_to_context（将聊天历史加入发给模型的 messages）、num_history_runs（包含的历史轮次数）、num_history_messages（包含的历史消息条数）、max_tool_calls_from_history（历史中允许携带的工具调用上限）；
• 会话存储：通过db参数（如SqliteDb）持久化会话，配合session_id精确定位会话；
• 结果离载（offloading）：offload_tool_results可将超过阈值（默认 16000 字符）的工具结果与成员回答以文件形式存储，仅在消息中保留带 result id 的短信封，避免上下文爆炸——这是 27_result_offloading 专题的核心机制。

07_session 专题进一步覆盖会话持久化选项、会话摘要与历史搜索；06_memory 则聚焦记忆管理器、Agentic Memory 与 LearningMachine；12_learning 覆盖团队学习模式（always、configured、实体记忆、会话规划、学到的知识、决策日志）。05_knowledge 讲解团队知识库、过滤器与自定义检索器。

工具、输入输出与流式响应

• 03_tools：自定义工具与工具钩子（tool hooks）模式。Tool 与 Hook 的底层机制可参考 libs/agno/agno/team/team.py 中的pre_hooks/post_hooks/on_tool_call_start/on_tool_call_end等字段——这些钩子在请求进入、响应返回以及每次工具调用的前后触发，是拦截与扩展团队行为的标准入口。
• 04_structured_input_output：结构化输入/输出 schema、覆盖（override）与流式输出。源码中对应的response_model、output_model、parser_model、structured_outputs、use_json_mode等字段用于约束响应为强类型结构。
• 08_streaming：响应流式输出与事件监控。Team的stream、stream_events、stream_member_events（是否流式输出成员事件，默认 True）、store_events、events_to_skip等字段共同控制流式行为与事件记录。

上下文、压缩与推理增强

• 09_context_management：上下文过滤、开场白（introduction）与少样本（few-shot）上下文。对应add_history_to_context、introduction、system_message等配置。
• 10_context_compression：工具结果压缩与压缩管理器（CompressionManager）。源码中compress_tool_results与compression_manager字段负责压缩工具调用结果以节省上下文窗口。
• 11_reasoning：多用途推理团队模式。源码中reasoning_model（必须是原生推理模型）与reasoning_agent字段用于开启团队级推理链。
• 15_distributed_rag：基于 PgVector/LanceDB + 重排序（reranking）的多成员分布式检索。该专题演示如何把检索拆给多个成员并行执行，再用重排序层合并最优结果。
• 16_search_coordination：跨成员协调的 RAG/搜索模式，是分布式检索之外的另一种"检索编排"形态。

运行控制、依赖与安全防护

• 13_hooks：输入前钩子（pre-hooks）、输出后钩子（post-hooks）与流式钩子（stream hooks），用于在团队运行的各生命周期节点注入自定义逻辑。
• 14_run_control：取消（cancellation）、重试（retries）、模型继承、远程团队与后台执行。源码中retries（重试次数，默认 0）、delay_between_retries（默认 1 秒）、exponential_backoff（指数退避）共同构成重试策略。
• 17_dependencies：运行时依赖（runtime dependencies）在上下文、工具与成员流程中的传递。
• 18_guardrails：提示注入（prompt-injection）防护、内容审核（moderation）与 PII（个人身份信息）保护。这是将团队推向生产环境的必备防线。
• 19_multimodal：音频、图像与视频工作流。Team与成员均支持多模态输入输出，配合 12_multimodal 的 Agent 级示例可形成完整参照。
• 20_human_in_the_loop：确认（confirmation）、外部执行（external execution）与"需要用户输入"（user-input-required）流程，用于在团队关键节点引入人工审批。

会话状态、指标与故障恢复

• 21_state：跨成员与嵌套团队共享会话状态（session_state）。
• 22_metrics：团队/会话/成员级别的指标检查（metrics inspection），用于评估运行质量与调试性能瓶颈。
• 23_checkpointing：团队运行的检查点（checkpointing）与崩溃恢复（crash recovery）。断点续跑是大规模、长时任务的关键保障。
• 23_remote_agents：将远程 Agent 作为团队成员，实现跨进程/跨机器的成员编排（常配合 AgentOS 远程实例使用）。
• 24_regenerate：重做团队运行的最后一次响应（regenerate last response）。
• 25_time_travel：将团队运行回退（rewind）到更早的时间点并从该处恢复执行。
• 26_fork_session：将整个团队会话分支（fork）为一个新会话，便于在不影响原会话的前提下做实验。
• 27_result_offloading：将成员答案与大型工具结果存储为文件，在 leader 的对话记录（transcript）中仅保留信封（envelope）引用。

缓存与性能优化

01_quickstart/09_caching.py 演示了"两层缓存"：既在成员层（researcher、writer的OpenAIResponses(id="gpt-5.2", cache_response=True)）启用响应缓存，也在 leader 层（Team 的model=OpenAIResponses(id="gpt-5.2", cache_response=True)）启用缓存：

content_team = Team( members=[researcher, writer], model=OpenAIResponses(id="gpt-5.2", cache_response=True), markdown=True, )

当收到相同或相似请求时，命中缓存的层可直接复用历史响应，显著降低延迟与 Token 消耗。对于重复查询量大的生产系统，缓存配置是性价比最高的优化手段之一。

学习路线建议

基于上述目录结构与源码实现，推荐按如下顺序学习：

1. 入门：跑通 01_basic_coordination.py，理解"成员 = Agent、协调者 = Team"的基本模型；
2. 掌握模式：依次对照 02_modes 下的 coordinate/route/broadcast/tasks 四类示例（含 tasks_stream.py），结合 mode.py 理解每种模式的调度语义，再尝试嵌套团队；
3. 打通状态层：学习团队历史（05_team_history.py）、会话（07）、记忆（06）与学习（12），让团队具备跨轮次记忆；
4. 生产化加固：按需接入工具（03）、守卫（18）、人机协作（20）、运行控制（14）、检查点（23）、时间旅行（25）与结果离载（27）；
5. 场景化实战：在分布式 RAG（15）、搜索协调（16）、多模态（19）等专题中选择与你业务最匹配的组合方案。

如需深入了解Team的全部配置字段（模式、成员、历史、存储、流式、重试、遥测等），可直接阅读 libs/agno/agno/team/team.py；各专题的.py示例均附有可独立运行的完整代码，是比 README 更精确的参考实现。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno