检索工具化封装：把 Elasticsearch、SQL、Confluence 包装成 Agent 的统一工具

个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录检索工具化封装把 Elasticsearch、SQL、Confluence 包装成 Agent 的统一工具一、为什么不能一个 Agent 直连所有数据源二、统一工具协议定义 Agent 所见三、适配器模式各数据源的实现3.1 向量库适配器3.2 ES 适配器query DSL 翻译3.3 SQL 适配器白名单查询 只读护栏3.4 Wiki/API 适配器四、权限注入工具层是最佳位置五、工具注册与发现六、性能与可靠性七、落地检查清单八、小结检索工具化封装把 Elasticsearch、SQL、Confluence 包装成 Agent 的统一工具Agentic RAG 的地基是检索即工具——Agent 不需要知道数据存在哪、怎么查它只调工具、拿结果。但企业内部的数据源五花八门向量库、Elasticsearch、MySQL、Confluence、公司自建 API……每个的查询方式、返回格式、权限模型都不一样。这篇文章讲怎么把这些差异封装掉给 Agent 一套统一的检索工具接口适配器模式怎么设计、返回格式怎么定、权限在哪一层注入、以及工具注册和发现机制。一、为什么不能一个 Agent 直连所有数据源天真架构长这样Agent 的 prompt 里列五个数据源Function Calling 直接调各家的 SDK。三个月内必崩原因prompt 被接口细节污染ES 的 query DSL、MySQL 的 SQL、Confluence 的 CQL——每个工具的 description 都在拉低 Agent 的路由准确率返回格式不统一ES 返回 hits、SQL 返回 rows、Confluence 返回 HTML——Agent 每接一种格式就多一类解析错误权限无法收口Agent 拿着一个数据库超级账号查哪都通——审计做不了越权查不了加一个数据源 改一遍 Agent耦合在 prompt 里每次加源都改动核心逻辑回归测试全重跑。工具化封装就是解决这四个问题的面向 Agent 的是统一协议面向数据源的是各自的适配器。二、统一工具协议定义 Agent 所见先定协议——Agent 看到的所有工具共享同一张脸frompydanticimportBaseModelfromtypingimportOptionalclassUnifiedRetrievalInput(BaseModel):所有检索工具的统一入参query:str# 自然语言查询或关键词filters:Optional[dict]None# 结构化过滤时间、类型、部门…top_k:int5# 返回条数classUnifiedHit(BaseModel):统一结果单元id:strscore:floattitle:strsnippet:str# 压缩摘要不是全文source:str# 数据源标识es_logs / mysql_orders / confluencemetadata:dict# 时间、作者、权限标签等full_text_ref:Optional[str]# 深挖取全文的引用句柄classUnifiedRetrievalOutput(BaseModel):hits:list[UnifiedHit]total:inttook_ms:int这个协议设计的三个关键决策snippet 而不是 full_text默认返回压缩内容一两百字全文通过full_text_ref按需取——上下文预算的第一道闸Agentic 场景下这是省钱关键source 字段必填Agent 知道每条信息来自哪答案里能做来源标注审计时能回溯filters 结构化Agent 可以表达只看 2026 年的但不用知道底层是 ES 的 range query 还是 SQL 的 WHERE——过滤语义统一。三、适配器模式各数据源的实现每种数据源一个适配器实现同一接口classBaseAdapter(ABC):source_name:strabstractmethoddefsearch(self,inp:UnifiedRetrievalInput,ctx:UserContext)-UnifiedRetrievalOutput:...abstractmethoddefget_full_text(self,ref:str,ctx:UserContext)-str:...3.1 向量库适配器classVectorStoreAdapter(BaseAdapter):source_namekb_docsdefsearch(self,inp,ctx):# 1. 查询改写可复用查询理解的中间产物# 2. 带权限过滤的向量检索关键权限下推到检索层resultsvector_store.search(embeddingembed(inp.query),top_kinp.top_k*2,# 多取一倍过滤后可能不够where{acl.visible_to:ctx.user_groups,# 多租户 ACLdoc_type:inp.filters.get(doc_type),created_at:{$gte:inp.filters[start]}ifinp.filterselseNone,})# 3. 转统一格式snippet 取 chunk 前 200 字returnUnifiedRetrievalOutput(hits[...],totallen(results),took_ms12)defget_full_text(self,ref,ctx):docvector_store.get(ref,check_aclctx)# 取全文时二次校验权限returndoc.full_text注意top_k * 2和get_full_text里的权限二次校验——权限在检索层过滤、在取全文层复核两层都要ACL 元数据可能滞后取全文时文档权限可能已变更。3.2 ES 适配器query DSL 翻译classElasticsearchAdapter(BaseAdapter):source_namees_logsdefsearch(self,inp,ctx):must[{multi_match:{query:inp.query,fields:[title^2,body]}}]# filters → ES 结构化过滤翻译ifinp.filters:ifstartininp.filters:must.append({range:{timestamp:{gte:inp.filters[start]}}})iflevelininp.filters:must.append({term:{level:inp.filters[level]}})# 权限按用户可见的业务线过滤must.append({terms:{biz_line:ctx.visible_biz_lines}})respself.es.search(indexlogs-*,body{query:{bool:{must:must}},size:inp.top_k,_source:[title,body,timestamp,biz_line],})# 统一成 hitsbody 截断成 snippet...翻译层是每个适配器的核心工作把统一 filters 译成 ES DSL / SQL WHERE / CQL。翻译逻辑集中在一处Agent 永远看不到底层差异。3.3 SQL 适配器白名单查询 只读护栏数据库类数据源是最危险的——Agent 生成的 SQL 要是能跑任意语句一次注入就是事故。安全做法禁止自由 SQL只允许模板化白名单查询classMySQLAdapter(BaseAdapter):source_namedb_orders# 白名单查询每个有明确语义和安全边界TEMPLATES{order_by_user:{sql:SELECT id, amount, status, created_at FROM orders WHERE user_id %(user_id)s AND created_at %(start)s LIMIT %(limit)s,params_schema:{user_id:str,start:date,limit:int},allowed_sort:[],},order_stats_by_status:{sql:SELECT status, COUNT(*) cnt, SUM(amount) total FROM orders WHERE created_at %(start)s GROUP BY status,...},}defsearch(self,inp,ctx):# 查询分类器决定用哪个模板LLM 分类 or 规则template_nameself.route(inp.query)tplself.TEMPLATES[template_name]paramsself.extract_params(inp,tpl[params_schema])rowsself.db.execute(tpl[sql],params)# 参数化查询returnUnifiedRetrievalOutput(hits[row_to_hit(r)forrinrows],...)三条铁律只走模板、参数化传参、连接用只读账号。Agent 想自由发挥 SQL门都没有。3.4 Wiki/API 适配器Confluence、飞书知识库这类适配器做三件事认证透传用 ctx 里的用户身份调 API不能用服务账号绕过权限、HTML/Markdown 转 snippet、分页归一。四、权限注入工具层是最佳位置权限该在哪管答案是工具层用调用上下文贯穿classUserContext(BaseModel):user_id:strgroups:list[str]# AD/LDAP 组visible_biz_lines:list[str]# 可见业务线doc_acl:dict[str,list[str]]# 文档级 ACL 映射如需要# 工具调用链agent.call_tool(search_docs,args,ctxuser_ctx)→ adapter.search(inp,ctx)→ 检索层 where 权限过滤向量库/ES 的 ACL 过滤 → 取全文层 ACL 复核 → 结果带 sourcemetadata 返回三个设计要点权限下推到检索层不是查完再过滤——查 1000 条过滤剩 5 条是浪费且 top_k 语义全乱取全文时复核检索到 → 决定看全文 → 这一刻再校验一次中间文档权限可能变了全程记录审计日志哪个用户、什么查询、命中哪些文档、调了哪个工具——合规审查和安全复盘全靠它。五、工具注册与发现数据源多了工具注册要动态化classToolRegistry:def__init__(self):self._adapters{}defregister(self,adapter:BaseAdapter,description:str,when_to_use:str):self._adapters[adapter.source_name]adapterdeftool_specs(self,ctx)-list[dict]:按用户可见的数据源生成工具清单specs[]forname,adinself._adapters.items():ifnotctx.can_access(name):continue# 没权限的数据源Agent 根本看不到这个工具specs.append({type:function,function:{name:fsearch_{name},description:f{description}。适用场景{when_to_use},parameters:UnifiedRetrievalInput.model_json_schema(),}})returnspecs没权限就看不到比看到了但报错好两个量级——Agent 不会去尝试不可用的工具prompt 也更干净。权限感知的工具清单是安全和效果的共同最优解。六、性能与可靠性统一封装层还顺手解决三个工程问题超时与降级每个适配器配超时ES 2s、SQL 5s、外部 API 10s超时返回该源暂不可用而不是拖死整个 Agent——Agent 可以基于其他源继续答并发调用Agent 一轮里调三个工具时并发执行延迟取 max 而非 sum结果缓存相同 query filters 用户组的检索结果缓 5 分钟工具层的语义缓存Agent 重复检索同一内容时直接命中——省 API 也省数据源压力。七、落地检查清单统一协议定义统一入参query/filters/top_k、统一结果snippet 模式 source metadata full_text_ref每个数据源一个适配器filters 翻译逻辑集中在适配器内SQL 类源只走白名单模板 参数化 只读账号禁止自由 SQL权限下推检索层 取全文复核两层UserContext 贯穿全链工具清单按权限生成无权限数据源不可见工具 description 写清适用/不适用场景这是路由准确率的一半适配器级超时 降级 并发 缓存审计日志用户/查询/命中/工具四要素落库。八、小结检索工具化的三句话面向 Agent 统一协议面向数据源各自适配统一入参、统一结果snippet 优先、全文按需、统一权限模型——Agent 永远只看到一套工具适配器模式隔离差异filters 翻译成各家的 DSL/SQL/白名单翻译逻辑集中加数据源不动 Agent 核心权限在工具层收口检索层下推过滤、取全文层复核、按权限生成工具清单——安全、效果、可审计三个目标一次满足。下一篇讲长文档处理当文档超过上下文窗口分层摘要和父子索引怎么兜住检索质量。