第十章 基础检索服务 - 完整总结本章核心实现向量检索基础流程从用户问题到返回可引用的证据一、本章要解决的核心问题前面章节已经把文档存入 MySQL子块向量存入 Milvus但系统还不知道怎么根据用户问题找资料。用户提问时不能直接把所有文档都塞给大模型文档太多超过模型上下文长度无关内容干扰回答没有证据筛选模型容易编造幻觉检索服务的定位用户问题 → 检索服务 → 少量相关证据 → 回答服务 → 最终答案二、核心概念详解候选candidatesvs 证据evidence候选candidates向量检索召回的原始结果可能包含低分或不相关内容数量较多如 top_k8证据evidence经过过滤、分数筛选、父块扩展后的最终结果直接交给大模型使用数量较少如 evidence_top_k3为什么要区分方便调试可以看召回阶段找到什么最终用了什么方便评估分别评估召回率和精确率子块 vs 父块子块短200字信息聚焦适合向量检索精确匹配向量检索命中子块父块长完整段落/章节上下文完整不适合直接检索太长命中子块后回查父块作为最终证据设计思路子块负责找得准父块负责读得懂过滤条件RetrievalFilters字段类型作用document_idslist[str]限制只从指定文档中检索topicstr限制只检索某个主题effective_beforedate限制只检索某个日期之前生效的政策注意document_ids 必须用 default_factorylist不能用 []防止多个请求共享同一个列表对象。trace 调试信息trace 是检索链路追踪信息记录每一步的关键数据original_query原始用户问题vector_results向量检索原始召回结果final_evidence最终证据用途开发调试排查为什么没召回或为什么召回太多的问题。三、代码详解RetrievalFilters 定义from datetime import datefrom pydantic import Fieldclass RetrievalFilters(APIModel):document_ids: list[str] Field(default_factorylist)topic: str | None Noneeffective_before: date | None None【代码解释】为什么用 default_factorylist保证每个请求有独立的列表避免多个请求共享同一个可变对象为什么不用 []虽然 Pydantic 会特殊处理但可读性差复制到 dataclass 会出 bugRetrievalResult 定义dataclassclass RetrievalResult:query: strcandidates: list[dict[str, Any]]evidence: list[dict[str, Any]]refused: boolrefusal_reason: str | Nonetrace: dict[str, Any]【代码解释】为什么用 dataclass自动生成init和repr方便打印和调试为什么区分 candidates 和 evidencecandidates 是召回结果evidence 是最终证据方便分别评估元数据过滤函数 apply_metadata_filtersdef apply_metadata_filters(rows: list[dict[str, Any]],filters: RetrievalFilters) - list[dict[str, Any]]:filtered rowsif filters.topic:filtered [item for item in filteredif item[“metadata”].get(“topic”) filters.topic]if filters.effective_before:boundary filters.effective_before.isoformat()filtered [item for item in filteredif item[“metadata”].get(“effective_date”)and item[“metadata”][“effective_date”] boundary]return filtered【代码解释】为什么 topic 和 effective_before 在向量结果上过滤因为 Milvus 只支持 document_ids 过滤其他元数据过滤需要在结果上做为什么 effective_date 用字符串比较Milvus 中日期存为字符串直接用 ISO 格式比较即可父块扩展函数 expand_parent_contextasync def expand_parent_context(session: Session, evidence: list[dict[str, Any]]) - list[dict[str, Any]]:# 收集所有父块 IDparent_ids {item[“metadata”].get(“parent_id”)for item in evidenceif item[“metadata”].get(“parent_id”)}if not parent_ids:return evidence# 查询父块 parents { chunk.id: chunk for chunk in ( session.scalars( select(DocumentChunk).where(DocumentChunk.id.in_(parent_ids)) ) ).all() } # 替换为父块内容并去重 expanded: list[dict[str, Any]] [] seen_parents: set[str] set() for item in evidence: parent_id item[metadata].get(parent_id) parent parents.get(parent_id) if parent and parent.id not in seen_parents: item { **item, content: parent.content, metadata: { **item[metadata], section: parent.section or item[metadata].get(section, ), page: parent.page or item[metadata].get(page, 0), context_chunk_id: parent.id, }, } seen_parents.add(parent.id) expanded.append(item) elif not parent: expanded.append(item) return expanded【代码解释】为什么用 set 收集 parent_ids自动去重为什么用字典 parents通过 id 快速查找父块为什么用 seen_parents 去重多个子块命中同一个父块时只保留一份为什么 context_chunk_id 存父块 ID方便追溯最终证据的来源主检索函数 retrieveasync def retrieve(session: Session,question: str,filters: RetrievalFilters,top_k: int 8,evidence_top_k: int 4,min_score: float 0.25,) - RetrievalResult:# 1. 向量检索vector_results await VectorStoreService().search(question,top_k,filters.document_ids or None,)# 2. 元数据过滤 vector_results apply_metadata_filters(vector_results, filters) candidates vector_results[:top_k] # 3. 分数筛选 evidence [] for item in candidates[:evidence_top_k]: score max(0.0, min(1.0, float(item.get(score, 0)))) if score min_score: item[evidence_score] score evidence.append(item) # 4. 父块扩展 evidence await expand_parent_context(session, evidence) # 5. 拒答判断 refused not evidence # 6. 构建 trace trace { original_query: question, vector_results: vector_results, final_evidence: evidence, } return RetrievalResult( queryquestion, candidatescandidates, evidenceevidence, refusedrefused, refusal_reason检索结果未达到证据阈值 if refused else None, tracetrace, )【代码解释】为什么向量检索取 top_kevidence 取 evidence_top_k检索阶段多取一些筛选阶段再截断为什么用 max(0, min(1, score))确保分数在 0-1 范围内为什么判断 refused not evidence证据为空时不回答减少幻觉为什么 trace 要保存 vector_results 和 final_evidence方便调试对比召回和最终结果四、配置说明本章没有新增配置使用已有配置配置项默认值作用top_k8向量召回数量evidence_top_k4最终证据数量min_score0.25最低证据分数阈值五、重点难点总结重点必须掌握检索服务的作用从知识库找相关证据不是直接回答candidates 和 evidence 的区别子块召回 父块扩展的设计思路RetrievalFilters 的三个过滤条件trace 的作用和内容难点理解原理为什么用子块检索、父块回答Pydantic 中 default_factorylist 和 [] 的区别父块扩展中去重的原因生产实践建议min_score 阈值调优建议用测试集调整找到精确率和召回率的平衡点trace 日志生产环境建议把 trace 写入日志方便排查问题监控指标关注拒答率太高说明知识库覆盖不足六、面试题预测Q1: 检索服务在整个 RAG 系统中扮演什么角色参考答案检索服务是 RAG 系统的第一步负责从知识库中找到与用户问题相关的证据。它把检索结果交给后续的回答服务生成答案。如果检索找不到证据系统应该拒答而不是强行编造。Q2: 为什么用子块做检索却用父块作为最终证据参考答案子块更短、更聚焦适合向量检索召回精度高。父块更长、上下文完整适合给大模型生成回答。子块负责找得准父块负责读得懂。Q3: candidates 和 evidence 有什么区别参考答案candidates 是向量检索召回的原始候选结果数量较多可能包含低分项。evidence 是经过分数筛选和父块扩展后的最终证据数量较少直接给大模型使用。Q4: trace 字段的作用是什么参考答案trace 是检索链路的调试信息记录原始问题、向量召回结果、最终证据等。方便开发人员排查为什么没召回或为什么召回太多的问题。Q5: 证据为空时为什么要拒答参考答案如果证据为空说明知识库中没有与问题相关的内容。强行让大模型回答容易编造内容幻觉。拒答并提示根据现有资料无法回答该问题更安全。七、自测清单能说出检索服务的作用能说出 candidates 和 evidence 的区别能解释为什么用子块检索、父块回答能说出 RetrievalFilters 的三个过滤条件能解释 default_factorylist 和 [] 的区别能说出 trace 包含哪些信息能解释证据为空时为什么要拒答能画出检索服务的完整流程八、完整流程图用户问题↓① 生成查询向量VectorStoreService.search↓② Milvus 向量召回top_k 个候选↓③ 元数据过滤topic / effective_before↓④ 分数阈值过滤score min_score↓⑤ 回查父块上下文expand_parent_context↓⑥ 证据为空→ 拒答refused True↓⑦ 返回 RetrievalResult