企业官网建设流程全解析

最近 RAG 相关的工作里，“Agentic RAG”出现频率越来越高。传统的做法是：把用户问题丢给 Embedding，匹配向量库，取 Top-K 条片段，拼到 Prompt 里交给大模型回答。这一步在多数场景下够用，但它是一个典型的黑盒操作：你只知道返回了 K 条内容和相似度分数，但不知道模型为什么选这些、漏掉了哪些、排序合不合理。如果查询是“对比两家公司的产品定价策略和售后政策差异”，Top-K 一次性检索几乎很难给出高质量证据链。

这次我们来看的方向，就是标题里这句话：Beyond Top-K，用可解释的 Agentic 操作替代黑盒检索。它不一定是某个现成开源项目，而是一套检索范式：把“检索”拆成一连串可观察、可控制、可审计的操作，由 Agent 根据问题动态决定执行哪些操作，而不是让一个黑盒向量检索把 Top-K 结果硬塞给模型。这篇文章会从思路、设计、概念原型、接口、批量任务和排查几个角度，带你把这套思路落到自己的 RAG 系统里。

先说结论：如果你是做复杂问答、多跳推理、可解释性要求高的知识库场景，这个方向值得认真试。如果只是给一个简单 FAQ 文档做检索，Top-K 短平快，没必要上 Agentic。后面所有讨论都是围绕“何时值得换、怎么换、换了怎么验证”展开。

1. 核心能力速览

上面表格是“能力速览”，不是软件清单。要让这套东西工作，至少需要三块：一个能调工具的大模型（API 或本地部署）、一组可执行的检索操作（向量搜索、关键词搜索、过滤、重排等）、一套控制这些操作的 Agent 循环。

2. 为什么 Top-K 检索不够用

先明确一个前提：Top-K 检索本身不是坏事。它快、稳定、资源占用低，是多数 RAG 系统的默认底座。但它有几个结构性短板。

2.1 相似度不等于相关性

向量检索计算的是语义相似度，但“相似”和“回答所需的信息相关”是两件事。用户问“2024 年第三季度销售增长主要来自哪个区域”，如果知识库里同时存在“销售增长原因分析”“区域业绩报表”“季度总结 PPT 标题”，它们和问题的相似度可能都很高，但真正能给出答案的可能是被排在第 11 位的那段。

Top-K 的 K 是拍脑袋定的。K=5 可能漏，K=20 则把大量噪音塞进上下文，模型更糊涂，token 成本也上去了。

2.2 黑盒不可解释

传统检索链路里，你只能看到每个片段的 score。但 score 是 Embedding 模型内部的向量距离，没人能解释它为什么觉得这段和问题相关。如果线上回答错了，你没法回答“是哪一步检索导致它选了错误证据”。

2.3 无法处理多步信息需求

很多真实问题需要多跳推理。比如“A 公司相比 B 公司在欧洲市场的渠道策略有哪些差异”，你需要先定位“A 公司欧洲渠道策略”，再定位“B 公司欧洲渠道策略”，最后做对比。传统 Top-K 只做一次全局检索，很难把这两个跳的上下文都精确找齐。

2.4 没有反馈和重试机制

黑盒 Top-K 是一次性的：检索完就完事了。大模型发现上下文里缺信息时，没法说“我再查一下”。它只能硬着头皮生成。

Agentic 检索的核心价值就是把这一串不可控的黑盒过程，变成一组显式的、可解释的、可失败重试的操作序列。

3. 可解释 Agentic 检索的设计思路

所谓“可解释的 Agentic 操作”，是指把检索过程拆成多个原子操作，每个操作有名字、有输入、有输出、有参数，并且整个过程由 Agent 动态编排。用户和开发者可以看到每一步在做什么，也可以干预和审计。

3.1 原子操作清单

常用的检索原子操作包括：

传统 Top-K 检索相当于把 vector_search 和 rerank 两步压缩成一个黑盒函数：给你一个 query，返回 K 条。Agentic 检索则允许 Agent 决定先做 keyword_search 再做 vector_search，或者先 filter 再 rerank，甚至发现信息不够时再补一轮搜索。

3.2 Agent 循环

Agentic 操作需要一个控制循环：一个大模型充当“决策器”，观察当前状态（用户问题、已有的检索结果、缺失的信息），选择一个操作执行，然后观察结果，决定下一步是继续检索还是生成最终回答。

一个最小循环是：

当前状态 = 用户问题 循环: 让 LLM 决策下一步操作 如果操作是 generate_answer，就跳出循环 否则执行对应操作，把结果追加到状态里 如果达到最大步数，强制跳出

这个流程本身是老 ReAct / Toolformer 模式，但重点在于：每一步操作都是显式命名的，执行过程有日志，最终答案可以挂到一条可追溯的操作链上。

3.3 可解释性从哪里来

• 操作名称本身就是语义化的人类可读标签。
• 每个操作有输入输出记录，可以生成 JSON 日志。
• Agent 在每一步会输出一段简短的自然语言“理由”，例如“用户问题涉及两家公司对比，我需要分别检索两家公司的渠道策略”。
• 最终答案可以引用操作链上的具体片段，比如[filter 后片段 #2]。

这套设计的好处是：当线上系统给出一个错误答案时，你能打开日志看到 Agent 去哪里检索了、检索了什么、为什么停止了检索。这就是“解释性”。

4. 适用场景与使用边界

4.1 适合的场景

• 多跳问答：需要多个实体、多轮检索才能回答的问题。
• 对比型问题：两种产品、两家公司、两个政策之间的对比。
• 证据链敏感场景：医疗、法律、金融等需要引用准确来源的场景。
• 企业知识库：文档量大、结构复杂、需要按部门/标签/时间过滤检索。
• 高质量长文生成：需要先收集多个角度的信息，再组织成章节。

4.2 不适合的场景

• 简单 FAQ：单轮问答，一条向量检索就够。
• 高吞吐低延迟：Agentic 循环每次决策都会调用 LLM，延迟和成本都会明显增加。
• 上下文长度受限极严格的应用：每一步操作结果都要塞进上下文，token 消耗比一次性 Top-K 高。
• 没有日志审计需求的轻量工具：没必要引入这么重的编排。

4.3 使用边界与合规提醒

如果你把 Agentic 检索接入企业知识库，处理客户数据、员工信息或受版权保护的文档，必须确保数据来源合法、检索范围经过授权，并且整个操作链的日志需要控制访问权限。不要用 Agentic 检索去抓取未经授权的网页、绕过登录限制或者爬取需要权限的内容。检索结果可能带有模型偏见和文档本身的不准确信息，发布前必须人工复核。

5. 环境准备与前置条件

虽然我们讨论的是方法论，但要落地验证，你至少需要准备以下环境。以 Python 生态为例，这是一个通用检查清单，具体版本以你的项目依赖为准。

如果你只想先验证“可解释操作链”的可行性，最简单的方式是：

• 用 Qdrant 或 Chroma 存一批测试文档。
• 用一个支持工具调用的 LLM API。
• 自己写 10 个检索函数，不引入重型 Agent 框架。

这样能快速跑通第一步，后续再决定是否上框架。

6. 概念原型：最小可运行的 Agentic Retrieval

下面给出一套概念原型代码，目的不是提供一个生产可用的工具，而是展示“可解释的 Agentic 操作”长什么样。如果你准备用在自己的项目里，需要按实际函数名、模型接口、向量库调用方式调整。

6.1 定义原子操作

先用枚举定义操作类型，方便 Agent 决策和日志记录：

from enum import Enum class RetrievalOp(str, Enum): KEYWORD_SEARCH = "keyword_search" VECTOR_SEARCH = "vector_search" FILTER = "filter" DEDUPLICATE = "deduplicate" RERANK = "rerank" VERIFY = "verify" GENERATE_ANSWER = "generate_answer"

每种操作对应一个工具函数，函数命名和枚举保持一致。这里给出 vector_search 和 filter 的示意：

def vector_search(query_text: str, top_n: int = 10) -> list[dict]: # 伪代码：用 embeddings 编码 query，在向量库查询 top_n 条 # results: [{"id": ..., "text": ..., "score": ..., "metadata": {...}}] return query_vector_db(query_text, top_n=top_n) def filter(candidates: list[dict], conditions: dict) -> list[dict]: # 伪代码：按 metadata 条件过滤，例如 {"department": "销售部"} return [c for c in candidates if match_metadata(c["metadata"], conditions)]

6.2 Agent 决策循环

核心是一个循环：把用户问题、历史操作记录、当前候选片段打包成消息，交给 LLM 输出一个结构化决策。

import json def run_agentic_retrieval(user_question: str, max_steps: int = 6): state = { "question": user_question, "candidates": [], "operations": [], "step": 0, } while state["step"] < max_steps: # 1. 让 LLM 从 RetrievalOp 中选择下一步操作，并给出参数和理由 decision = ask_llm_for_decision(state) # 伪代码函数 op_name = decision["op"] params = decision.get("params", {}) reason = decision.get("reason", "") # 2. 记录操作日志 state["operations"].append({ "step": state["step"], "op": op_name, "params": params, "reason": reason, }) # 3. 执行操作 if op_name == RetrievalOp.VECTOR_SEARCH: state["candidates"].extend(vector_search(state["question"], **params)) elif op_name == RetrievalOp.FILTER: state["candidates"] = filter(state["candidates"], **params) elif op_name == RetrievalOp.GENERATE_ANSWER: return generate_final_answer(state), state else: # 其他操作按需实现 raise NotImplementedError(f"未实现操作: {op_name}") state["step"] += 1 # 4. 达到最大步数仍未生成答案，基于当前状态兜底生成 return generate_final_answer(state), state

这里有几个关键点：

• 每一步决策都由 LLM 输出 JSON，方便解析和审计。
• state["operations"]就是可解释的操作链。
• 必须设置max_steps防止 Agent 陷入死循环。
• 兜底策略：达到最大步数时，用当前已收集的片段生成答案。

6.3 可解释输出格式

最终返回的结构应该包含两部分：操作链和答案。操作链是核心，绝对不能丢掉。一个理想的返回结果长这样：

{ "answer": "A 公司在欧洲主要通过经销商渠道...", "operation_chain": [ { "step": 0, "op": "vector_search", "params": {"top_n": 10, "query": "A公司欧洲渠道策略"}, "reason": "需要定位A公司的欧洲渠道信息", "candidate_count": 10 }, { "step": 1, "op": "vector_search", "params": {"top_n": 10, "query": "B公司欧洲渠道策略"}, "reason": "需要定位B公司的欧洲渠道信息，以便对比", "candidate_count": 10 }, { "step": 2, "op": "filter", "params": {"department": "国际业务部"}, "reason": "排除与渠道策略无关的文档", "candidate_count": 6 } ] }

这种输出，才是“可解释检索”的价值所在。

7. 功能测试与效果验证

换了检索范式，验证方法也要跟着变。不能只看最终答案的 Rouge/BLEU 分数，需要同时验证检索质量和过程可解释性。

7.1 测试数据集准备

建议准备 50 到 200 条真实查询，标注以下信息：

• 问题文本。
• 正确答案（或参考证据片段）。
• 需要几跳才能回答（单跳/双跳/三跳）。
• 是否存在需要过滤的元数据约束。
• 是否会出现误导性相似片段。

排序优先级：先做双跳和对比型问题，这类是 Agentic 检索最能发挥优势的场景。

7.2 单条功能测试

以“A 公司相比 B 公司的开源协议有何不同”为例：

1. 启动 Agentic Retrieval 服务。
2. 传入问题。
3. 观察操作链，确认 Agent 是否执行了至少两次检索：一次查 A 公司开源协议，一次查 B 公司开源协议。
4. 查看每一步候选片段数，确认没有在第一步就生成答案。
5. 检查最终答案引用的片段是否能直接支撑结论。

判断成功的标准：

• 操作链中确实存在两次针对不同实体的检索操作。
• 每一步的理由文本和实际执行的参数一致。
• 最终答案里的对比结论能在检索片段中找到对应原文。

7.3 对比测试：Top-K vs Agentic

在相同测试集上分别跑传统 Top-K RAG 和 Agentic Retrieval，记录：

这里要给一个冷建议：Agentic 检索在准确率上未必全面超过 Top-K，尤其在简单问题上。所以先跑小样本对比，再决定是否全量切换。不要为了“新”而换。

7.4 失败场景验证

必须准备一些坏 case，比如：

• 知识库里完全没有答案的问题。
• 查询语句有歧义的问题。
• 需要同时过滤多个标签的问题。
• 两个实体名称高度相似的问题。

观察 Agent 在失败场景下是否出现以下问题：

• 在一个并无信息的实体上反复检索。
• 提前生成答案，没有执行足够的检索。
• 操作链断裂，无法回溯。

如果出现这些问题，你的最大步数、决策 Prompt 和工具描述都需要调。

8. 接口 API 与批量任务

如果要在工程里落地，需要把 Agentic Retrieval 封装成服务。下面给一个 FastAPI 接口示例，方便接到现有系统。

8.1 同步接口

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class RetrievalRequest(BaseModel): question: str max_steps: int = 6 class RetrievalResponse(BaseModel): answer: str operation_chain: list @app.post("/agentic_retrieval", response_model=RetrievalResponse) def agentic_retrieval(request: RetrievalRequest): answer, state = run_agentic_retrieval( request.question, max_steps=request.max_steps ) return RetrievalResponse( answer=answer, operation_chain=state["operations"] )

启动服务：

uvicorn main:app --host 127.0.0.1 --port 8000

调用接口：

curl -X POST "http://127.0.0.1:8000/agentic_retrieval" \ -H "Content-Type: application/json" \ -d '{"question": "A公司相比B公司的开源协议有何不同", "max_steps": 8}'

8.2 批量任务设计

同步接口适合单条调试。如果要对几百个问题跑批量，最好拆成任务队列。核心流程是：

1. 读取问题列表。
2. 为每个问题创建任务，记录状态。
3. 后台 worker 逐个调用run_agentic_retrieval。
4. 每个任务结束，把答案和操作链写入 JSON 文件或数据库。
5. 针对失败任务设置重试，建议最多重试 1-2 次。

简化的 Python 批量脚本：

import json import time from concurrent.futures import ThreadPoolExecutor questions = [ {"id": 1, "text": "A公司 vs B公司 开源协议"}, {"id": 2, "text": "C产品在2024年Q4的销量变化"}, ] def process_one(item): question_id = item["id"] try: answer, state = run_agentic_retrieval( item["text"], max_steps=6 ) result = { "id": question_id, "status": "ok", "answer": answer, "operation_chain": state["operations"], } except Exception as exc: result = { "id": question_id, "status": "failed", "error": str(exc), } return result if __name__ == "__main__": with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(process_one, questions)) with open("batch_results.json", "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2)

注意：如果 LLM 是 API 方式，并发需要留意限流；如果本地模型，则需要考虑显存和 GPU 队列。实际实现需要结合你的模型部署方式。

9. 资源占用与性能观察

9.1 显存和内存

Agentic Retrieval 本身并不是一个模型，它不直接占用显存。显存来自底层 Embedding 模型和生成式 LLM：

如果你直接用 API 模型，本地只需要跑 Embedding 和向量库，资源要求很低。显存占用这一项必须以实际部署模型为准，上面只是常见经验参照，不是精确数字。

9.2 Token 成本

Agentic 模式相比 Top-K 多出的核心成本是每步决策都要调一次 LLM。假设一次决策消耗 500 token 输出 + 1000 token 输入（问题、历史操作、候选摘要），5 步操作就是 7500 token。这在 API 模式下会直接影响费用。

降低成本的几个办法：

• 限制max_steps不要超过 6。
• 每次决策只传递操作摘要，不传递完整候选文本。
• 对简单问题先做一次分类，只有分类为“复杂查询”才进入 Agentic 流程。

9.3 延迟观察

每步决策的延迟 = LLM 响应时间 + 工具执行时间。本地 LLM 一步推理可能就需要 1-5 秒，5 步就是 5-25 秒。如果对延迟敏感，建议：

• 用更小的决策模型。
• 缓存重复检索结果。
• 把 Agentic 检索作为离线分析任务的增强模块，而不是线上实时检索的默认路径。

9.4 降低资源占用思路

• 使用量化模型。
• 使用混合检索，简单关键词先过滤。
• 控制每次向量搜索的top_n，不要动辄返回 50 条。
• 对操作链做持久化时只保留必要字段，避免日志爆炸。

10. 常见问题与排查方法

11. 最佳实践与落地建议

11.1 先做小规模 A/B 测试

不要一上来就把所有流量切到 Agentic。挑 50 个真实复杂查询，跑一版 Top-K，再跑一版 Agentic，对比准确率、召回、延迟、费用。数据会让你决定是否值得。

11.2 把操作链当作一等公民

操作链不是调试辅助，而是产品的一部分。在前端展示“AI 检索过程”，后端保存操作链 JSON，用户反馈错误答案时直接关联到对应步骤。这会大大降低 RAG 系统的排查难度。

11.3 给每个操作函数设计统一接口

建议每个工具函数都遵循以下签名：

def op_function(state: dict, **kwargs) -> dict: # 返回更新后的 state

这样 Agent 循环可以统一调用，不需要针对每个操作写分支。统一接口也方便插入日志和监控。

11.4 用迷你模型做决策，大模型做生成

很多场景下，决策和生成可以分离：

• 用便宜、快的小模型（如 7B 或 API 的小模型）做操作决策。
• 用高质量的大模型做最终答案生成。
• 这样既控制成本，又保证答案质量。

11.5 设置安全边界

Agent 在执行检索操作时必须有边界：

• 可以访问哪些数据源。
• 哪些字段可以被过滤。
• 哪些查询不允许执行（例如涉及未授权数据）。
• 操作链日志的访问权限。

一定要在工程实现里加一层授权校验，不能把检索工具直接暴露给不可信输入。

11.6 定期复盘失败案例

把线上回答错误的 case 收集起来，连同操作链一起分析。如果发现很多错误都来自“过早生成答案”，就在决策 Prompt 里强调“必须完成至少一次检索才能生成答案”。如果错误都来自“过滤条件误伤”，就需要调整过滤操作的行为。

12. 总结与下一步

这个方向最值得尝试的点，是它把检索从一个不可解释的数学计算变成了一串可以阅读的操作日志。你需要先跑一组小样本，对比一下两份结果：一份是 Top-K RAG 的输出，一份是带操作链的 Agentic 输出。优先验证“多跳对比型问题”，也最容易体现价值。

最容易踩的坑有三个：第一，操作链太长，token 成本飙高；第二，Agent 决策不稳定，反复执行同一操作；第三，没有兜底逻辑，卡在循环里。第一版实现的时候，先把最大步数设小一点，比如 4 步，跑通了再放宽。

后续可以扩展的方向：把操作链接入 LangSmith / MLflow 这类追踪工具；把工具函数扩展为 web_search、database_query、arxiv_search 等外部操作；在决策循环里加入“自我反思”能力，让 Agent 在上一步检索结果不足时自动改关键词重新搜索。最实用的建议是：先别追求通用，挑一个你手里最头疼的复杂查询场景，把这套可解释操作链接进去。效果好不好，操作链会告诉你原因。