在给 AI Agent 设计“联网搜索”能力时我踩过不少坑单挂一个搜索 API结果太少且不够精准同时调多个数据源又面临接口格式不统一、结果重复、延迟不可控的问题。更麻烦的是企业内部的知识库、工单系统、GitHub Issues 经常无法被同一个检索入口覆盖。后来我把思路从“找一个万能搜索接口”换成“做一层联邦搜索聚合层”问题才真正开始被解决。这篇文章会围绕面向 AI Agent 的 Federated Search联邦搜索展开讲清楚它的核心概念、架构设计、代码实现和落地避坑点并提供一套可直接运行的最小示例。1. 为什么 AI Agent 需要联邦搜索1.1 AI Agent 的搜索困境LLM 驱动的 AI Agent 和普通聊天机器人最大的区别在于“自主行动能力”。Agent 为了完成一个任务需要自己去查资料、读代码、看文档、调用工具。但这里有一个很现实的矛盾模型的参数知识是静态的而 Agent 处理的任务往往是动态的。我在实际项目中看到过三类高频问题知识截止问题模型训练好的时间点之后发生的事情Agent 一概不知道比如某开源项目今天新增了一个接口Agent 无法从参数知识中获取。数据孤岛问题企业内部有 Wiki、GitHub Enterprise、数据库、工单系统每一个系统都有自己的搜索语法和权限模型Agent 难以在多个系统之间自由穿梭。工具碎片化问题为了让 Agent 能搜索开发者在代码里堆了一堆 tool 调用函数每个函数去请求不同的接口再把结果拼成一长串文本丢给模型。结果经常是格式混乱、重复信息多Agent 不知道该信哪条数据。这三个问题叠加在一起会直接导致 Agent 的“检索质量”下降进而影响最终任务的成功率。1.2 什么是 Federated SearchFederated Search 并不是一个全新概念。在传统信息检索领域它其实出现得很早也叫分布式信息检索或元搜索。它解决的问题是当信息分散在多个异构数据源中时用一个统一入口发起查询分别向各个数据源检索然后把结果合并、去重、排序后返回给用户。在 AI Agent 的场景下Federated Search 的含义可以进一步具体化它是一层位于 Agent 与多数据源之间的检索中间层通常包含适配器、查询分发、结果融合、缓存、权限校验和审计。Agent 只需要调用一个统一的搜索函数或 HTTP 接口联邦层负责把查询分发到多个不同类型的来源并把结果整理成结构化的、按相关度排序的数据再交给 Agent 使用。换句话说Federated Search 让“一个 Agent 面对多个搜索后端”变成了“一个 Agent 面对一个统一检索入口”。1.3 联邦搜索与集中式搜索、RAG 的区别很多同学会把联邦搜索和集中式搜索引擎、RAG检索增强生成混在一起这里做一个简单区分。检索方式核心特征典型场景集中式搜索先建索引再统一查询如 Elasticsearch对内部已有文档库做统一检索向量检索 RAG将文档切片后向量化按语义相似度召回大模型回答前补充领域知识Federated Search不统一建索引而是同时查询多个已有检索系统Agent 需要同时查 Web、GitHub、内部系统Federated Search 不关心每个后端用什么引擎、怎么建索引它只关心能不能通过对方已提供的能力完成检索并把结果统一回来。这与 RAG 可以互补在 RAG 场景里联邦搜索可以作为召回阶段的一个“多路召回器”把不同来源的结果喂给重排模型。2. 面向 AI Agent 的联邦搜索整体架构2.1 架构分层从工程实现角度看联邦搜索可以拆成下面几层数据源层真正存放数据的系统比如 Web 搜索 API、GitHub Issues、Jira、文件系统、数据库、内部 Wiki。适配器层把不同数据源的协议、鉴权方式、返回格式统一成内部 Schema。每接入一个新数据源本质上是写一个新适配器。联邦调度层负责接收 Agent 的查询请求解析出查询词、过滤条件、期望返回数量然后并发调用多个适配器并处理超时、重试、降级。结果融合层对多个数据源返回的结果做归一化、去重、打分、排序最终输出一个有序列表。Agent 接口层以函数调用、HTTP API 或 MCP 协议的方式暴露给 Agent让 Agent 可以自然地把联邦搜索当作一个工具使用。横切关注点包括配置管理、日志追踪、权限控制、缓存、审计等。2.2 一次联邦查询的完整流程一次典型的联邦查询流程可以理解为下面几个步骤Agent 收到用户问题判断需要外部知识。Agent 调用联邦搜索入口传入 query例如“FastAPI 如何实现流式响应”。调度层根据配置决定路由到哪些数据源可以全量广播也可以按规则选择。多个适配器并行查询各自后端每个适配器负责做鉴权和参数转换。各适配器在约定时间内返回统一格式的结果超时源直接降级跳过。结果融合模块把所有结果合并计算相关度分数返回 top-k 给 Agent。Agent 把结构化的搜索结果转换成上下文交给 LLM 生成最终回答。这整个流程对 Agent 来说只需要一次工具调用真正复杂的过程都被封装在联邦层内部。2.3 核心设计与挑战联邦搜索看起来“不就是并发调几个接口吗”但真正落地时要面对下面这些挑战异构性每个数据源返回的字段完全不同有的有标题有的只有正文有的带 stars有的带作者。可靠性一次查询依赖多个外部系统任何一个系统变慢都可能拖慢整个 Agent 的响应。相关性融合Web 搜索给出的分数和 GitHub 搜索给出的分数并不直接可比不能简单相加取 max。权限边界内部数据和外部公开数据不能混在同一个权限模型下否则容易出现越权暴露。成本控制每多一个数据源就多一笔 API 调用费用尤其是外部搜索 API必须通过缓存和路由策略控制。理解了这些问题再去写代码的时候就会更有方向。3. 环境准备与项目结构3.1 运行环境与依赖本文示例使用 Python 3.10主要原因是用到了asyncio和类型标注的现代特性。外部依赖尽量少方便你理解和二次开发。核心依赖如下依赖库用途fastapi提供 HTTP 接口方便 Agent 调用uvicorn运行 FastAPI 服务httpx异步请求外部 APIpydantic定义统一 Schemapython-dotenv读取环境变量配置如果只做本地实验可以把 FastAPI 去掉直接在 Python 脚本里调用联邦搜索类核心逻辑完全一致。下面依然按项目工程的方式拆分文件方便你扩展。版本方面这里不绑定某一个精确版本建议在安装后按当前环境调整。以我常用的环境为例FastAPI 0.100、httpx 0.24 都能稳定运行。先创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install fastapi uvicorn[standard] httpx pydantic python-dotenv3.2 项目结构为了不把代码写得过于抽象示例文件结构如下federated-search-agent/ ├── app/ │ ├── __init__.py │ ├── schema.py # 统一结果定义 │ ├── searchers.py # 多数据源适配器 │ ├── federated.py # 联邦调度与结果融合 │ ├── agent_tool.py # 面向 Agent 的工具函数 │ └── main.py # FastAPI 入口 ├── docs/ │ └── fastapi_notes.txt # 本地文档检索示例数据 ├── requirements.txt ├── .env.example └── README.md实际项目中每个适配器可以单独放在一个文件里方便团队并行开发。这里为了演示清晰把适配器集中在一个模块里你也可以在代码里看到每个类的边界。3.3 统一结果 Schema 设计联邦搜索的第一步是先定数据结构而不是先写调用逻辑。如果每个数据源都返回自己的结构后面的融合逻辑会非常痛苦。下面定义一个统一的SearchResult# 文件路径app/schema.py from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class SearchResult(BaseModel): 统一联邦检索结果结构。 id: str Field(description结果唯一标识) title: str Field(description标题) snippet: str Field(description摘要或正文片段) url: Optional[str] Field(defaultNone, description原始链接) source: str Field(description数据源名称例如 web / github / local) score: float Field(default0.0, description数据源内部的相关度分数) published_at: Optional[datetime] Field(defaultNone, description发布时间) extra: dict Field(default_factorydict, description扩展字段)这个 Schema 有几点考虑id用于后续去重常见做法是将数据源名称和原始 ID 拼接比如github:12345。source字段要保留不仅用于调试也方便 Agent 在回答里注明信息来源。score是各数据源自己的分数不能直接用于跨源排序跨源排序由融合层处理。extra是兜底字段避免因为某个数据源多返回几个字段就导致模型解析失败。4. 核心模块实现4.1 数据源适配器抽象适配器层是所有联邦搜索的基础。它的职责是接收标准查询输入调用外部系统再返回标准结果列表。为了便于扩展可以先定义一个抽象基类# 文件路径app/searchers.py from abc import ABC, abstractmethod from app.schema import SearchResult class BaseSearcher(ABC): 数据源适配器抽象类。 name: str base abstractmethod async def search(self, query: str, top_k: int 5) - list[SearchResult]: 执行一次检索并返回标准结果。 raise NotImplementedError实际开发中有些检索是同步阻塞的比如读取本地文件。处理方式是在适配器内部用asyncio.to_thread转成异步调用避免阻塞事件循环。这里的核心思想是对 Agent 暴露异步接口内部实现可以自由选择同步或异步。4.2 本地文档检索适配器本地文档检索是最容易理解的一个适配器。它不依赖任何外部 API最适合用来验证联邦搜索整体流程。这里用一个简单的关键词打分实现不引入向量数据库。逻辑是遍历 docs 目录下的文本文件。按换行切分成段落对每个段落统计查询词的出现次数。按命中次数排序返回前 top_k 条。# 文件路径app/searchers.py延续上面代码 import asyncio import re from pathlib import Path from app.schema import SearchResult class LocalDocsSearcher(BaseSearcher): 本地文本文件检索适配器。 name local def __init__(self, docs_dir: str ./docs): self.docs_dir Path(docs_dir) async def search(self, query: str, top_k: int 5) - list[SearchResult]: return await asyncio.to_thread(self._sync_search, query, top_k) def _sync_search(self, query: str, top_k: int) - list[SearchResult]: if not self.docs_dir.exists(): return [] words re.findall(r[\w\u4e00-\u9fff], query.lower()) results: list[SearchResult] [] for file_path in self.docs_dir.glob(*.txt): text file_path.read_text(encodingutf-8) paragraphs [p.strip() for p in text.split(\n) if p.strip()] for idx, paragraph in enumerate(paragraphs): low_para paragraph.lower() hit sum(1 for w in words if w in low_para) if hit 0: continue # 简单截取摘要避免返回过长内容 snippet paragraph[:200] results.append( SearchResult( idflocal:{file_path.stem}:{idx}, titlef{file_path.stem} 第 {idx 1} 段, snippetsnippet, sourceself.name, scorefloat(hit), ) ) results.sort(keylambda r: r.score, reverseTrue) return results[:top_k]在实际业务中这个适配器通常会被替换成 Elasticsearch、OpenSearch 或向量库检索。替换时只需要保持search方法签名不变联邦调度层不需要做任何修改。4.3 GitHub Issues 检索适配器GitHub Issues 是非常适合 Agent 使用的数据源因为很多技术问题的答案就在 Issue 讨论中。如果你在开发 Agent让它可以搜索 GitHub Issues等于给它开了“开源社区经验库”。GitHub Search API 是一个公开接口可以直接用httpx异步调用。为了示例简洁这里直接用Authorization: Bearer token做鉴权token 可选但强烈建议配置否则速率限制很低。# 文件路径app/searchers.py新增 GitHub 适配器 import os import httpx from app.schema import SearchResult GITHUB_API https://api.github.com class GithubIssueSearcher(BaseSearcher): GitHub Issues 检索适配器。 name github def __init__(self, token: str ): self.token token or os.getenv(GITHUB_TOKEN, ) async def search(self, query: str, top_k: int 5) - list[SearchResult]: headers {Accept: application/vnd.githubjson} if self.token: headers[Authorization] fBearer {self.token} params { q: query, per_page: top_k, } async with httpx.AsyncClient(timeout10.0) as client: resp await client.get( f{GITHUB_API}/search/issues, paramsparams, headersheaders, ) resp.raise_for_status() data resp.json() results [] for item in data.get(items, []): results.append( SearchResult( idfgithub:{item[id]}, titleitem[title], snippetitem.get(body) or , urlitem[html_url], sourceself.name, scorefloat(item.get(score, 0)), published_atitem.get(created_at), extra{state: item.get(state), repo: item.get(repository_url, ).split(/repos/)[-1]}, ) ) return results注意这里解析repository_url的方式比较粗暴只是演示用。真实项目建议从item[repository_url]中解析 owner/repo或者改用 GraphQL API 一次拿到更多结构化字段。4.4 Web 搜索适配器Web 搜索适配器是最常被替换的模块。不同厂商的搜索 API 参数差异很大这里做一个通用封装外部传入SEARCH_ENDPOINT和SEARCH_API_KEY适配器内部用 GET 请求把 query 发过去再统一解析通用结构。为了避免绑定特定服务商下面代码采用一个常见的返回结构假设JSON 中包含results数组数组元素包含title、link、snippet。如果对接真实服务只需调整解析逻辑。# 文件路径app/searchers.py新增 Web 搜索适配器 import os import httpx from app.schema import SearchResult class WebSearcher(BaseSearcher): Web 搜索适配器使用通用搜索 API 配置。 name web def __init__(self): self.endpoint os.getenv(SEARCH_ENDPOINT, ) self.api_key os.getenv(SEARCH_API_KEY, ) async def search(self, query: str, top_k: int 5) - list[SearchResult]: if not self.endpoint: return [] headers {Authorization: fBearer {self.api_key}} params {q: query, count: top_k} async with httpx.AsyncClient(timeout8.0) as client: resp await client.get(self.endpoint, paramsparams, headersheaders) resp.raise_for_status() data resp.json() results [] for idx, item in enumerate(data.get(results, [])): results.append( SearchResult( idfweb:{item.get(id, idx)}, titleitem.get(title, ), snippetitem.get(snippet, ), urlitem.get(link, ), sourceself.name, scorefloat(item.get(score, 0)), ) ) return results很多搜索 API 的返回结构里没有score或者 score 的含义差异很大。这种情况下建议在适配器里给结果一个“归一化后的内部分数”比如按位置倒序给分第一名 1.0第二名 0.9。这一步能让融合层拿到相对可比的初始分数。4.5 联邦调度器并行查询与超时降级有了多个适配器之后接下来就是核心的联邦调度逻辑。示例使用asyncio.gather并发执行所有适配器并允许每个适配器失败时降级为空结果而不是让整个查询失败。这里有几个工程细节需要强调每个适配器建议单独设置短超时默认 8 到 10 秒。使用return_exceptionsTrue收集异常然后在日志里记录避免一个数据源故障阻塞全部查询。可以对不同数据源做并行度的控制防止同时发出过多请求导致限流。# 文件路径app/federated.py import asyncio import logging from app.schema import SearchResult from app.searchers import BaseSearcher, LocalDocsSearcher, WebSearcher, GithubIssueSearcher logger logging.getLogger(__name__) class FederatedSearcher: 联邦搜索调度器。 def __init__(self, searchers: list[BaseSearcher] | None None): self.searchers searchers or [ LocalDocsSearcher(), WebSearcher(), GithubIssueSearcher(), ] async def search(self, query: str, top_k: int 5) - list[SearchResult]: tasks [searcher.search(query, top_kmax(top_k, 3)) for searcher in self.searchers] results await asyncio.gather(*tasks, return_exceptionsTrue) merged: list[SearchResult] [] for searcher, result in zip(self.searchers, results): if isinstance(result, Exception): logger.warning(searcher %s failed: %s, searcher.name, result) continue if result is None: continue merged.extend(result) # 这里先做简单去重融合排序在 RRF 模块中完成 seen set() unique_results [] for item in merged: if item.id in seen: continue seen.add(item.id) unique_results.append(item) return unique_results4.6 结果融合RRF 与加权过滤很多新手第一次做联邦搜索时最容易犯的错就是把不同源的 score 直接相加排序。但 GitHub 的 score 和搜索 API 的 score 完全不是一个量级直接相加等于让分数高的数据源完全主导结果。我常用的方案是 RRFReciprocal Rank Fusion倒数排名融合。它的思想很简单不看分数绝对值只看每条结果在每个数据源中的排名。RRF 的核心公式是score_rrf(item) sum(1 / (k rank(item, source)))其中 k 通常取 60是经验值。每个结果在数据源中排名越靠前贡献的分数越高。这样能很好消除不同数据源打分尺度不一致的影响。# 文件路径app/federated.py新增 RRF 融合方法 def reciprocal_rank_fusion( results: list[SearchResult], k: int 60, top_k: int 5, ) - list[SearchResult]: RRF 结果融合返回 top_k 条结果。 rrf_scores: dict[str, float] {} detail: dict[str, SearchResult] {} # 按 source 分组保证排名是在同一数据源内部计算 source_groups: dict[str, list[SearchResult]] {} for item in results: source_groups.setdefault(item.source, []).append(item) for source, items in source_groups.items(): # 先按原始分数降序得到每个结果在当前源中的排名 sorted_items sorted(items, keylambda r: r.score, reverseTrue) for rank, item in enumerate(sorted_items): rrf_scores[item.id] rrf_scores.get(item.id, 0.0) 1.0 / (k rank 1) detail[item.id] item ranked_ids sorted(rrf_scores, keyrrf_scores.get, reverseTrue) merged_results [detail[rid] for rid in ranked_ids[:top_k]] return merged_resultsRRF 有两个明显优势不需要不同数据源的分数可比天然适用于联邦搜索。对离群分数不敏感某个源异常给了超高分也不会直接霸榜。在联邦搜索类里最终的search方法返回前先调用一次 RRF 融合# 文件路径app/federated.py延续前面代码 async def search_with_rank(self, query: str, top_k: int 5) - list[SearchResult]: merged await self.search(query, top_k) return reciprocal_rank_fusion(merged, top_ktop_k)4.7 缓存模块联邦搜索往往比普通搜索更贵因为一次查询要消耗多个后端资源。给高频查询加缓存是性价比最高的优化手段。这里实现一个简单的 TTL 内存缓存按(query, top_k)为 key 存储结果。# 文件路径app/federated.py新增缓存逻辑 import time from collections import OrderedDict class TTLCache: 简单的 TTL 缓存用于避免重复联邦查询。 def __init__(self, capacity: int 128, ttl: int 300): self.capacity capacity self.ttl ttl self._store: OrderedDict[str, tuple[float, list]] OrderedDict() def get(self, key: str): if key not in self._store: return None expire_at, value self._store[key] if time.time() expire_at: self._store.pop(key) return None # 移到末尾表示最近访问 self._store.move_to_end(key) return value def set(self, key: str, value): self._store[key] (time.time() self.ttl, value) self._store.move_to_end(key) if len(self._store) self.capacity: self._store.popitem(lastFalse)然后将缓存放进FederatedSearcher# 文件路径app/federated.py在 __init__ 中增加 self.cache TTLCache(capacity128, ttl300) # 在使用时 cache_key f{query}:{top_k} cached self.cache.get(cache_key) if cached is not None: return cached # ... 执行联邦搜索 ... self.cache.set(cache_key, final_results)要点缓存 key 必须包含查询参数和 top_k避免不同参数之间互相污染。生产项目可以用 Redis 做分布式缓存并给缓存设置合理的过期时间。5. 完整实战为 Agent 提供联邦搜索工具5.1 Agent 工具函数封装对 Agent 来说它不关心你的 Class 内部怎么实现它只关心“我调用一个函数能不能拿到有用的结果”。所以最好是直接提供一个纯函数返回 JSON 字符串方便嵌入到 Function Calling 的 tool 定义中。# 文件路径app/agent_tool.py import json from app.federated import FederatedSearcher searcher FederatedSearcher() async def federated_search_tool(query: str, top_k: int 5) - str: Agent 工具执行联邦搜索并返回 JSON 字符串。 results await searcher.search_with_rank(query, top_ktop_k) payload [item.model_dump() for item in results] return json.dumps(payload, ensure_asciiFalse)如果你用的是 OpenAI Function Calling 或者类似框架可以在 tool 定义里把federated_search_tool注册成一个异步工具。这里不绑定具体框架方便你迁移到 LangChain、LlamaIndex 或自研的 Agent 引擎。5.2 FastAPI 接口封装除了函数调用有时 Agent 运行在独立服务里需要通过 HTTP 调用联邦搜索。用 FastAPI 封装一层接口很简单。# 文件路径app/main.py from fastapi import FastAPI, Query from app.agent_tool import federated_search_tool app FastAPI(titleFederated Search for AI Agents) app.get(/search) async def search(query: str Query(..., description查询词), top_k: int Query(5, ge1, le20)): 面向 Agent 的统一联邦搜索接口。 result_json await federated_search_tool(queryquery, top_ktop_k) return {query: query, result_json: result_json}5.3 运行与验证启动服务前先在docs目录下放一个示例文件。比如docs/fastapi_notes.txtFastAPI 支持通过 StreamingResponse 实现流式响应。 这个过程需要返回一个异步生成器并且设置正确的媒体类型。 如果客户端是 OpenAI SDK还可以直接把流式响应传给模型接口。 使用 httpx 调用 GitHub API 时需要设置 Authorization 请求头。然后启动服务export GITHUB_TOKENyour_github_token_here export SEARCH_ENDPOINThttps://your-search-api.example.com/search export SEARCH_API_KEYyour_api_key_here uvicorn app.main:app --reload --port 8000如果你暂时没有可用的搜索 API可以把WebSearcher从FederatedSearcher的初始化列表中去掉仅保留本地文档和 GitHub Issues 两个源同样可以跑通流程。浏览器访问http://localhost:8000/search?queryFastAPI%20流式响应top_k5预期会得到类似下面的 JSON{ query: FastAPI 流式响应, result_json: [{\id\:\local:fastapi_notes:0\,\title\:\fastapi_notes 第 1 段\,\snippet\:\FastAPI 支持通过 StreamingResponse 实现流式响应。\,\source\:\local\,\score\:0.0},{\id\:\github:123456\,\title\:\Need streaming response example\,\snippet\:\...\,\source\:\github\,\score\:0.0}] }注意经过 RRF 融合后每条结果的score已经变成 RRF 分数不再是数据源原始分数。这是符合预期的因为score字段在统一 Schema 里代表“融合后的排序分”。6. 常见问题与排查思路学习联邦搜索的过程中下面几个问题出现频率非常高。问题现象常见原因解决思路整个查询特别慢某个数据源响应超时拖慢整体 gather给每个适配器设置独立 timeout并对慢源做好降级结果里全是同一个数据源直接对不同源的分数求和排序改用 RRF 这类基于排名融合的算法本地文件检索不到中文内容没有正确切分查询词使用re.findall(r[\w\u4e00-\u9fff])兼容中英文GitHub API 报 403未配置 token触发速率限制配置GITHUB_TOKEN并做好重试退避搜索结果重复较多多个数据源索引了相同内容用id去重并把 URL 归一化处理返回给 Agent 的内容太长top_k 过大或 snippet 过长控制 top_k 数量对 snippet 做截断或摘要下面重点展开两个容易忽略的点。第一asyncio.gather本身的超时与子任务超时不是一回事。即使给httpx.AsyncClient设置了 timeout也只是单个请求读超时。更稳妥的做法是在联邦调度层再包一层asyncio.wait_for确保最坏情况下整个联邦查询在限定时间内返回。第二RRF并不总能保证高相关性。它更多是提供一个稳定、公平的融合基准。如果对排序质量要求很高可以考虑在 RRF 之后再加一个rerank模型对候选结果做语义重排。不过 rerank 会增加延迟和成本实践时要做好取舍。7. 最佳实践与工程建议7.1 Schema 先行接口清晰在写联邦搜索代码之前先把统一结果 Schema 定义清楚。数据源可以后续慢慢接入但 Schema 一旦在多个适配器中铺开修改成本会迅速上升。建议在extra字段里预留扩展空间而不是频繁修改顶层字段。7.2 超时、重试与熔断联邦搜索天然依赖多个外部系统任何一个系统抖动都会影响 Agent 体验。推荐在适配器层做“短超时”在调度层做“整体超时”在适配器内部做“有限重试”。重试策略优先选择指数退避并设置最大重试次数为 1 到 2 次。如果某个源连续失败超过阈值可以打开熔断让该源暂时从调度列表中摘除。7.3 缓存与成本控制联邦搜索的成本与调用次数强相关。除了 TTL 缓存还可以做对同一 query 的不同 top_k 做结果复用。对热门查询做预热缓存。对低优先级搜索或者 Agent 仅仅是“猜测性补充知识”时降低 top_k 到 3 以下。记录每个数据源的调用次数和成本方便后评价数据源性价比。7.4 权限与数据边界Agent 搜索内部数据时必须严格遵循最小权限原则。建议每个数据源适配器只使用对应系统的最小权限 token并在查询结果中保留来源标记避免内部数据被错误地拼进外部上下文。如果 Agent 需要把搜索结果发送给另一个外部模型务必确认没有泄露内部敏感信息。这个点很重要尤其在接入企业 Wiki 或工单系统时。7.5 可观测性联邦搜索是典型的多依赖系统必须有完整的日志链路。推荐为每次联邦查询生成一个trace_id贯穿 Agent 调用、调度、各适配器请求、融合排序全流程。日志至少包含查询词、命中的数据源、各源耗时、各源返回数量、融合结果 id 列表、总耗时。7.6 让 Agent 学会引用来源联邦搜索的价值不只是“拿到答案”还包括“知道答案来自哪里”。建议把source字段和url字段一并传给 Agent并约束 Agent 在回答中标注来源。这既能提升回答可信度也方便用户回溯问题。8. 总结与下一步学习路线这篇文章从 AI Agent 的检索痛点出发介绍了 Federated Search 的概念、与 RAG 和集中式搜索的区别并落地了一套最小可运行的联邦搜索项目。核心收获可以概括为三点一是统一 Schema 是联邦搜索的基础设施二是适配器隔离了不同数据源的差异三是 RRF 是解决跨源排序不一致问题的有效手段。下一步可以继续深入的方向包括结果重排序Rerank与语义融合、基于意图的路由分发、自适应 top_k 选择、与 RAG 管线深度整合以及基于 MCP 协议把联邦搜索能力通用化。最后留一个建议联邦搜索的上限往往取决于结果融合质量而不是数据源数量。先把两条异构数据源跑通再用同样的模式慢慢扩展会比你一次性接十个数据源可控得多。如果你在实践过程中遇到新的问题欢迎在评论区留言交流。