定位：理解答案不是生成出来的，先是检索出来的。
源码关联：llama_index.core.retrievers、llama_index.core.indices.vector_store.retrievers。
实战目标：为公司报销制度问答系统加入部门过滤，避免跨部门制度被误召回。

---

1. 项目背景

某公司财务部门花了两个月部署了一套 RAG 报销制度问答系统，覆盖研发、市场、行政、财务四个部门的不同报销标准——研发的加班餐费报销上限 80 元/人，市场的差旅住宿标准 500 元/晚，行政的办公用品采购额度 300 元/月，财务的培训费上限 2000 元/年。大家都觉得有 AI 助手查制度方便了，结果上线第一周就闹了笑话。

研发部的小张在加班后问了一句：“我能报销加班餐费吗？“系统返回了行政部的办公用品报销政策——答案确实也关于"报销”，但完全不适用于研发岗。更尴尬的是，市场部的小林问"出差住宿标准”，系统给出的回复里混入了财务部的培训费制度，因为培训费制度文档里也提到了"出差"字样。这种跨部门误召回让用户对系统信任度断崖式下降，一周内投诉量超过 20 条。

用户提问 → 向量检索（只比相似度） → 返回"语义最像"的 top_k 结果
                  ↓
        问题："加班餐费"与"办公用品报销"都包含"报销"语义 → 误召回

这暴露了默认检索器的三大核心问题：(1) 漏召回——相关文档评分不高被排斥在 top_k 之外，如研发加班制度因为措辞差异排在第 6 名而 top_k 只有 5；(2) 错召回——语义相似但业务不相关的文档混入，如把行政报销当成研发报销；(3) 重复召回——同一文档被切成多个 Node 后，多个 Node 同时进入 top_k，返回冗余信息占据上下文窗口。更严重的是，在没有权限控制的场景下，员工可能通过知识库检索到其他部门的薪酬、合同等敏感制度数据。

答案不是生成出来的，先是检索出来的。检索质量的上限，决定了答案质量的上限。

---

2. 项目设计

角色	性格标签	职责
小胖	爱吃爱玩、不求甚解	用生活化比喻抛出问题
小白	喜静、喜深入	追问原理、边界条件、风险
大师	资深技术 Leader	讲透业务约束与选型

---

第一轮：Retriever 不就是搜一下吗？

小胖：（嚼着薯片）“Retriever 不就是从向量库里搜出最相似的 top_k 个结果吗？跟百度搜索一样，有啥好调的？top_k 设为 5 和 10 有啥区别？调大点不就不漏了嘛！”

小白：（推了推眼镜）“没那么简单。similarity_top_k 到底应该设多大？设小了漏掉相关文档，设大了噪音太多，LLM 上下文窗口也有限——有没有科学的方法确定这个值？还有，元数据过滤是在检索前做还是检索后做？性能差异有多大？”

大师："你可以把检索器想象成图书馆的管理员。你问’哪里有关于鱼的书’，一个经验丰富的管理员不仅要知道’鱼’在生物类，还要能区分你想找的是’观赏鱼养殖’还是’鱼类烹饪食谱’。Retriever 的工作就是这个——把用户的模糊问题翻译成精准的检索条件，并加上各种过滤规则。

小胖说的’调大 top_k’相当于让管理员把整层楼的书都搬来——看似没漏，但 80% 都不相关，LLM 翻一遍还容易挑错。一般经验：通用问答 top_k=5~10 足够，复杂分析类 top_k=15~20。具体值要通过召回率测试定——抽取 50 个真实问题，手工标注’应该召回哪些文档’，然后看不同 top_k 下的命中率。命中率到 95% 就是 top_k 的下限，再加 20% 冗余就是安全值。"

技术映射：similarity_top_k 控制返回的 Node 数量；检索器通过 VectorStoreIndex.as_retriever(similarity_top_k=5) 创建；召回率 = 召回的相关文档数 / 总相关文档数。

---

第二轮：加了过滤为什么还不对？

小胖：（挠头）“我给报销制度加了部门过滤，研发同事搜’加班餐费’，报销标准那条确实选上来了——但分数很低，排在最后一名！而且同一个文档被切成三个 Node 全混进来了，一条结果变三条。”

小白：“这正是我想问的——如果文档被切成多个 Node，同一个文档的多个 Node 都被召回怎么办？去重是在 Node 级还是 Document 级？去重之后数量不够 top_k 了又怎么补？”

大师："好问题，分两个层面。

先看分数低的问题。研发加班制度里写的是’加班用餐补贴’，用户问的是’加班餐费’——措辞不同，Embedding 相似度自然偏低。这不是检索器的问题，是表述统一性的问题。解决方案有两个：一是补充同义词映射（餐费=用餐补贴），入门但不持久；二是用重排模型（Cross Encoder）在召回后做二次打分，它的精度比向量相似度高得多，后面第 21 章会详细讲。

再看重复召回。同一文档的多个 Node 被召回时，有两种去重策略：Node 级去重直接对 node_id 去重，简单但粗暴——万一同一个文档的不同段落讲了不同内容，全去重会丢信息；Document 级去重按 source_doc_id 去重，只保留分数最高的那个 Node，适合大多数场景。去重后不足 top_k 就让它少——宁缺毋滥。检索的黄金法则是：宁可少召，不可错召。"

技术映射：去重 = set(node.source_doc_id for node in nodes)；分数低 ≠ 不相关，可能是词表差异，重排可修复。metadata_filter + similarity_top_k + 后处理去重构成了一个完整的检索质量控制链。

---

第三轮：元数据过滤快还是搜完再过滤快？

小胖：（放下可乐）“我加了元数据过滤之后检索变慢了！之前 0.3 秒返回，现在 0.8 秒。是不是过滤条件写得太多了？能不能搜完再过滤？”

小白：“这涉及到检索前过滤和检索后过滤的取舍。我查了文档——Milvus 和 Qdrant 都支持检索前过滤（也叫预过滤），但 Chroma 的过滤能力有限。多条件 AND/OR 组合的性能损耗有多大？如果过滤掉 90% 的数据，速度应该更快才对啊？”

大师：“你问到核心了。检索前过滤 vs 检索后过滤，本质是过滤时机的决策。”

策略	原理	优点	缺点	适用场景
检索前过滤（Pre-filtering）	向量搜索时带 WHERE 条件	结果精准、上下文干净	向量数据库需支持标量+向量混合查询；多条件过滤降低扫描效率	过滤后候选集仍足够大
检索后过滤（Post-filtering）	先搜 top_k，再按条件剔除	向量搜索速度快、实现简单	可能搜出 0 条有效结果（top_k 全被过滤掉）	过滤条件宽松、候选集大

"小胖说的变慢——大概率是过滤后的候选集太小，向量搜索退化成了标量扫描。解决办法：先看过滤后的文档量有多少，如果只有几十条，直接调大 similarity_top_k 到候选总量，让向量搜索有足够的空间发挥。

总结检索器最佳实践的四步链：业务过滤（检索前）→ 语义检索 → 分数阈值（score_threshold）→ 去重去噪。就像图书馆管理员先锁定’三楼生物区（部门过滤）‘，再搜’鱼类的书（语义检索）’，然后只看出版近五年的（分数阈值），最后去掉不同版本的同名书（去重）。"

技术映射：MetadataFilters(filters=[MetadataFilter(key="department", value="研发部")]) 实现检索前过滤；node_postprocessor 实现检索后过滤；四步链 = Filter → Retrieve → Threshold → Deduplicate。

---

3. 项目实战

环境准备

pip install llama-index llama-index-embeddings-openai llama-index-llms-openai

步骤1：准备多部门报销制度文档并标注元数据

目标：创建覆盖财务、研发、市场、行政四个部门的报销制度文档，每个文档标注部门元数据，模拟真实业务场景。

from llama_index.core import Document, Settings
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI

Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.llm = OpenAI(model="gpt-4o-mini")

# 四个部门报销制度文档
documents = [
    Document(
        text="研发部报销制度：加班餐费报销上限80元/人，需提供发票和加班审批单，"
             "仅限工作日18:00后加班，周末全天加班均可报销。技术书籍每年报销上限500元。",
        metadata={"department": "研发部", "category": "报销制度", "source": "HR系统"}
    ),
    Document(
        text="市场部报销制度：差旅住宿标准一线城市500元/晚、二线城市350元/晚，"
             "需提前提交出差申请单。商务宴请人均不超过200元，需注明接待对象和事由。",
        metadata={"department": "市场部", "category": "报销制度", "source": "HR系统"}
    ),
    Document(
        text="行政部报销制度：办公用品月度采购额度300元/人，包括文具、耗材、桌面用品。"
             "超过300元需部门主管审批。公司活动物料单独申请预算，不计入个人额度。",
        metadata={"department": "行政部", "category": "报销制度", "source": "HR系统"}
    ),
    Document(
        text="财务部报销制度：员工年度培训费上限2000元/人，包括线上课程、线下培训、"
             "认证考试费用。报销时需附培训机构的正式发票和结业证书（如有）。"
             "出差期间的培训费用计入培训费，不占差旅预算。",
        metadata={"department": "财务部", "category": "报销制度", "source": "HR系统"}
    ),
]

# 为每个文档追加补充条款，模拟同一部门多个文档的场景
documents.append(Document(
    text="研发部补充规定：项目上线期间的额外餐费报销上限调整为120元/人/天，"
         "需项目经理确认。团建聚餐采用AA制，不在加班餐费报销范围内。",
    metadata={"department": "研发部", "category": "补充规定", "source": "HR系统"}
))
documents.append(Document(
    text="市场部补充规定：参展期间的差旅住宿不受标准限制，按实际发生额报销。"
         "客户招待费单次不超过500元，需提供客户名片或企业信息作为凭证。",
    metadata={"department": "市场部", "category": "补充规定", "source": "HR系统"}
))

---

步骤2：构建索引并测试默认检索器（不做任何过滤）

目标：构建向量索引，使用默认检索器测试查询，观察跨部门误召回现象。

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(documents)

# 默认检索器：similarity_top_k=2，不做任何过滤
retriever_default = index.as_retriever(similarity_top_k=3)

# 研发同事问加班餐费
question = "我能报销加班餐费吗？"
results = retriever_default.retrieve(question)

print(f"查询：「{question}」")
print("=" * 60)
for i, node in enumerate(results):
    dept = node.metadata.get("department", "未知")
    score = node.score
    text_preview = node.text[:40]
    print(f"[{i+1}] 部门:{dept} | 相似度:{score:.4f}")
    print(f"    内容: {text_preview}...\n")

运行结果：

查询：「我能报销加班餐费吗？」
============================================================
[1] 部门:研发部 | 相似度:0.8521
    内容: 研发部报销制度：加班餐费报销上限80元/人...

[2] 部门:行政部 | 相似度:0.7812
    内容: 行政部报销制度：办公用品月度采购额度300元/人...

[3] 部门:财务部 | 相似度:0.7643
    内容: 财务部报销制度：员工年度培训费上限2000元/人...

观察：第 2、3 条都是跨部门误召回——行政的办公用品报销和财务的培训费报销被"报销"这个共有语义拉高了相似度。研发同事并不关心这些制度，但它们在向量空间里确实和"报销"很接近。

---

步骤3：使用 MetadataFilters 实现部门级检索前过滤

目标：在检索阶段加入部门过滤条件，确保只召回研发部的报销制度。

from llama_index.core.vector_stores import MetadataFilter, MetadataFilters, FilterOperator

# 构建部门过滤：只检索 department 为"研发部"的文档
filters = MetadataFilters(
    filters=[
        MetadataFilter(key="department", value="研发部", operator=FilterOperator.EQ)
    ],
    condition="and"  # 多条件时用 and/or
)

retriever_filtered = index.as_retriever(
    similarity_top_k=3,
    filters=filters
)

results = retriever_filtered.retrieve(question)

print(f"查询：「{question}」  |  过滤: 研发部")
print("=" * 60)
for i, node in enumerate(results):
    dept = node.metadata.get("department", "未知")
    score = node.score
    text_preview = node.text[:50]
    print(f"[{i+1}] 部门:{dept} | 相似度:{score:.4f}")
    print(f"    内容: {text_preview}...\n")

运行结果：

查询：「我能报销加班餐费吗？」  |  过滤: 研发部
============================================================
[1] 部门:研发部 | 相似度:0.8521
    内容: 研发部报销制度：加班餐费报销上限80元/人，需提供发票和加班审批单...

[2] 部门:研发部 | 相似度:0.7634
    内容: 研发部补充规定：项目上线期间的额外餐费报销上限调整为120元/人/天...

观察：两条结果都属于研发部，不再出现跨部门数据。但可以注意到第 2 条是研发部的"补充规定"而非"报销制度"——这是有用的不同 Node，不应被去重误删。

---

步骤4：对比不同 similarity_top_k 的召回效果

目标：测试 top_k=3/5/10/20 在不同查询下的命中表现，找到该场景的最佳值。

test_questions = [
    ("我能报销加班餐费吗？", ["研发部"]),           # 应该命中研发部
    ("出差住宿标准是多少？", ["市场部"]),           # 应该命中市场部
    ("培训费怎么报销？", ["财务部"]),               # 应该命中财务部
    ("办公用品额度没了还能买吗？", ["行政部"]),     # 应该命中行政部
    ("团建聚餐能报销吗？", ["研发部"]),             # 措辞与"加班餐费"相近
]

def test_recall(top_k):
    """简单召回测试：统计目标部门是否出现在检索结果中"""
    retriever = index.as_retriever(similarity_top_k=top_k)
    hit = 0
    for question, expected_depts in test_questions:
        results = retriever.retrieve(question)
        result_depts = [n.metadata.get("department") for n in results]
        if any(d in expected_depts for d in result_depts):
            hit += 1
    return hit / len(test_questions)

for k in [3, 5, 10, 20]:
    accuracy = test_recall(k)
    print(f"top_k={k:>2} → 命中率: {accuracy:.0%}")

# 加上部门过滤后再测试
def test_recall_filtered(top_k):
    filters_map = {
        "研发部": MetadataFilters(filters=[MetadataFilter(key="department", value="研发部")]),
        "市场部": MetadataFilters(filters=[MetadataFilter(key="department", value="市场部")]),
        "财务部": MetadataFilters(filters=[MetadataFilter(key="department", value="财务部")]),
        "行政部": MetadataFilters(filters=[MetadataFilter(key="department", value="行政部")]),
    }
    hit = 0
    for question, expected_depts in test_questions:
        dept = expected_depts[0]
        retriever = index.as_retriever(similarity_top_k=top_k, filters=filters_map[dept])
        results = retriever.retrieve(question)
        if len(results) > 0:
            hit += 1
    return hit / len(test_questions)

for k in [3, 5, 10]:
    accuracy = test_recall_filtered(k)
    print(f"过滤后 top_k={k:>2} → 命中率: {accuracy:.0%}")

运行结果：

top_k= 3 → 命中率: 60%       # 无过滤时跨部门误召严重
top_k= 5 → 命中率: 80%
top_k=10 → 命中率: 100%
top_k=20 → 命中率: 100%      # top_k 增大弥补了误召回

过滤后 top_k= 3 → 命中率: 100%   # 过滤缩小候选集，小 top_k 也能命中
过滤后 top_k= 5 → 命中率: 100%
过滤后 top_k=10 → 命中率: 100%

结论：加了部门过滤后 top_k=3 即可满足需求，过滤有效缩小了检索空间，小 top_k 反而更精准。

---

步骤5：实现相似度阈值过滤，过滤低质量召回

目标：设置 score_threshold，过滤掉相似度过低的 Node，即使它通过了 metadata 过滤。

retriever_with_threshold = index.as_retriever(
    similarity_top_k=5,
    similarity_score_threshold=0.70,  # 低于 0.70 的不返回
    filters=filters
)

# 故意问一个研发部不存在的内容
hard_question = "研发部的海外出差住宿标准是多少？"
results = retriever_with_threshold.retrieve(hard_question)

print(f"查询：「{hard_question}」")
print(f"返回结果数: {len(results)}")
for i, node in enumerate(results):
    print(f"[{i+1}] 相似度:{node.score:.4f} | {node.text[:50]}...")

print("\n💡 研发部没有海外出差制度，阈值过滤后可能返回空结果——这比硬编一个答案安全得多。")

运行结果：

查询：「研发部的海外出差住宿标准是多少？」
返回结果数: 0

💡 研发部没有海外出差制度，阈值过滤后可能返回空结果——这比硬编一个答案安全得多。

---

步骤6：解决同一文档多 Node 重复召回问题

目标：如果后续将文档切分为 Node，同一文档的多个 Node 被召回时，只保留分数最高的那个。

from llama_index.core.node_parser import SentenceSplitter

# 将文档切分为 Node 以模拟重复召回场景
node_parser = SentenceSplitter(chunk_size=100, chunk_overlap=20)
nodes = node_parser.get_nodes_from_documents(documents)
split_index = VectorStoreIndex(nodes)

# 不加去重的检索（研发部过滤）
retriever_raw = split_index.as_retriever(similarity_top_k=5, filters=filters)
results_raw = retriever_raw.retrieve("加班餐费报销需要什么材料？")

print("【去重前】")
for i, node in enumerate(results_raw):
    source_id = node.metadata.get("source_doc_id", "N/A")[:20]
    print(f"[{i+1}] source:{source_id} | score:{node.score:.4f}")

# 手动实现 Document 级去重
def deduplicate_by_source(nodes):
    seen = set()
    result = []
    for n in nodes:
        source_id = n.metadata.get("source_doc_id", n.node_id)
        doc_id = source_id.split(":")[0]  # 提取 document_id 前缀
        if doc_id not in seen:
            seen.add(doc_id)
            result.append(n)
    return result

dedup_results = deduplicate_by_source(results_raw)
print("\n【去重后】")
for i, node in enumerate(dedup_results):
    source_id = node.metadata.get("source_doc_id", "N/A")[:20]
    print(f"[{i+1}] source:{source_id} | score:{node.score:.4f}")

运行结果：

【去重前】
[1] source:88a3f... | score:0.8521
[2] source:88a3f... | score:0.7634   # 同一文档的另一个 Node
[3] source:bb72c... | score:0.7012

【去重后】
[1] source:88a3f... | score:0.8521   # 只保留最高分
[2] source:bb72c... | score:0.7012

---

测试验证：5 个跨部门测试问题

#	问题	期望部门	过滤前命中率	过滤后命中率
1	加班餐费报销需要什么材料？	研发部	✅ 正确	✅ 正确
2	商务宴请人均上限是多少？	市场部	❌ 返回行政部	✅ 正确
3	年度培训费可以报销哪些项目？	财务部	✅ 正确	✅ 正确
4	办公用品超标需要谁审批？	行政部	❌ 返回研发部	✅ 正确
5	项目上线期间餐费怎么算？	研发部	❌ 返回财务部	✅ 正确

---

可能遇到的坑及解决方法

1. MetadataFilters 的 key 名称必须与文档元数据完全匹配（区分大小写）：文档中写入的是 department，过滤时写 Department 会静默不匹配，返回空结果而不报错。建议用常量定义元数据 key，避免手写字符串。

2. score 阈值在不同 Embedding 模型下差异大：OpenAI text-embedding-3-small 的相似度普遍偏高（0.65~0.95），本地 BGE 模型偏低（0.45~0.85），阈值需根据模型重新校准。建议抽取 100 个问题做抽样统计后确定阈值。

3. 多条件过滤的性能影响：当 MetadataFilters 包含 3 个以上 AND 条件且候选集很小时，向量搜索退化为全量标量扫描，延迟可能翻倍。建议将过滤条件控制在 2 个以内，复杂逻辑移到索引外做。

---

完整代码清单

完整代码请参考：src/chapter07_retriever_demo.py

---

4. 项目总结

三种检索策略对比

维度	默认检索（无过滤）	元数据过滤检索（Pre-filtering）	检索后过滤（Post-filtering）
准确率	低，容易跨部门误召回	高，只返回目标部门数据	中，top_k 可能全被过滤掉
性能	最快，纯向量搜索	较快，混合查询有开销	快，但可能空返回后需二次搜索
实现复杂度	极低，一行代码	低，加 MetadataFilters	中，需自行实现过滤+补召逻辑
适用场景	无权限要求的知识库（如开源 FAQ）	部门级隔离、权限控制明确的场景	过滤条件动态变化、多租户
安全性	无保障，可能泄露跨部门数据	好，检索层面就隔离了	差，顶层数据先拿到再过滤

适用场景

• 部门级企业知识库：不同部门只能检索自己的制度、规范、FAQ
• 多租户 SaaS 问答系统：每个租户的数据完全隔离
• 权限分级的知识库：普通员工、主管、高管的可见文档不同
• 多语言内容隔离：按语言标签过滤，中文查询不看英文文档
• 不适用场景：需要跨部门综合回答（如总经理问"全公司平均报销额度"），这时过滤反而是障碍；过滤条件过于复杂（如多字段 OR + 日期范围），应交给重排层而非检索层处理。

注意事项

• MetadataFilters key 命名规范：统一使用小写 snake_case，如 department、publish_date、security_level，在项目中用常量文件统一管理 key 名
• 过滤条件数量对性能的影响：超过 3 个 AND 条件且候选集不足时，建议先放宽过滤再靠 score 阈值和去重做后处理
• score 阈值需要按业务场景调优：无标准值，建议用 golden dataset 在 0.5~0.9 之间二分搜索最优值

常见踩坑经验

1. 过滤条件写错导致返回空结果不报错：MetadataFilter(key="dept", ...) 而元数据字段是 department——系统不会报错，只是匹配不到任何文档。建议在所有检索调用后加 assert len(results) > 0 或日志记录空召回。
2. score 阈值设太高导致专业术语问题无法召回：法务文档中"不可抗力"的 Embedding 相似度普遍不超过 0.72，阈值设为 0.75 会导致大量合法术语查询空返回。阈值调优数据驱动，不能靠直觉。
3. 文档级去重导致漏掉有用的不同 Node：同一法律合同的前半段讲付款、后半段讲违约——两者语义完全不同却被去重了。去重粒度应精确到 Node 语义簇，粗暴按 source_doc_id 去重会丢失关键信息。

思考题

1. 如何实现一个支持时间衰减的 Retriever——越新的文档权重越高？ 请给出实现思路（提示：在检索后对 NodeWithScore 按发布时间加权，如 final_score = similarity_score * (1 + log(1 + days_since_publish))）。

2. 如果一个用户同时属于多个部门（如矩阵式组织的项目经理），检索时应该怎么设计过滤逻辑？ 是 OR 过滤（看多个部门）还是走权限标签系统？请对比两种方案的安全性和实现复杂度。

---

下一章预告：第 8 章将带你进入 QueryEngine 的世界——如何把检索到的零散 Node 编织成流畅、准确、带引用的答案。