用 Agno Team 编排多智能体协同搜索：从共享知识库 RAG 到分布式推理检索实战

本篇技术指南以 cookbook/03_teams/16_search_coordination 的三个可运行示例为核心，讲解如何在 Agno 中利用 Team 将"检索—分析—推理—综合"拆解为分工明确的成员 Agent，从而构建比单 Agent RAG 更稳健、可追溯、适合多源知识库的协同检索系统。读完你不仅能直接运行目录下的三个示例，还能掌握 Knowledge + LanceDB 混合检索、Cohere/Infinity 重排、ReasoningTools 推理工具链以及 Team 协调模式的底层实现关系。

目录概览：三种搜索协调形态

cookbook/03_teams/16_search_coordination 目录是 Agno cookbook 中专门演示"搜索协调（search coordination）"的模块，共三个递进式示例与一份测试日志：

文件	主题	核心分工
01_coordinated_agentic_rag.py	协调式 Agentic RAG	搜索 → 内容分析 → 综合成稿（检索管线）
02_coordinated_reasoning_rag.py	协调式推理 RAG	信息收集 → 推理分析 → 证据评估 → 应答协调（推理管线）
03_distributed_infinity_search.py	分布式 Infinity 检索	主检索 + 次检索（双知识库） → 交叉验证 → 结果排序（分布式检索）
TEST_LOG.md	验证记录	展示三个示例的 cookbook 模式检查与运行诊断

三个示例共享同一套架构骨架：先构造带向量库 + Embedder + Reranker 的 Knowledge 作为成员共享的知识底座，再定义多个各司其职的成员 Agent，最后把它们交给一个 Team 统一协调，向团队整体提问即可获得带引用、经过推理与交叉验证的最终回答。差异在于分工方式从"纯检索-综合"逐步升级到"检索 + 分布式推理"与"双库分布式检索 + 高性能重排"。

运行前置条件

按照目录 README.md 的说明，运行前需要：

• 通过 direnv allow 加载环境变量，例如 OPENAI_API_KEY；
• 使用项目虚拟环境 python（README 中写作 .venvs/demo/bin/python）来运行示例；
• 部分示例需要额外服务或依赖，示例文件 docstring 与 TEST_LOG.md 中均已注明。具体而言：
  • 前两个示例依赖 cohere 包（CohereEmbedder / CohereReranker）；TEST_LOG.md 记录的三次运行失败原因均为 ModuleNotFoundError: No module named 'cohere'，源码 embedder/cohere.py 会抛出 "cohere not installed. Please install using pip install cohere."，因此先执行 pip install cohere；
  • 第三个示例额外需要一个本地 Infinity reranking 服务（见下文"分布式 Infinity 检索"一节），并需要安装 infinity-emb[all]；
  • 所有示例都依赖 LanceDB 向量库以及模型 SDK（OpenAI 等）。

另外，.py 示例中引用的模型标识（如 gpt-5.2）、Embedder（embed-v4.0）与 Reranker（rerank-v3.5、BAAI/bge-reranker-base）是仓库编写示例时的取值，实际运行时请以你所用服务当前可用型号为准。

通用构建基石：Knowledge、混合检索与重排

三个示例的"知识底座"写法几乎完全一致，理解它就理解了整个目录的公共设施。

组装共享知识库

以示例一为例：

from agno.knowledge.embedder.cohere import CohereEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.reranker.cohere import CohereReranker
from agno.vectordb.lancedb import LanceDb, SearchType

knowledge = Knowledge(
    vector_db=LanceDb(
        uri="tmp/lancedb",
        table_name="agno_docs_team",
        search_type=SearchType.hybrid,
        embedder=CohereEmbedder(id="embed-v4.0"),
        reranker=CohereReranker(model="rerank-v3.5"),
    ),
)

关键点：

• Knowledge 是 Agno 中统一的知识容器（实现在 libs/agno/agno/knowledge/knowledge.py），通过注入 vector_db 与文档存储解耦后端；
• LanceDb 作为向量数据库，uri 指向本地目录 tmp/lancedb，table_name 用于区分不同主题的表——注意示例二、三分别使用了 agno_docs_reasoning_team、agno_docs_primary、agno_docs_secondary 等不同表名，保证多个示例/多个知识库在同一 uri 下互不污染；
• search_type=SearchType.hybrid 表示混合检索。SearchType 定义于 libs/agno/agno/vectordb/search.py，可取 vector（纯向量相似度）、keyword（纯关键词/全文检索）、hybrid（两者融合）三档；
• embedder 负责把文本片段编码成稠密向量（示例用 Cohere embed-v4.0），reranker 负责对召回结果做二次精排（示例用 Cohere rerank-v3.5）。

灌入文档数据

knowledge.insert_many(urls=["https://docs.agno.com/agents/overview.md"])

insert_many 是 Knowledge 提供的数据导入接口（源码见 knowledge.py），可接收 URL 列表、本地路径或文本片段，导入时会对文档分块、embed 并写入向量库。示例二、三中给出的 async 版本调用 await knowledge.ainsert_many(urls=[...])，与 Team 的 aprint_response 异步接口一一对应。写入位置 tmp/lancedb 为仓库示例约定，如需持久化数据可改成任意可写路径。

文档源 URL 的可替换性

示例把 docs.agno.com 的 Agent 文档作为测试语料，直接运行需联网抓取。在你自己项目中使用时，可以把该 URL 替换为待检索的任意文档 URL、本地文件目录或代码片段，Knowledge 的分块与向量化流程不变。

Team 协调编排：参数与模式背后的实现

示例中每次调用 Team.print_response(query, stream=True) 时，协调动作由 Team 类驱动。通过 libs/agno/agno/team/team.py 源码可以确认如下关键字段：

• members：团队成员，类型为 List[Union[Agent, "Team"]]，即成员本身也可以嵌套子 Team；
• model：团队"领导者"所用模型，负责理解任务、向成员派活并汇总结果；
• instructions：团队级系统指令，用于约束协作流程顺序；
• show_members_responses：为 True 时把每个成员 Agent 的完整响应展示在最终输出流里，便于观察"搜索到了什么、分析得如何"，是排查检索质量的重要开关；
• markdown：要求领导者以 Markdown 格式化最终回答。

而编排到底以何种方式发生，取决于 libs/agno/agno/team/mode.py 中定义的 TeamMode 枚举。三个示例没有显式传 mode，因此走的是默认的 coordinate 模式——按 mode.py 源码注释，这是"默认 supervisor 模式：Leader 挑选成员、定制任务、综合各成员响应"，它天然适合本目录"先检索、再分析、后综合"的流程化诉求。如果只想把任务路由给某一个专家并原样返回其回答，可改用 route；若希望 Leader 持续自主派活直至工作完成，可选用 tasks 模式并配合 max_iterations。

把这段源码逻辑与示例一一对照，你会发现 instructions 里写的"Knowledge Searcher 先检索、Content Analyzer 再分析、Response Synthesizer 最后成稿"本质上是在告诉 Leader 按什么顺序调用哪些成员，最终协作质量既依赖成员各自的 role、instructions，也依赖 Leader 的调度。

示例一：协调式 Agentic RAG

01_coordinated_agentic_rag.py 演示"协调式 Agentic RAG"，把传统的"单 Agent 一次检索+生成"改造成三段式流水线，成员设计如下：

成员	role（职责声明）	挂载能力	承担阶段
Knowledge Searcher	从知识库检索相关信息	knowledge + search_knowledge=True	召回：全面检索、输出带上下文的详细结果
Content Analyzer	分析并综合检索内容	纯 LLM	分析：抽取关键概念、关系，识别信息缺口
Response Synthesizer	生成带引用的最终完整回答	纯 LLM	成稿：汇总结论并附上来源引用

三个成员都没有自建工具，唯一能"触碰知识库"的是 Knowledge Searcher——它通过 knowledge=knowledge, search_knowledge=True 获得知识库检索工具，这一步正是单 Agent Agent(search_knowledge=True) 用法的团队化复用。其余成员只处理文本，配合 instructions 显式声明各自只做一件事，让最终回答带引用、可追溯且不容易被中间噪声带偏。

团队与运行代码：

coordinated_rag_team = Team(
    name="Coordinated RAG Team",
    model=OpenAIResponses(id="gpt-5.2"),
    members=[knowledge_searcher, content_analyzer, response_synthesizer],
    instructions=[...],   # 显式约定三段式协作顺序
    show_members_responses=True,
    markdown=True,
)

query = "What are Agents and how do they work with tools and knowledge?"
coordinated_rag_team.print_response(query, stream=True)

该设计把"检索质量"与"回答质量"解耦成独立可替换的环节：检索效果不好时只需优化 Searcher 的检索策略或换更强的 Reranker，而无需改动成稿逻辑。

示例二：协调式推理 RAG：ReasoningTools 与异步调用

02_coordinated_reasoning_rag.py 在检索流程之上叠加了显式推理管线，且成员数量增至四个：

成员	挂载能力	承担阶段
Information Gatherer	knowledge + search_knowledge=True + ReasoningTools	先用推理工具规划检索策略，再全面收集信息与证据
Reasoning Analyst	ReasoningTools	用结构化推理（演绎/归纳）拆解复杂主题、梳理逻辑关系
Evidence Evaluator	ReasoningTools	评估证据质量、识别信息与推理缺口、衡量逻辑连接强度
Response Coordinator	ReasoningTools	汇总全员贡献，输出逻辑一致、带引用与透明推理链的应答

与示例一最显著的区别是每个成员都注入了 agno.tools.reasoning 中的 ReasoningTools(add_instructions=True)。从 libs/agno/agno/tools/reasoning.py 的签名可以看到，ReasoningTools 接收 add_instructions 开关（为真时把推理相关指引注入提示词），它把"规划-推理-评估"这类元能力以工具形式暴露给成员，使推理过程可观察、可结构化。团队级 instructions 也据此约定严格顺序：Gatherer → Analyst → Evaluator → Coordinator，并要求"All agents should use reasoning tools to structure their contributions"。

该示例同时演示了同步与异步两套入口：

async def async_reasoning_demo() -> None:
    await knowledge.ainsert_many(urls=["https://docs.agno.com/agents/overview.md"])
    await coordinated_reasoning_team.aprint_response(
        query, stream=True, show_full_reasoning=True,
    )

def sync_reasoning_demo() -> None:
    knowledge.insert_many(urls=["https://docs.agno.com/agents/overview.md"])
    coordinated_reasoning_team.print_response(
        query, stream=True, show_full_reasoning=True,
    )

注意 aprint_response(..., show_full_reasoning=True)：当成员使用推理工具时，该参数让推理链在流式输出中完整可见，适合需要"过程可审计"的知识问答/决策场景。主入口默认执行同步版本，若需异步可将 asyncio.run(async_reasoning_demo()) 的注释打开。

示例三：分布式 Infinity 检索

03_distributed_infinity_search.py 是目录中"分布式"程度最高的示例，架构上有两个显著特征。

特征一：双知识库并行检索。 示例构造了 knowledge_primary 与 knowledge_secondary 两个独立的 Knowledge（表名分别为 agno_docs_primary、agno_docs_secondary），分别由 Primary Searcher 与 Secondary Searcher 持有。两者任务互补：Primary 做宽口径综合检索，Secondary 专注查询中的边角细节与专精信息，从而模拟"不同检索单元分布式并行"的效果。运行前两者都会灌入同一份源文档（insert_many / ainsert_many 各调用一次），但可自由替换为不同语料以模拟跨库检索。

特征二：Infinity 本地重排服务。 两个知识库的 Reranker 都指向自建服务：

reranker=InfinityReranker(
    base_url="http://localhost:7997/rerank",
    model="BAAI/bge-reranker-base",
),

示例还给出了服务启动与排错命令（位于源码 if __name__ == "__main__" 的异常分支中）：

pip install 'infinity-emb[all]'
infinity_emb v2 --model-id BAAI/bge-reranker-base --port 7997

这意味着运行前必须先启动 Infinity server，否则示例捕获到异常后会打印上述提示。该设计展示了一种不依赖第三方云重排 API的本地化高性能排序方案：检索阶段仍用 Cohere Embedder 编码，精排阶段却把请求转发给本地 infinity_emb 服务完成。

团队由四个成员组成：

成员	挂载能力	承担阶段
Primary Searcher	knowledge_primary + 检索	宽口径主检索，用 Infinity 重排保证高相关结果优先
Secondary Searcher	knowledge_secondary + 检索	定向检索边缘场景与专精细节，与主检索互补
Cross-Reference Validator	纯 LLM	交叉比对两份检索结果，识别一致性与矛盾
Result Synthesizer	纯 LLM	汇总全部成员结果，按相关性与可靠性排序、标注来源与置信度

同样提供 sync_distributed_search() 与 async_distributed_search() 双入口，默认运行同步版本；try/except 结构把 Infinity 服务的安装提示直接内嵌到了运行时错误处理中，属于可直接照搬到生产脚手架的错误提示写法。

三个示例对比与选型建议

维度	示例一（Agentic RAG）	示例二（Reasoning RAG）	示例三（Infinity 分布式）
分工轴	检索→分析→成稿	检索→推理→证据→成稿	双库检索→交叉验证→排序成稿
成员数	3	4	4
知识库数量	1	1	2（可扩展为更多）
推理工具	无	ReasoningTools（全员）	无
重排器	Cohere rerank-v3.5（云 API）	Cohere rerank-v3.5（云 API）	InfinityReranker + bge-reranker-base（本地服务）
异步入口	无（仅同步）	ainsert_many + aprint_response	ainsert_many + aprint_response
典型场景	多轮知识问答、带引用的客服/文档问答	需要推理可审计的复杂决策问答	跨语料/边缘场景查询、对延迟与数据本地性敏感的环境

选型上，简单问答优先示例一；问题需要逻辑推演且过程需留痕时上示例二；当语料分散、追求检索精度或不能把数据交给外部重排服务时，参考示例三。

验证记录与常见运行问题

目录内 TEST_LOG.md 记录了 2026-02-15 的一次验证：三个示例的 Pattern Check 均为 PASS（cookbook 规范检查通过），但实际执行阶段三个示例都因缺少 cohere 依赖而失败，报错均指向：

File ".../libs/agno/agno/knowledge/embedder/cohere.py", line 9, in <module>
    from cohere import AsyncClient as AsyncCohereClient
ModuleNotFoundError: No module named 'cohere'

对照 embedder/cohere.py 的源码可知其导入策略：尝试从 cohere 导入异步客户端，失败后抛出带安装指引的 ImportError。这说明：

1. 依赖缺失时 Agno 给出的错误信息本身就是可操作的安装提示，直接 pip install cohere 即可；
2. 由于三个示例共用了 Cohere Embedder（而示例三还额外需要本地 Infinity 服务），依赖安装是比代码逻辑更常见的第一道门槛；
3. 验证日志同时提示 TEST_LOG.md 自身仍含 OpenAIChat 引用（即文档与最新 OpenAIResponses 用法存在轻微不一致），阅读示例时应以各 .py 文件实际导入为准。

深入当前仓库继续探索

• 团队编排参数全貌与 show_members_responses 等字段的默认值：libs/agno/agno/team/team.py
• coordinate / route / tasks 三种领导模式语义：libs/agno/agno/team/mode.py
• 知识库容器与 insert_many / ainsert_many 数据导入接口：libs/agno/agno/knowledge/knowledge.py
• SearchType 枚举与 hybrid 混合检索定义：libs/agno/agno/vectordb/search.py
• 推理工具集 ReasoningTools 与 add_instructions 参数：libs/agno/agno/tools/reasoning.py
• 三种编排方案的完整可运行源码，见 cookbook/03_teams/16_search_coordination 目录下的 01_coordinated_agentic_rag.py、02_coordinated_reasoning_rag.py、03_distributed_infinity_search.py