最近在折腾基于大模型的应用你会发现一个非常现实的问题知识不是存放在一个地方而是散落在数据库、向量库、企业 Wiki、第三方 API 和各类内部系统里。Agent 拿着用户的问题却不知道该去哪找也不知道怎么把不同来源的结果拼成一段可信的上下文。很多人第一反应是“把所有内容都灌进向量库跑一遍 RAG”。这个方案在演示场景里没问题一旦面对权限隔离、实时更新、多租户、源系统权威性冲突很快就会兜不住。联邦搜索Federated Search提供了一个完全不同的思路不复制数据而是在多个数据源之上加一层“检索调度层”让 Agent 用统一入口访问所有来源同时把每个来源的实时性、权限和权威性保留下来。这个想法并不新早期企业搜索时代就是用它解决多站点统一检索的。但到了 AI Agent 时代它重新变得重要因为它实际解决的是 Agent 的信息获取架构问题。这篇文章会先厘清联邦搜索与集中式搜索、向量检索的本质区别然后拆解它的核心架构链路再给出一份可直接运行的 Python 最小实现包括多数据源适配器、RRF 结果融合、统一检索 API以及如何封装成 Agent 可调用的工具。最后还会聊一聊生产环境的融合策略、故障排查和最佳实践。如果你正在做 Agent、RAG 或企业知识应用这篇文章建议先收藏再读。1. 联邦搜索是什么和集中式搜索、向量检索的本质差异要理解联邦搜索必须先把它和另外两个容易混淆的概念放在一起看集中式搜索和向量检索。三者都在解决“让用户找到想要的信息”这件事但底层架构逻辑完全不同。集中式搜索的代表是 Elasticsearch、搜索引擎等。它的核心动作是“搬数据”通过爬虫、同步任务、CDC 等方式把多个源头的数据统一采集到一个中央索引里然后在这个索引上做倒排、分词、打分和排序。优点是查询性能好、排序可控、能统一做权限。缺点是数据会延迟、存储成本和维护成本高而且源系统往往不愿意把数据完整复制出去。向量检索则是 RAG 应用里最常见的方案。它把文档切块用 Embedding 模型转成向量放进向量数据库查询时把用户问题也转成向量做近似度检索。它的优势是支持语义匹配能解决同义词、口语化表达的问题。但它同样是“先建副本”的思路而且对权限、元数据过滤、事实准确性的支持往往比较弱。联邦搜索的思路完全不同。它不做全量数据复制而是在查询发生时把请求分发给多个数据源每个数据源用自己的索引、自己的排序逻辑去检索最后再把各来源的结果汇总融合。每个数据源仍然是“自己的主人”联邦层只负责路由、适配、归并和排序。维度集中式搜索向量检索RAG联邦搜索数据存放复制到中央索引复制为向量副本数据留在原系统数据新鲜度取决于同步频率取决于索引更新周期查询时实时读取权限控制索引内统一设计元数据过滤复杂权限下沉到各数据源排序方式中央统一打分向量余弦相似度各源独立打分后再融合容错能力中央索引故障全挂向量库故障全挂单源故障可降级典型成本存储同步链路Embedding向量存储查询分发融合逻辑可以这样类比集中式搜索像是把各分馆的书全部搬到总馆读者只需要去总馆查向量检索像是给每本书做一张语义摘要卡按语义相似度找书联邦搜索则像总馆门口的调度台收到读者问题后打电话问各分馆“你们那儿有没有相关的书”再把各分馆报上来的书目做一次汇总排序。这个区别决定了联邦搜索的适用边界当数据源数量多、权限隔离要求高、数据实时更新频繁、源系统不同意复制数据时联邦搜索几乎是唯一合理的架构。2. 为什么 Agent 场景会重新需要联邦搜索联邦搜索并不是新概念为什么现在又值得拿出来讨论因为 AI Agent 的流行把过去企业搜索领域的痛点又放大了而且放大了好几倍。第一个痛点是工具爆炸。如果你给 Agent 直接对接多个数据源每增加一个来源就要为 Agent 多设计一套工具调用参数、返回格式和错误处理逻辑。假设内部有 CRM、工单系统、知识库、代码仓库、日志平台五个来源Agent 的工具列表会变得非常长LLM 的调用准确率会明显下降。与其让 Agent 直接面对五个工具不如只暴露一个“联邦搜索”工具由它在内部做二次调度。第二个痛点是 RAG 的失效模式。RAG 在单知识库场景表现很好但一旦涉及多业务系统就会出现很多问题不同系统的数据更新频率不同集中索引很难及时跟进跨系统存在同一实体、同一概念的表达冲突某些数据出于合规要求不能出域根本无法进入集中索引。这些问题不是调参能解决的而是架构问题。第三个痛点是权限与合规。金融、医疗、企业内部场景里“数据不出域”是硬性要求。联邦搜索的优势在于查询请求在进入数据源时由数据源本地的权限模块做鉴权数据不需要物理集中到联邦层。你在联邦层拿到的只是各数据源允许你看到的检索结果而不是底层数据的完整副本。第四个痛点是成本与上下文。把所有相关信息都塞进 prompt 不现实。Agent 需要的是“先检索、再筛选、最后拼装上下文”。联邦搜索刚好能扮演“第一轮粗筛”的角色用多源检索找到候选内容再由重排策略筛选出最值得进入上下文的片段。这也意味着联邦搜索的输出格式要尽量紧凑、结构化方便 LLM 做下一步处理。所以要有一个清晰判断Agent 不需要拥有所有数据它需要拥有“找数据的能力”。联邦搜索就是把这个能力沉淀成一个独立组件。3. 联邦搜索核心架构从 Query 到 Answer 的完整链路一个面向 Agent 的联邦搜索系统并不是简单地把多个搜索请求都跑一遍再把结果拼在一起。完整的链路至少要经过以下几个环节。用户问题 → Agent 规划并调用 federated_search 工具 → 查询解析与改写扩展同义词、识别实体 → 数据源路由与选择哪些源参与本次检索 → 多数据源并行分发SQL 库 / 向量库 / HTTP API → 适配器结果归一化统一字段结构 → 结果融合排序RRF 或加权融合 → 上下文拼装生成精简摘要、附带来源信息 → Agent 组织最终回答查询解析与改写是容易被忽略的一环。不同数据源对查询的理解能力不同SQL 数据源适合关键词精确匹配向量库适合语义检索外部 API 可能需要特定的参数格式。因此联邦搜索层通常要有一个查询改写模块把用户的自然语言问题转成各来源更友好的检索表达式。数据源路由需要解决两个问题哪些来源参与本次检索以及要不要同时检索。有些来源名称敏感、含义不同需要按意图路由有些来源检索成本高可以在低置信度时跳过。路由策略可以做成规则也可以做成一个小的分类模型初期用规则完全够用。多数据源分发最核心的工程要点是“并行”。如果串行调用多个数据源整体延迟会等于所有来源耗时之和这在实时场景里很难接受。并行调用时要考虑慢数据源对整体的影响所以每一路调用都要有独立的超时和降级逻辑。某个来源挂了不能让整个联邦搜索不可用。适配器归一化解决的是“方言不同”的问题。数据库返回行、向量库返回文档块、外部接口返回 JSON字段名五花八门。联邦搜索层应该在适配器内统一转换最终输出给上层的是结构完全一致的结果对象包含 title、snippet、source、url、score_raw、metadata 等字段。这一步做不好后续的融合和上下文拼装都会很痛苦。结果融合排序是整个联邦搜索最特殊的一环。不同数据源的“得分”体系完全不可比SQL 里可能没有相关性分数向量库给的是 0 到 1 的余弦相似度HTTP 接口给的可能是自己算的关键词重叠率。直接用分数相加没有意义。所以联邦搜索通常采用不依赖原始分数、只依赖排名的融合算法最常用的是 RRFReciprocal Rank Fusion这个我们在后面用代码演示。上下文拼装是在融合排序之后进行的。不能把原始结果原封不动丢给 Agent而是要把每个来源的 title、snippet、url、source 组装成结构化文本保证来源可追溯。这一步不仅影响回答质量还直接决定 Agent 在回答时能不能准确引用出处减少幻觉。4. 最小实现多数据源适配器理论讲完现在进入可运行的最小实现。这个演示不追求功能完整而是把联邦搜索的核心链路跑通。我们先设计一个模拟项目结构。federated-search-agent-demo/ ├── adapters.py # 多数据源适配器 ├── fusion.py # RRF 结果融合 ├── app.py # FastAPI 统一检索入口 ├── agent_tool.py # Agent 工具封装与调用示例 └── requirements.txt # 依赖声明先写数据源适配器。这里实现三个完全不同的来源一个 SQLite 数据库、一个轻量向量知识库、一个模拟外部 HTTP 搜索接口。为了让代码可以直接运行这个演示不依赖任何第三方数据库和机器学习库向量检索部分用 TF-IDF 模拟语义检索的思想。# 文件路径federated-search-agent-demo/adapters.py import math import sqlite3 import time from typing import Dict, List, Any from dataclasses import dataclass, field dataclass class SearchResult: title: str snippet: str source: str unknown url: str score_raw: float 0.0 metadata: Dict[str, Any] field(default_factorydict) def as_dict(self) - Dict[str, Any]: return { title: self.title, snippet: self.snippet, source: self.source, url: self.url, score_raw: self.score_raw, metadata: self.metadata, } def tokenize(text: str) - List[str]: # 简化分词英文按小写分词中文按字母数字串切分。 # 生产环境建议使用 jieba 或其他中文分词库。 text text.lower() tokens [] current [] for ch in text: if ch.isalnum(): current.append(ch) else: if current: tokens.append(.join(current)) current [] if current: tokens.append(.join(current)) return tokens class SQLSource: 从 SQLite 数据库检索模拟企业内部关系型数据源。 def __init__(self, name: str internal_sql): self.name name self.conn sqlite3.connect(:memory:) self.conn.execute( CREATE TABLE articles ( id INTEGER PRIMARY KEY, title TEXT, content TEXT, tags TEXT ) ) sample_rows [ (1, 企业内部文档搜索规范, 搜索是知识管理的关键环节需要覆盖文档、数据和外部信息。, 搜索,知识管理), (2, AI Agent 架构设计, Agent 需要规划、记忆与工具调用能力才能在真实业务中稳定工作。, Agent,架构), (3, RAG 检索增强生成实践, RAG 通过外部知识库增强大模型的回答质量减少幻觉。, RAG,检索), (4, 联邦学习与数据安全, 联邦学习尝试在不共享原始数据的情况下完成联合建模。, 联邦,安全), ] self.conn.executemany( INSERT INTO articles VALUES (?, ?, ?, ?), sample_rows ) self.conn.commit() def search(self, query: str, top_k: int 5) - List[SearchResult]: # 简化实现按关键词 LIKE 匹配。 # 生产环境建议使用数据库自带的全文索引例如 SQLite FTS5。 like f%{query}% rows self.conn.execute( SELECT id, title, content, tags FROM articles WHERE title LIKE ? OR content LIKE ? OR tags LIKE ? LIMIT ? , (like, like, like, top_k), ).fetchall() results [] for row_id, title, content, tags in rows: results.append( SearchResult( titletitle, snippetcontent[:80], sourceself.name, urlfsql://articles/{row_id}, score_raw1.0, metadata{table: articles, id: row_id, tags: tags}, ) ) return results class VectorSource: 用 TF-IDF 模拟向量检索只依赖 Python 标准库。 def __init__(self, docs: List[Dict[str, str]], name: str vector_kb): self.name name self.docs docs self.doc_tokens: List[List[str]] [] self.doc_tf: List[Dict[str, int]] [] self.idf: Dict[str, float] {} self.norms: List[float] [] self._build_index() def _build_index(self): doc_count len(self.docs) df: Dict[str, int] {} for doc in self.docs: tokens tokenize(doc[title] doc[content]) self.doc_tokens.append(tokens) for token in set(tokens): df[token] df.get(token, 0) 1 for token, d in df.items(): self.idf[token] math.log((1 doc_count) / (1 d)) 1.0 for tokens in self.doc_tokens: tf: Dict[str, int] {} for token in tokens: tf[token] tf.get(token, 0) 1 self.doc_tf.append(tf) norm math.sqrt( sum((tf.get(tok, 0) * self.idf.get(tok, 0)) ** 2 for tok in tf) ) self.norms.append(norm) def search(self, query: str, top_k: int 5) - List[SearchResult]: q_tokens tokenize(query) q_tf: Dict[str, int] {} for token in q_tokens: q_tf[token] q_tf.get(token, 0) 1 q_vec { token: count * self.idf.get(token, 0) for token, count in q_tf.items() } q_norm math.sqrt(sum(w ** 2 for w in q_vec.values())) or 1.0 scored [] for idx, doc in enumerate(self.docs): doc_vec { token: count * self.idf.get(token, 0) for token, count in self.doc_tf[idx].items() } dot sum(q_vec.get(token, 0) * weight for token, weight in doc_vec.items()) doc_norm self.norms[idx] or 1.0 score dot / (q_norm * doc_norm) if score 0: scored.append((score, idx)) scored.sort(keylambda x: x[0], reverseTrue) results [] for score, idx in scored[:top_k]: doc self.docs[idx] results.append( SearchResult( titledoc[title], snippetdoc[content][:80], sourceself.name, score_rawround(score, 4), metadata{doc_id: idx}, ) ) return results class HttpSource: 模拟外部 HTTP 搜索接口内部用关键词重叠度打分。 def __init__(self, name: str external_http): self.name name self.mock_index [ { title: Federated Search Overview, content: Federated search allows querying multiple data sources at once without copying data into a central index., url: https://example.com/federated-search-overview, }, { title: AI Agent Tool Calling, content: LLM agents use function calling to access external tools and structured data sources., url: https://example.com/agent-tool-calling, }, { title: Vector Database for RAG, content: Vector databases are widely used in retrieval augmented generation to store embeddings., url: https://example.com/vector-db-rag, }, { title: MCP: Model Context Protocol, content: Model Context Protocol standardizes how AI applications connect to external tools and data sources., url: https://example.com/mcp, }, ] def search(self, query: str, top_k: int 5) - List[SearchResult]: time.sleep(0.03) # 模拟网络 IO 延迟 query_tokens set(tokenize(query)) scored [] for doc in self.mock_index: doc_tokens set(tokenize(doc[title] doc[content])) score len(query_tokens doc_tokens) / max(1, len(query_tokens)) if score 0: scored.append((score, doc)) scored.sort(keylambda x: x[0], reverseTrue) results [] for score, doc in scored[:top_k]: results.append( SearchResult( titledoc[title], snippetdoc[content][:80], sourceself.name, urldoc[url], score_rawround(score, 4), ) ) return results这段代码最关键的设计是SearchResult统一数据结构。三个适配器返回的都是同一种对象这样上层融合逻辑不需要关心结果到底来自数据库还是外部接口。实际项目中适配器层还可以做字段映射、内容清洗、权限过滤甚至把原系统返回的 HTML 转成纯文本但统一结果出口的思路是一样的。5. 结果融合RRF 排序与上下文组装多数据源拿到结果之后接下来的问题是怎么把这些结果排成一个有序列表直接按原始分数融合是不行的。SQLSource 返回的 score 是固定的 1.0VectorSource 返回的是 0 到 1 的余弦相似度HttpSource 返回的是关键词重叠率。三种分数分布完全不同加权平均没有意义。更稳妥的办法是用 RRFReciprocal Rank Fusion它只关心每个数据源内部的名次不关心分数绝对值。RRF 的核心公式是每个文档的融合得分等于它在每个来源中排名位置的倒数之和并引入一个常数 K 来平滑。# 文件路径federated-search-agent-demo/fusion.py from typing import Dict, List, Any from adapters import SearchResult def reciprocal_rank_fusion( results_by_source: Dict[str, List[SearchResult]], k: int 60, top_k: int 5, ) - List[Dict[str, Any]]: 对多个数据源的检索结果做 RRF 融合排序。 k 是平滑常数经验上常用 60。k 越小名次越靠前的结果优势越大。 fused_scores: Dict[str, Dict[str, Any]] {} for source_name, results in results_by_source.items(): for rank,