知识库塞满数据但大模型答不上来——RAG 基础检索焊死，向量召回+过滤+父块扩展一条龙

本文基于 Python 3.12 + Milvus + SQLAlchemy 编写。核心检索逻辑均经过验证；代码依赖项目已实现的 VectorStoreService / DocumentChunk / SessionLocal 等模块。前置条件：文档已写入 MySQL，子块向量已写入 Milvus。

你是不是也卡在这一步

文档上传了，分块做了，向量也写了，MySQL 和 Milvus 里数据都齐了。然后呢？

用户问了一句"小微企业补贴支持哪些范围"，系统直接把所有文档塞给大模型——

  • 文档太多，token 直接超限，模型报错。
  • 无关内容太多，模型被干扰，答得驴唇不对马嘴。
  • 没有证据筛选，模型瞎编一通，还信誓旦旦。

问题不在模型，在于你压根没做检索。 知识库有数据不代表大模型能用上。你得先从几百分块里挑出最相关的几条，过滤掉垃圾，补全上下文，再把干干净净的证据喂给模型。

这篇文章就是把 RAG 基础检索流程从头到尾焊死：向量召回 → 元数据过滤 → 父块扩展 → 分数截断 → 拒答兜底，一条龙跑通。

说明：BM25 关键词检索、混合检索、RRF 融合和重排序不在本章范围，后续单独写。

检索服务到底在干嘛

先看数据存在哪：

MySQL                    Milvus
├── 文档信息              └── 子块向量
├── 父块
└── 子块

MySQL 存结构化数据（文档元信息、父块、子块），Milvus 只存子块向量。

检索服务做的事，一句话概括：用户提问 → 生成查询向量 → Milvus 向量召回 → 过滤不满足条件的 → 回查父块上下文 → 返回候选证据。

是

否

用户问题

生成查询向量

Milvus 向量召回

元数据过滤

分数截断

父块扩展

证据为空?

返回拒答标记

返回候选证据

有个比喻：检索服务像图书馆管理员。你跟他说"我要小微企业补贴的资料"，他不会把整个图书馆搬给你，而是先去索引柜（Milvus）查哪几页最相关，再把那几页所在的完整章节（父块）抽出来递给你。如果翻了一圈啥也没找到，他会直接说"没有相关资料"（拒答），而不是随便拿本书糊弄你。

涉及三个文件：

文件作用
backend/app/schemas.py定义过滤条件和结果结构
backend/app/services/retrieval.py检索服务核心逻辑
backend/scripts/check_basic_retrieval.py测试脚本

第一步：定义过滤条件 RetrievalFilters

文件名：backend/app/schemas.py

用户提问时，经常需要限制检索范围。比如只查某几篇文档、只查某个主题、只查某个日期之前生效的政策。

from datetime import date
from pydantic import BaseModel, Field


class RetrievalFilters(BaseModel):
    document_ids: list[str] = Field(default_factory=list)
    topic: str | None = None
    effective_before: date | None = None
字段作用
document_ids限制只从指定文档中检索
topic限制只检索某个主题
effective_before限制只检索某个日期之前生效的政策

这个类在 API 接口和 service 中频繁用于传参，是检索模块的"入场券"。

为什么用 Field(default_factory=list) 而不是 =[]

这是个经典 Python 坑。list[str] = [] 看起来没问题，但列表是可变对象——类定义阶段只创建一个全局列表，所有实例共享同一个：

from dataclasses import dataclass, field

# 写法 1：推荐（显式 default_factory，每次新建实例得到独立列表）
@dataclass
class ModelA:
    document_ids: list[str] = field(default_factory=list)

# 写法 2：不推荐（原生类里把列表写成类属性，被所有实例共享）
class ModelB:
    document_ids = []

b1 = ModelB()
b2 = ModelB()
b1.document_ids.append("doc1")

print(b1.document_ids)  # ['doc1']
print(b2.document_ids)  # ['doc1']  b2 也被污染了（共享同一个类属性）

b2 明明啥也没干，document_ids 里却多了个 "doc1"。这就是共享可变对象 bug。

Pydantic v2 对可变默认值（=[]）会自动做深拷贝，每个实例拿到独立列表，所以上面这段在 Pydantic 里实际上不会出问题。但如果你用原生类把列表写成类属性（= []），=[] 就一定会踩共享坑；而原生 dataclass + = [] 会直接报错拦下（写不出来）。所以不管用什么框架，统一写 Field/field(default_factory=list) 最安全，一眼看懂意图，换到任何场景都不会出错。

场景=[]Field(default_factory=list)
Pydantic v2底层自动转换，不会出 bug推荐，可读性最好
原生类（类属性 = []）所有实例共享，必出污染用 default_factory 或在 __init__ 内初始化
原生 dataclass（= []）直接报错，写不出来必须用 field(default_factory=list)
团队阅读容易让人疑惑"会不会共享"一眼看懂，无歧义

第二步：定义结果结构 RetrievalResult

文件名：backend/app/services/retrieval.py

检索完得有个东西把结果装好。用 dataclass 定义：

from dataclasses import dataclass
from typing import Any


@dataclass
class RetrievalResult:
    query: str
    candidates: list[dict[str, Any]]
    evidence: list[dict[str, Any]]
    refused: bool
    refusal_reason: str | None
    trace: dict[str, Any]
字段作用
query实际用于检索的查询文本
candidates初步召回的候选结果
evidence最终交给回答服务的证据
refused是否拒答
refusal_reason拒答原因
trace调试信息

这里有个关键设计：为什么区分 candidates 和 evidence？

  • candidates：向量检索拿到的原始候选，可能有很多条，可能有垃圾。
  • evidence：经过过滤、分数截断、父块扩展后，真正能喂给大模型的证据。

区分它们的好处是后续做评估时，可以分别看"召回阶段找没找到"和"最终证据能不能用"。如果召回找到了但最终证据为空，说明是过滤或分数阈值的问题；如果召回就没找到，说明是向量模型或分块的问题。

第三步：写核心入口 retrieve 函数

继续在 retrieval.py 中写。这是整个检索服务的入口函数：

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import DocumentChunk
from app.schemas import RetrievalFilters
from app.services.vector_store import VectorStoreService


async 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)
    # 过滤在取 top_k 之前执行，过滤可能减少候选数（不会补召回）
    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
    trace = {
        "original_query": question,
        "vector_results": vector_results,
        "final_evidence": evidence,
    }
    return RetrievalResult(
        query=question,
        candidates=candidates,
        evidence=evidence,
        refused=refused,
        refusal_reason="检索结果未达到证据阈值" if refused else None,
        trace=trace,
    )
参数作用
question用户问题
filters检索过滤条件
top_k向量召回数量（默认 8）
evidence_top_k最终证据数量上限（默认 4）
min_score最低证据分数（默认 0.25）

五步拆解：

  1. 向量检索：把问题文本传给 Milvus，返回 top_k 条最相似的子块。
  2. 元数据过滤：按主题、生效日期再筛一遍。
  3. 分数截断：分数低于 min_score 的直接踢掉。这是减少幻觉的第一道防线——分数太低说明跟问题不相关，喂给模型只会添乱。
  4. 父块扩展：子块短但准，父块长但全。命中子块后回查父块，补全上下文。
  5. 拒答判断：如果证据为空，标记拒答，告诉后面的回答服务"别让大模型自由发挥"。

min_score 的值需要根据实际向量模型调。0.25 是一个偏严格的起点（阈值偏高，召回偏少），太高会漏掉相关结果，太低会放进垃圾。建议先用 0.1 跑通，再逐步上调。

分数截断里的一行代码

score = max(0.0, min(1.0, float(item.get("score", 0))))

这行做了三件事：

  • item.get("score", 0)：从结果里取 score 字段，没有就给 0。
  • float(...)：转成浮点数，防止字符串混进来。
  • max(0.0, min(1.0, ...))：把分数夹在 0 到 1 之间。Milvus 返回的相似度有时会略超 1 或为负数，这步是安全兜底。

第四步：写元数据过滤函数

Milvus 的元数据中，生效日期保存成字符串。所以日期过滤可以在向量结果上直接做：

def apply_metadata_filters(
    rows: list[dict[str, Any]], filters: RetrievalFilters
) -> list[dict[str, Any]]:
    filtered = rows
    if filters.topic:
        filtered = [
            item for item in filtered
            if item["metadata"].get("topic") == filters.topic
        ]
    if filters.effective_before:
        boundary = filters.effective_before.isoformat()
        filtered = [
            item for item in filtered
            if item["metadata"].get("effective_date")
            and item["metadata"]["effective_date"] <= boundary
        ]
    return filtered

逻辑很直白：有 topic 就按主题筛，有 effective_before 就按日期筛。两个都有就依次筛两遍。

document_ids 没在这个函数里处理，因为它是直接传给 Milvus 的——Milvus 原生支持按元数据过滤文档 ID，在向量检索阶段就过滤了，不需要拿到 Python 里再筛。

为什么日期用字符串比较？因为 Milvus 元数据里的 effective_date 存的是 isoformat() 字符串（如 "2024-06-01"），ISO 格式的日期字符串天然支持字典序比较，"2024-06-01" <= "2024-12-31" 直接成立。前提是日期格式必须统一为 YYYY-MM-DD，如果混入 "2024/6/1" 这种就会出错。

第五步：写父块扩展函数

向量检索命中的是子块。子块短、定位准，但上下文可能不够完整。

比如用户问"补贴标准"，命中的子块可能只有一句"补贴金额为每年 5 万元"。但完整的补贴标准包括申请条件、发放方式、有效期等，这些都在父块里。

所以命中子块后，要根据 parent_id 回到 MySQL 找父块，用父块内容替换子块内容：

async def expand_parent_context(
    session: Session, evidence: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    # 1. 收集所有父块 ID
    parent_ids = {
        item["metadata"].get("parent_id")
        for item in evidence
        if item["metadata"].get("parent_id")
    }
    if not parent_ids:
        return evidence

    # 2. 批量查询父块
    parents = {
        chunk.id: chunk
        for chunk in (
            session.scalars(
                select(DocumentChunk).where(DocumentChunk.id.in_(parent_ids))
            )
        ).all()
    }

    # 3. 用父块内容替换子块内容，去重
    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

三个细节：

parent_ids 用集合不用列表。 多个子块可能命中同一个父块，集合自动去重，避免重复查询。

seen_parents 去重。 如果两个子块命中同一个父块，父块只需要加入一次。seen_parents 记录已经处理过的父块 ID，遇到重复的直接跳过。

context_chunk_id 保存父块 ID。 最终证据里的 context_chunk_id 指向父块，方便后续溯源——如果模型回答有问题，可以反查是哪个父块提供的上下文。

字典解包语法补充

代码里 **item 和 **item["metadata"] 用的是 Python 字典解包：

# * 解包列表/元组
def add(a, b, c):
    print(a + b + c)

nums = [1, 2, 3]
add(*nums)  # 等价 add(1, 2, 3)

# ** 解包字典
def show(name, age):
    print(name, age)

info = {"name": "张三", "age": 20}
show(**info)  # 等价 show(name="张三", age=20)

在父块扩展里，{**item, "content": parent.content} 的意思是：把 item 字典的所有键值对展开，再覆盖 content 字段。同理 **item["metadata"] 展开原元数据，再追加 section、page、context_chunk_id。

拒答和 trace 是什么

拒答

refused = not evidence
refusal_reason = "检索结果未达到证据阈值" if refused else None

这行用的是 Python 三元表达式：值1 if 条件 else 值2，条件为 True 返回值 1，为 False 返回值 2。

拒答不是最终回答文本。它只是告诉后面的回答服务：没有找到可靠证据，不应该让大模型自由发挥。 后面的回答服务拿到 refused=True 后，可以直接返回"抱歉，知识库中没有找到相关资料"，而不是让模型瞎编。

这是 RAG 系统防幻觉的第二道防线（第一道是分数截断）。

trace

trace = {
    "original_query": question,
    "vector_results": vector_results,
    "final_evidence": evidence,
}

trace（直译：追踪、链路日志、调试埋点）是检索模块的内部链路追踪信息，不给前端用户看，专供开发和排查问题用。

key含义
original_query原始用户提问，防止后续 query 被改写后找不到用户原本输入
vector_results向量数据库原始召回结果，未经过滤和扩展
final_evidence经过过滤、扩展后的最终证据

如果检索效果不好，看 trace 就能定位问题出在哪一步：vector_results 有没有命中？命中后过滤掉了多少？扩展后证据数量对不对？

完整代码

文件名：backend/app/services/retrieval.py

from dataclasses import dataclass
from typing import Any

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import DocumentChunk
from app.schemas import RetrievalFilters
from app.services.vector_store import VectorStoreService


@dataclass
class RetrievalResult:
    query: str
    candidates: list[dict[str, Any]]
    evidence: list[dict[str, Any]]
    refused: bool
    refusal_reason: str | None
    trace: dict[str, Any]


def apply_metadata_filters(
    rows: list[dict[str, Any]], filters: RetrievalFilters
) -> list[dict[str, Any]]:
    filtered = rows
    if filters.topic:
        filtered = [
            item for item in filtered
            if item["metadata"].get("topic") == filters.topic
        ]
    if filters.effective_before:
        boundary = filters.effective_before.isoformat()
        filtered = [
            item for item in filtered
            if item["metadata"].get("effective_date")
            and item["metadata"]["effective_date"] <= boundary
        ]
    return filtered


async def expand_parent_context(
    session: Session, evidence: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    parent_ids = {
        item["metadata"].get("parent_id")
        for item in evidence
        if 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


async def retrieve(
    session: Session,
    question: str,
    filters: RetrievalFilters,
    top_k: int = 8,
    evidence_top_k: int = 4,
    min_score: float = 0.25,
) -> RetrievalResult:
    vector_results = await VectorStoreService().search(
        question,
        top_k,
        filters.document_ids or None,
    )
    vector_results = apply_metadata_filters(vector_results, filters)
    # 过滤在取 top_k 之前执行，过滤可能减少候选数（不会补召回）
    candidates = vector_results[:top_k]

    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)

    evidence = await expand_parent_context(session, evidence)
    refused = not evidence
    trace = {
        "original_query": question,
        "vector_results": vector_results,
        "final_evidence": evidence,
    }
    return RetrievalResult(
        query=question,
        candidates=candidates,
        evidence=evidence,
        refused=refused,
        refusal_reason="检索结果未达到证据阈值" if refused else None,
        trace=trace,
    )

项目最终版本会继续在这个文件中增加 BM25、RRF、重排序和证据过滤。但这一章先把基础向量检索流程跑通。

测试脚本

文件名：backend/scripts/check_basic_retrieval.py

import asyncio
import sys
from pathlib import Path


BACKEND_ROOT = Path(__file__).resolve().parents[1]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))

from app.database import SessionLocal
from app.schemas import RetrievalFilters
from app.services.retrieval import retrieve


async def main() -> None:
    question = "小微企业数字化补贴支持哪些范围"
    filters = RetrievalFilters()

    with SessionLocal() as session:
        result = await retrieve(
            session=session,
            question=question,
            filters=filters,
            top_k=5,
            evidence_top_k=3,
            min_score=0.1,
        )

    print("原始问题：", question)
    print("检索用查询：", result.query)
    print("是否拒答：", result.refused)
    print("拒答原因：", result.refusal_reason)
    print("候选数量：", len(result.candidates))
    print("证据数量：", len(result.evidence))

    print("候选结果：")
    for item in result.candidates[:5]:
        print(
            {
                "chunk_id": item.get("chunk_id"),
                "score": round(float(item.get("score", 0)), 4),
                "source": item.get("source"),
                "filename": item.get("metadata", {}).get("filename"),
                "section": item.get("metadata", {}).get("section"),
                "content_preview": item.get("content", "")[:60],
            }
        )

    print("最终证据：")
    for item in result.evidence:
        print(
            {
                "chunk_id": item.get("chunk_id"),
                "evidence_score": round(float(item.get("evidence_score", 0)), 4),
                "context_chunk_id": item.get("metadata", {}).get("context_chunk_id"),
                "content_preview": item.get("content", "")[:80],
            }
        )


if __name__ == "__main__":
    asyncio.run(main())

脚本做的事：创建一个问题 → 调用检索服务 → 打印候选数量 → 打印证据数量 → 展示命中文档、章节、分数和内容片段。

进入 backend 目录执行：

.\.venv\Scripts\python.exe scripts\check_basic_retrieval.py

预期输出类似：

原始问题： 小微企业数字化补贴支持哪些范围
检索用查询： 小微企业数字化补贴支持哪些范围
是否拒答： False
拒答原因： None
候选数量： 5
证据数量： 3
候选结果：
{'score': 0.7312, 'source': 'vector', 'filename': '星河市小微企业数字化补贴细则.txt', ...}
最终证据：
{'evidence_score': 0.7312, 'context_chunk_id': '...', 'content_preview': '第二条 支持范围...'}

如果 证据数量 是 0，一般有三种可能：

  • 还没有上传文档。
  • 已上传文档但 Milvus 没有向量。
  • min_score 设置过高。

可以先运行文档写入脚本确认 MySQL 和 Milvus 中都有数据。

排坑表

坑现象原因解决
apply_metadata_filters 忘了同步/异步一致TypeError: object list can't be used in 'await' expression函数定义成 async 但调用时没 await，或者定义成同步但调用时 await 了统一：过滤函数用同步（不需要 IO），retrieve 里调用时不加 await
日期过滤不生效明明有符合条件的数据，过滤后全空Milvus 元数据里 effective_date 格式不统一，比如有 "2024-06-01" 也有 "2024/6/1"写入时统一用 isoformat()，保证 YYYY-MM-DD 格式
父块查不到expand_parent_context 返回的 evidence 里 content 没变parent_id 在子块元数据里不存在或写错了检查写入分块时 metadata 里是否正确写入了 parent_id
同一个父块重复出现evidence 数量比预期多seen_parents 去重逻辑没生效，或者父块 ID 不一致确认 seen_parents.add(parent.id) 在正确位置
min_score 设太高召回有结果但证据为空，触发拒答分数阈值过高于向量模型实际输出范围先用 0.1 跑通看分数分布，再逐步上调
向量检索返回 NoneTypeError: 'NoneType' is not subscriptableMilvus 连接失败或 collection 不存在检查 Milvus 连接配置和 collection 是否已创建
score 字段不存在所有结果 score 为 0，全部被截断VectorStoreService 返回的字典里 score 字段名不一致检查 VectorStoreService().search() 返回结构，确认字段名

什么情况不该用

文档量很小（几十条以下）不需要检索。 如果知识库只有十几条 FAQ，直接全量塞给大模型就行。引入向量检索反而增加维护成本（Milvus 部署、向量同步、索引调优），得不偿失。

需要精确匹配的场景不适合纯向量检索。 比如查订单号、查合同编号、查手机号——这些是精确匹配需求，向量检索的语义相似度反而可能召回错误结果。这类场景应该用 MySQL 直接查，或者等后续 BM25 关键词检索加上后用混合检索。

对实时性要求极高的场景要谨慎。 向量检索有延迟（生成查询向量 + Milvus 检索 + 父块回查），如果要求毫秒级响应，需要做缓存或预计算。

父块扩展会增加 token 消耗。 父块比子块大得多（通常 5~10 倍），扩展后喂给大模型的上下文会显著增加。如果模型上下文窗口有限，需要控制 evidence_top_k 或在父块扩展后做二次截断。

# adb
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区