1. 为什么 Top-K 检索不够用了先回想一个非常常见的开发场景我们在做企业知识库问答、客服智能助手或者日志检索系统时最标准的做法是把用户问题输入 embedding 模型转成向量再到向量数据库中做相似度检索取出分数最高的 Top-K 条候选文本喂给大模型生成回答。这个流程看起来顺滑部署也快但真正上了生产环境后问题会一个接一个暴露出来。第一个问题是“不可解释”。用户问“最近一周有哪些未完成的订单”系统返回了三段内容其中两段是本周已完成的订单。你问它为什么返回这两段答案只有一个向量相似度分数高。但为什么高是关键词重叠还是句法结构相似还是碰巧语义空间靠得近没有人能回答因为向量相似度本身不产生任何可阅读的中间结论。第二个问题是“不可干预”。当业务方说“我不想让已关闭的工单进入候选集”你只能做两件事要么调整 Prompt要么把状态过滤条件写死在取数 SQL 里。前者不稳定后者又绕过了检索链路导致过滤逻辑分散在多处。K 值到底取 5 还是 20阈值设在 0.7 还是 0.75这些问题经常靠拍脑袋决定。第三个问题是“不可审计”。一旦用户投诉“系统提供了错误信息”运维人员只能看到一条检索请求和一段候选 ID 列表完全无法复盘信息是从哪个业务环节进来的、哪一步过滤没生效、哪一步召回的置信度本来就很低。Top-K 本身并不是错误它是信息检索领域沉淀多年的经典做法。真正的问题在于当系统复杂度上升后Top-K 这条决策路径太短了query 进来向量相似度算完直接取前 K 条。中间没有任何可以被验证、被干预、被审计的环节。这就迫使我们在检索层引入一种更结构化的思路把一次检索拆成一串可执行的、可解释的、可观测的原子操作也就是本文要讨论的 Agentic Operations。2. 核心概念可解释的 Agentic Operations 到底是什么通俗地讲Agentic Operations 是“把一次检索动作从一次黑盒查询变成一个有步骤、有记录、有中间产物的工作流”。传统检索的流程是用户问题 - 向量化 - 向量库 Top-K - 拼接上下文 - LLM 生成Agentic Operations 的流程变成了用户问题 - 理解查询意图 - 拆解操作步骤 - 逐步执行操作 - 每一步记录日志 - 组装证据 - LLM 生成这里有一个关键变化检索过程中的每一步操作都有明确的名称、职责、输入和输出。比如“时间范围过滤”操作它接收查询意图中的时间实体在候选集上执行时间条件过滤输出过滤后的文档 ID 列表。这个操作可以被单独测试可以被记录日志如果结果异常也能单步回放。为什么要用“操作”这个概念因为操作具备三个传统检索没有的工程特性第一可组合性。操作像积木一样可以按不同查询类型组合成不同的执行链路。查询“某工单当前状态”和“最近一个月已解决工单分布”两者需要的操作序列完全不同。操作可以在不同链路中复用。第二可观测性。每个操作执行后都会产生结构化日志包括操作名、耗时、输入数量、输出数量、命中结果。整条链路执行完毕系统可以输出一份执行轨迹用户和开发者都能看到检索结论是如何一步步形成的。第三可干预性。如果某个操作执行失败或者结果置信度低系统可以在该步骤直接降级、重试、或转人工。而传统 Top-K 检索要么一次性给出结果要么全部失败没有中间状态。可以做一个直观对比对比维度传统 Top-K 黑盒检索Agentic Operations 检索决策路径query - 相似度 - TopKquery - 操作序列 - 每步结果 - 最终候选中间结果无每个操作都有输出结果解释只有相似度分数完整执行轨迹和证据链失败处理整体降级或空结果单操作重试、降级、兜底干预方式调 Prompt、调 K、改阈值增删操作、调整编排规则审计能力无结构化操作日志可回放从这个角度看Agentic Operations 不是要完全否定向量检索而是把向量检索“降级”为整条操作链中的一个普通环节——把黑盒的空间缩短把可以白盒化的部分全部白盒化。3. 环境准备与项目结构在动手写代码前先把环境准备讲清楚。本文示例以 Python 为主版本建议使用 3.9 或更高版本因为后续代码中会用到类型注解和 dataclass 等特性。如果你用的是 3.10还可以使用 match 语句做编排分支但我们这里为了兼容性尽量用常见的 if-else 写法。依赖方面本文的核心逻辑不依赖特定的大模型厂商 SDK所以不引入重型的框架。你需要准备以下基础组件Python 3.9用于编写流程编排和执行器。一个向量数据库用于语义召回常见选择包括 Milvus、Qdrant、pgvector、Elasticsearch 的向量插件等具体版本随你的实际项目而定。一个可调用的 LLM 接口用于查询理解和操作编排可以是云端 API也可以是本地部署模型。如果你希望重排效果更好可以准备一个 rerank 模型的 HTTP 服务比如 bge-reranker 或 cross-encoder 类模型。注意这里所有外部组件的版本都不做硬性指定因为不同团队的基础设施差异很大。本文重点演示的是编排逻辑、操作抽象和可解释链路的设计思路你完全可以将示例中的向量库调用替换成自己团队的接口。下面给出一个建议的工程目录结构agentic_retriever/ ├── operations/ │ ├── __init__.py │ ├── base.py # 操作原语与结果定义 │ ├── exact_match.py # 精确匹配操作 │ ├── semantic_retrieve.py # 向量召回操作 │ ├── filter.py # 时间/状态过滤操作 │ └── rerank.py # 语义重排操作 ├── orchestration/ │ ├── __init__.py │ ├── planner.py # 操作编排器负责生成操作序列 │ └── executor.py # 操作执行器负责串行/并行执行操作 ├── evidence/ │ └── assembler.py # 证据组装器将操作结果整理为上下文 ├── main.py # 示例入口 └── config.yaml # 配置文件这个结构并不复杂核心设计思想是把操作封装成独立模块把编排和执行分开把最终结果交给证据组装器。这样无论后续增加新操作还是调整执行策略都不会影响整体架构。4. 核心设计把检索引擎拆成可执行的原子操作4.1 操作原语与上下文先定义最底层的操作模型。我们需要一个统一的数据结构来表示“一次操作执行后的结果”以及一个承载执行过程中共享信息的上下文对象。先看操作结果的定义# 文件路径operations/base.py from dataclasses import dataclass, field from typing import Any, Optional dataclass class OperationResult: op_name: str # 操作名称 status: str # success / failed / degraded input_count: int 0 # 输入候选数量 output_count: int 0 # 输出候选数量 output_ids: list field(default_factorylist) # 输出的文档 ID detail: Optional[Any] None # 额外信息如日志、分数 error_msg: str # 失败时的错误信息这个结果对象是整条链路可解释的基石。每一步操作完成后都会生成一个 OperationResult被记录到操作日志中。后续如果要输出给用户看只要把这些结果序列化即可。接下来定义操作上下文。上下文负责在多个操作之间传递共享数据比如原始 query、解析后的意图、当前候选 ID 集合等。dataclass class OperationContext: query: str intent: dict field(default_factorydict) # 查询意图实体 candidate_ids: list field(default_factorylist) # 当前候选 ID 集合 metadata_map: dict field(default_factorydict) # 文档元信息映射 trace: list field(default_factorylist) # 执行轨迹4.2 操作基类为了让所有操作保持统一接口我们定义一个 BaseOperation 抽象类。所有具体操作都继承它。from abc import ABC, abstractmethod class BaseOperation(ABC): op_name: str base_operation def __init__(self, priority: int 100): self.priority priority abstractmethod def execute(self, ctx: OperationContext) - OperationResult: 执行具体操作逻辑并返回可审计的操作结果 pass def explain(self) - str: 返回操作的人类可读描述用于组装给用户看的执行轨迹 return f执行操作: {self.op_name}这里的 priority 字段用于排序。比如“时间过滤”优先级应该高于“向量召回”因为先缩小范围再做语义检索效率更高、更准确。4.3 一个可运行的最小操作示例下面实现两个具体的操作时间范围过滤操作和语义召回操作。这两个操作覆盖了“先缩小范围、再语义检索”的典型链路。# 文件路径operations/filter.py from datetime import datetime from .base import BaseOperation, OperationResult, OperationContext class TimeRangeFilterOperation(BaseOperation): 时间范围过滤从查询意图中提取 start_time 和 end_time 然后根据文档元数据中的 created_at 字段过滤候选 ID。 op_name time_range_filter def __init__(self, priority: int 10): super().__init__(prioritypriority) def execute(self, ctx: OperationContext) - OperationResult: intent ctx.intent start_time intent.get(start_time) end_time intent.get(end_time) # 如果查询意图中没有时间范围直接返回成功不改变候选集 if not start_time and not end_time: return OperationResult( op_nameself.op_name, statussuccess, input_countlen(ctx.candidate_ids), output_countlen(ctx.candidate_ids), output_idsctx.candidate_ids, detail意图中无时间范围跳过过滤, ) filtered_ids [] for doc_id in ctx.candidate_ids: meta ctx.metadata_map.get(doc_id, {}) created_at meta.get(created_at, ) if not created_at: continue # 过滤逻辑要求创建时间在 [start_time, end_time] 内 if start_time and created_at start_time: continue if end_time and created_at end_time: continue filtered_ids.append(doc_id) ctx.candidate_ids filtered_ids return OperationResult( op_nameself.op_name, statussuccess, input_countlen(ctx.candidate_ids), output_countlen(filtered_ids), output_idsfiltered_ids, detailf过滤时间范围: {start_time} ~ {end_time}, )这段代码的核心逻辑很简单拿到查询意图中的时间范围遍历候选文档根据元数据过滤。但请注意它真正有价值的地方在于每一步都记录了输入数量、输出数量和过滤条件。这为后续排查提供了精确依据。再看语义召回操作。这里我假设你已经封装了一个向量检索函数vector_search(query, limit)它返回文档 ID 列表和分数。实际项目中你可以把这段替换为对 Milvus、pgvector 或 Elasticsearch 的真实调用。# 文件路径operations/semantic_retrieve.py from .base import BaseOperation, OperationResult, OperationContext def vector_search(query: str, limit: int 20): 模拟向量检索函数实际项目中应替换为向量数据库的真实调用。 返回 (ids, scores)例如 ([doc_1, doc_2], [0.85, 0.72]) # 示意实现不代表真实逻辑 return [], [] class SemanticRetrieveOperation(BaseOperation): 向量语义召回从向量库中检索出与 query 最相关的 top_n 文档。 这里不追求精确召回重要的是把这一步作为整条链路的子环节 而不是整个检索的全部。 op_name semantic_retrieve def __init__(self, priority: int 50, top_n: int 20): super().__init__(prioritypriority) self.top_n top_n def execute(self, ctx: OperationContext) - OperationResult: # 如果已有候选集则基于候选 ID 做向量召回否则全库召回 if ctx.candidate_ids: # 带候选集的向量召回可加速检索 ids, scores vector_search(ctx.query, limitself.top_n, filter_idsctx.candidate_ids) else: ids, scores vector_search(ctx.query, limitself.top_n) ctx.candidate_ids ids return OperationResult( op_nameself.op_name, statussuccess, input_count0, output_countlen(ids), output_idsids, detail{top_n: self.top_n, scores: scores}, )这里的vector_search只是一个示意函数实际项目中需要你根据向量库的 SDK 实现。这也是为什么我在代码注释里特意标明“示意实现”。4.4 操作编排层规则与 LLM 的取舍有了操作原语接下来要考虑如何为一次查询选择操作序列。编排策略一般有两种选择。第一种是规则编排适合逻辑清晰、意图可枚举的场景。比如产品约定查询包含时间实体时必须先执行时间过滤查询包含单号时必须先执行精确匹配。这种方式的优点是完全可控、零额外延迟、容易测试缺点是覆盖不了复杂语义。第二种是 LLM 编排让大模型根据用户问题自动生成操作序列。这种方式灵活但必须做严格的结构化输出校验否则大模型可能生成一个不存在的操作名或者输出非法 JSON。实际的工程实践中推荐混合策略先通过规则识别强意图强意图直接绑定操作序列弱意图再交给 LLM 编排并对 LLM 输出做 schema 校验。下面给出一个规则编排的示例# 文件路径orchestration/planner.py from operations.exact_match import ExactMatchOperation from operations.filter import TimeRangeFilterOperation, StatusFilterOperation from operations.semantic_retrieve import SemanticRetrieveOperation from operations.rerank import RerankOperation def decide_operations_by_rule(intent: dict, query: str): 基于规则的操作编排函数。 返回一个 BaseOperation 列表。 operations [] # 精确匹配优先如果识别到订单号/工单号等强标识 if intent.get(entity_id): operations.append(ExactMatchOperation(priority5)) # 时间过滤 if intent.get(start_time) or intent.get(end_time): operations.append(TimeRangeFilterOperation(priority10)) # 状态过滤 if intent.get(status): operations.append(StatusFilterOperation(priority20)) # 语义召回 operations.append(SemanticRetrieveOperation(priority50, top_n20)) # 重排 operations.append(RerankOperation(priority80)) return operations这个编排函数的输入是意图解析结果。意图可以由一个独立的 NER 模型或 LLM 从 query 中抽取。为了步骤清晰我建议把意图解析视为整条链路的第一步操作而不是外部前置逻辑。如果是 LLM 编排核心代码会类似这样import json def decide_operations_by_llm(query: str, llm_client) - list: 使用 LLM 生成操作序列。注意这里依赖 llm_client.complete() 实际接入时需要根据你的模型服务 SDK 调整。 prompt f 你是一个信息检索流程编排器。请根据用户问题输出一个 JSON 数组 数组元素是操作名操作名必须是以下之一 exact_match, time_range_filter, status_filter, semantic_retrieve, rerank 用户问题{query} 输出格式示例 [time_range_filter, semantic_retrieve, rerank] resp llm_client.complete(prompt) try: ops json.loads(resp) valid_ops {exact_match, time_range_filter, status_filter, semantic_retrieve, rerank} ops [op for op in ops if op in valid_ops] return ops except Exception: # 解析失败时回退到默认操作序列 return [semantic_retrieve]这段代码的关键在于LLM 输出必须先做合法性校验过滤掉不在白名单内的操作名并且解析失败要回退到默认链路。这一点在线上非常重要因为生成流程不允许因为一次 JSON 解析失败而整条链路崩溃。5. 完整实战企业知识库场景下的 Agentic 检索链路现在我们把上面的模块组装成一个完整的可执行流程。场景设定为企业知识库系统用户提问“查询 2025 年第二季度状态为已关闭的测试报告”。这个查询有三个明显特征包含时间范围2025 年第二季度、包含状态条件已关闭、实体类型是测试报告。传统做法是直接 embedding 检索把所有语义相近的段落拉出来Agentic 做法则是先拆分意图再按操作序列逐步处理。先定义意图解析。为了演示我们写一个简单的规则解析函数实际项目中可以替换为 NER 模型或 LLM 解析。# 文件路径main.py from orchestration.planner import decide_operations_by_rule from orchestration.executor import WorkflowExecutor from operations.base import OperationContext def parse_intent(query: str) - dict: 简化版意图解析从查询中抽取时间范围、状态、实体标识。 实际项目建议使用命名实体识别模型或 LLM。 intent {} # 简单时间识别这里只是为了演示实际请用日期解析器 if 2025 年第二季度 in query or 2025 Q2 in query: intent[start_time] 2025-04-01 intent[end_time] 2025-06-30 if 已关闭 in query or 关闭 in query: intent[status] closed if 测试报告 in query: intent[doc_type] test_report return intent def main(query: str): # 1. 解析意图 intent parse_intent(query) print(识别到的意图, intent) # 2. 根据规则编排操作 operations decide_operations_by_rule(intent, query) print(编排的操作序列, [op.op_name for op in operations]) # 3. 初始化上下文假设文档库有 5 个候选文档 ctx OperationContext( queryquery, intentintent, candidate_ids[doc_1, doc_2, doc_3, doc_4, doc_5], metadata_map{ doc_1: {created_at: 2025-05-10, status: closed, title: Q2 回归测试报告}, doc_2: {created_at: 2025-03-15, status: closed, title: 3 月测试报告}, doc_3: {created_at: 2025-04-20, status: open, title: Q2 性能测试报告}, doc_4: {created_at: 2025-06-18, status: closed, title: Q2 稳定性测试报告}, doc_5: {created_at: 2024-11-30, status: closed, title: 年终测试总结}, }, ) # 4. 执行操作链 executor WorkflowExecutor() trace executor.execute(operations, ctx) # 5. 输出执行轨迹 print(\n 执行轨迹 ) for step in trace: print(step) # 6. 输出最终候选 print(\n最终候选文档, ctx.candidate_ids) if __name__ __main__: main(查询 2025 年第二季度状态为已关闭的测试报告)执行器 WorkflowExecutor 的逻辑很简单按优先级排序操作逐个执行并把每一步的 OperationResult 记录到 trace 中。# 文件路径orchestration/executor.py from operations.base import OperationContext, OperationResult class WorkflowExecutor: def __init__(self): self.trace [] def execute(self, operations, ctx: OperationContext): # 按优先级排序确保先缩小候选集再继续后续操作 operations.sort(keylambda op: op.priority) for op in operations: try: result op.execute(ctx) except Exception as e: result OperationResult( op_nameop.op_name, statusfailed, error_msgstr(e), ) self.trace.append(result) ctx.trace.append(result) if result.status failed: # 失败后是否终止链路可根据业务策略调整 print(f操作 {op.op_name} 执行失败链路提前终止) break return self.trace预期输出类似这样识别到的意图 {start_time: 2025-04-01, end_time: 2025-06-30, status: closed, doc_type: test_report} 编排的操作序列 [time_range_filter, status_filter, semantic_retrieve, rerank] 执行轨迹 操作 time_range_filter 执行完成输入 5 个候选输出 3 个候选 操作 status_filter 执行完成输入 3 个候选输出 2 个候选 操作 semantic_retrieve 执行完成输入 2 个候选输出 2 个候选 操作 rerank 执行完成输入 2 个候选输出 2 个候选 最终候选文档 [doc_1, doc_4]从这个输出可以看出最终结果是 doc_1 和 doc_4正好是第二季度内已关闭的测试报告。更重要的是系统可以明确告诉用户时间过滤去掉了 doc_2 和 doc_5状态过滤去掉了 doc_3每一步都有据可查。这就是可解释检索的价值所在。需要说明的是这个示例中的semantic_retrieve和rerank在内置的vector_search为空的实现下不会改变候选集。实际项目中你需要把vector_search替换为真实的向量数据库调用把RerankOperation替换为真实的重排模型请求。6. 常见问题与排查思路在实际应用这套设计时大家容易踩到一些共性的坑。这里整理成一张排查表方便随时查阅。问题现象常见原因解决思路最终答案为空过滤操作过多候选集被完全滤掉检查每一步操作日志定位是哪一步输出为 0为关键过滤操作增加兜底策略操作执行顺序不符合预期executor 没有按优先级排序确认 operations 已按 priority 排序或统一在 executor 内排序LLM 编排输出非法操作名提示词未给出白名单或模型输出不稳定增加操作名白名单过滤解析失败时回退到默认链路过滤条件生效但结果仍错误文档元数据缺失或格式不规范在文档写入阶段保证元数据完整性过滤操作对缺失字段要做可配置处理链路延迟明显增加操作数量变多串行执行耗时叠加在 executor 中引入并行执行能力对不依赖彼此的操作可并发执行同一个 query 多次结果不一致LLM 编排输出不稳定对强意图使用规则编排LLM 只负责弱意图开启缓存策略可解释轨迹太长用户看不懂展示粒度太细将操作按阶段聚合展示比如“意图识别 - 条件过滤 - 语义召回 - 结果确认”一个值得重点强调的问题是操作之间的依赖关系。比如状态过滤操作必须在语义召回之后做还是之前做这取决于你的数据结构和业务语义。如果状态是结构化字段建议在向量召回的过滤参数中一并传入避免先召回再过滤导致漏召。如果状态只能通过语义判断则应该在召回后增加一个专门的校验操作。这个顺序问题没有标准答案需要在项目中进行测试验证。另一个常见问题是操作失败后的降级策略。比如向量库超时是直接返回错误还是跳过该操作只返回前面过滤的结果从工程经验来看对于检索链路推荐采用“部分降级”策略如果关键过滤操作失败则终止链路如果重排操作失败则可以返回未重排的结果并在响应中标记“结果未经过重排”这样既不阻塞主流程又能保留可解释性。7. 最佳实践与工程建议把 Agentic Operations 落到生产环境不能只停留在写出几个操作类。下面这些工程建议是踩过坑之后总结出来的。第一可观测性设计要从第一天开始。每个操作不仅要返回 OperationResult还要在中间件层记录结构化日志包含操作名、耗时、入参、出参、候选集变化量。建议使用统一的 trace_id 贯穿整条链路方便后续用日志平台回放一次完整的检索过程。没有 trace_id所谓的“可解释”就失去了落地点。第二操作要支持独立测试。理想状态下每个操作都应该有独立的单元测试。测试数据不需要很大但必须覆盖三类情况正常输入、边界输入、异常输入。比如时间过滤操作要覆盖开始时间缺失、结束时间缺失、时间格式非法等情况。这样后续在操作组合时定位问题会非常快。第三权限边界要收敛。如果操作链路中涉及内部数据访问每个操作执行器必须显式声明自己访问的数据范围不允许有隐式的越权读取。建议在操作上下文中增加权限标识执行器启动时先做权限校验。尤其是“精确匹配”这类操作一旦用户上传了某个实体 ID系统需要确认该用户是否有权限查看。第四采用渐进式上线策略。不建议一次性把现有 Top-K 检索全部替换成 Agentic 链路。更稳妥的做法是先在现有链路上增加操作日志看看实际查询中有多少比例能被规则解析然后对能解析的查询切换到 Agentic 链路不能解析的继续走旧链路运行稳定后再逐步扩大转换范围。这样可以降低回归风险。第五要建立评估体系。传统 Top-K 链路通常只看 RecallK 或者 NDCG。Agentic 链路需要额外关注操作级指标比如时间过滤操作的准确率、状态过滤操作的召回率、重排后结果是否优于重排前。建议在操作结果中记录排序列表定期抽取样本进行人工标注对比新旧链路的搜索结果质量。第六缓存策略要细化。过滤类操作的结果相对稳定可以按操作输入参数做缓存语义召回操作的结果会随模型更新而变化缓存时间要短一些。缓存 key 建议包含 query 的归一化文本、操作名、操作参数和权限标识避免用户 A 的缓存结果被用户 B 看到。第七注意操作编排的可维护性。当操作数量多起来之后规则编排函数会变得很长。建议把编排规则写成配置而非代码。比如在 YAML 文件中声明当意图包含 status 时必须插入 StatusFilterOperation。这样一来产品和算法同学也能参与编排规则的维护而不需要每次都改代码。下面是一个简单的编排配置示例# 文件路径config.yaml rules: - when: intent_contains: entity_id then: - exact_match - when: intent_contains: start_time then: - time_range_filter - when: intent_contains: status then: - status_filter - always: - semantic_retrieve - rerank这种配置方式的优势很明显编排规则可视化、可审计也更容易做单元测试。你可以针对配置文件中每一条规则写测试用例确保任何情况下生成的操