AI Agent 开发学习与面试指南：03-生产级 RAG 与知识工程

1. RAG 的目标与边界

RAG（Retrieval-Augmented Generation）在生成前检索外部证据，把知识更新与模型参数解耦。它适合：

• 企业内部文档问答；
• 频繁更新的政策、产品和操作手册；
• 需要返回引用和原文依据的任务；
• 数据不适合或无法进入模型训练的场景。

RAG 不适合单独解决：

• 需要实时事务状态但没有实时数据工具；
• 原始数据错误或互相矛盾；
• 任务需要执行动作而不是查知识；
• 模型缺少某种稳定行为能力；
• 严格计算、精确规则或权限判断。

🔥 P0 高频必会：RAG 和 Fine-tuning 的区别？
RAG 在推理时提供外部知识，更新快、可引用；微调改变模型参数中的行为或能力，更新成本更高，也不能保证事实实时。企业知识问答通常先 RAG，格式/风格/特定行为长期不稳定时再考虑微调。

2. 一条完整 RAG 链路

数据源
  -> 解析与清洗
  -> 文档结构恢复
  -> 分块与元数据
  -> Embedding / 稀疏索引
  -> 召回
  -> 融合与去重
  -> Rerank
  -> 上下文组装
  -> 生成与引用
  -> 评测与反馈

任何一环都可能导致最终答案错误。面试时不要只谈向量数据库，要能沿链路定位失败。

3. 数据解析：垃圾进，垃圾出

PDF 不是天然的段落文本。常见问题包括：

• 双栏排版顺序错乱；
• 页眉页脚重复；
• 扫描件没有文本层；
• 表格被拆散；
• 标题层级丢失；
• 代码、公式和脚注位置错误；
• 同一文档多版本重复入库。

推荐为每个解析单元保留：

from pydantic import BaseModel


class ParsedBlock(BaseModel):
    document_id: str
    document_version: str
    block_id: str
    block_type: str       # paragraph/table/code/title
    text: str
    page: int | None
    heading_path: list[str]
    source_uri: str
    updated_at: str
    access_tags: list[str]

source_uri、页码和标题路径直接影响引用体验；access_tags 用于检索前权限过滤；document_version 用于增量更新和回滚。

增量索引

不要每次全量重建。计算内容哈希：

import hashlib


def content_hash(text: str) -> str:
    normalized = " ".join(text.split())
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

只有哈希变化的块需要重新 Embedding。删除的文档要同步删除索引，否则模型会继续检索到过期内容。

4. Chunking：分块不是固定字符切割

Chunk 太小：语义不完整、答案需要跨块拼接。
Chunk 太大：召回不精准、上下文浪费、Rerank 成本高。

常见策略：

一个结构感知的简单实现：

from dataclasses import dataclass


@dataclass
class Chunk:
    chunk_id: str
    text: str
    heading_path: tuple[str, ...]
    source_uri: str


def chunk_sections(sections, max_chars: int = 1200) -> list[Chunk]:
    chunks = []
    for section in sections:
        paragraphs = section.paragraphs
        buffer = []
        length = 0

        for paragraph in paragraphs:
            if buffer and length + len(paragraph) > max_chars:
                index = len(chunks)
                chunks.append(Chunk(
                    chunk_id=f"{section.id}:{index}",
                    text="\n".join(buffer),
                    heading_path=tuple(section.heading_path),
                    source_uri=section.source_uri,
                ))
                buffer = []
                length = 0

            buffer.append(paragraph)
            length += len(paragraph)

        if buffer:
            index = len(chunks)
            chunks.append(Chunk(
                chunk_id=f"{section.id}:{index}",
                text="\n".join(buffer),
                heading_path=tuple(section.heading_path),
                source_uri=section.source_uri,
            ))
    return chunks

这个例子按字符近似，生产中更适合按模型 Tokenizer 计数，并避免把表格、代码和标题切开。

🔥 P0 高频必会：Chunk 大小如何选择？
不能只给固定数字。先看文档结构与问题粒度，建立标注查询集，比较不同 Chunk/overlap 对 Recall@k、答案质量、延迟和成本的影响。通常先结构化分块，再用实验确定范围。

5. Embedding 与相似度

Embedding 把文本映射到向量空间，使语义相近的文本距离更近。常见相似度包括余弦相似度、点积和欧氏距离。必须保证索引和查询使用兼容模型及相同归一化方式。

工程选择要看：

• 中文和多语言效果；
• 查询与文档是否使用不同指令前缀；
• 最大输入长度；
• 维度、存储成本和检索延迟；
• 是否支持领域文本；
• 模型升级后是否需要重建索引。

不要仅凭公开榜单选 Embedding。用自己的查询—相关文档标注集测试。

6. Dense、Sparse 与 Hybrid Search

Dense 检索擅长语义、同义改写；BM25 等 Sparse 检索擅长产品编号、专有名词、精确关键词。混合检索同时利用两者。

Pinecone 官方文档也指出：语义检索可能漏掉精确词，词法检索可能漏掉同义表达；Hybrid Search 用于互补。参见 Hybrid search。

RRF 融合

Reciprocal Rank Fusion 不要求不同检索器分数同尺度：

RRF(d) = Σ 1 / (k + rank_i(d))

实现：

from collections import defaultdict


def reciprocal_rank_fusion(
    ranked_lists: list[list[str]],
    k: int = 60,
) -> list[tuple[str, float]]:
    scores = defaultdict(float)
    for ranked in ranked_lists:
        for rank, document_id in enumerate(ranked, start=1):
            scores[document_id] += 1.0 / (k + rank)

    return sorted(scores.items(), key=lambda item: item[1], reverse=True)

RRF 简洁稳健，但最优参数仍应通过标注集选择。也可以对 Dense/Sparse 分数归一化后加权，但要警惕两种分数分布不一致。

🔥 P0 高频必会：为什么向量检索后还需要 BM25？
向量检索可能错过精确标识符、错误码、人名和新词；BM25 对关键词匹配更稳定。Hybrid 用语义与词法信号互补，之后可用 Reranker 统一排序。

7. Query 改写与路由

用户问题不一定适合直接检索：

• “它什么时候到期？”需要结合会话中的对象；
• “比较 A 和 B”需要拆成多个子查询；
• “如何重置 XX-1042”含精确错误码，应提高 Sparse 权重；
• 闲聊不应访问知识库；
• 实时订单状态应路由到业务工具，不应查文档。

推荐把改写输出结构化：

from typing import Literal
from pydantic import BaseModel


class RetrievalPlan(BaseModel):
    route: Literal["knowledge", "transaction_api", "no_retrieval"]
    queries: list[str]
    keyword_terms: list[str]
    filters: dict[str, str]

改写本身也可能出错，因此 Trace 中要同时保存原始问题和改写查询，评测时分别衡量。

8. Metadata Filter：先过滤权限，再做相似度

企业知识库的第一原则是：用户无权查看的文档不能进入候选集，更不能先检索后“希望模型别说”。

过滤维度常见：

• tenant / organization；
• department；
• classification；
• document status；
• version；
• language；
• updated_at；
• product / region。

权限过滤必须由可信身份和服务端规则生成，不能直接采用模型输出的 department="finance"。

9. Rerank：两阶段检索

第一阶段召回较多候选，追求不漏；第二阶段用 Cross-Encoder 或 Reranker 对“查询—文档”对逐一评分，追求精排质量。

Hybrid Top 50 -> 去重 -> Rerank -> Top 6 -> 上下文组装

Rerank 会增加延迟和成本，因此要控制候选数、批处理并缓存。Pinecone 将 Rerank 描述为提升两阶段 RAG 质量的直接方法，参见 Rerank results。

🔥 P0 高频必会：召回和重排的目标有什么不同？
召回阶段优先提高 Recall，允许带回一些噪声；重排阶段在候选中提高精度，把真正相关证据放到前面。只把 Top 3 向量结果直接给模型，常会因为召回遗漏而无解。

10. 上下文组装与引用

上下文不应只是字符串拼接。每段证据要带稳定编号：

[S1]
source: employee_handbook.pdf
page: 12
section: 请假制度 > 年假
text: ...

[S2]
source: hr_policy_2026.md
section: 特殊休假
text: ...

要求答案中的事实使用 [S1] 形式引用。生成后验证：

1. 引用 ID 是否存在；
2. 引用文本是否真的支持相应句子；
3. 如果无支持证据，是否正确拒答；
4. 是否引用了用户无权访问的来源。

引用存在不代表引用正确。需要 Citation Correctness 评测或人工抽检。

11. RAG 评测：分层测，不要只看最终回答

11.1 检索指标

• Recall@k：相关文档是否出现在前 k 个结果中；
• Precision@k：前 k 个结果中有多少相关；
• MRR：第一个相关结果的排名倒数；
• NDCG@k：考虑多级相关度与排名位置。

例如 100 个查询中有 82 个在 Top 5 找到至少一个相关 Chunk，则 Hit/Recall 类成功率是 82%。需要先定义“相关”标注口径。

11.2 生成指标

• 答案是否正确、完整；
• 是否忠实于证据；
• 引用是否支持对应事实；
• 无答案问题能否拒答；
• 是否泄露无权限信息。

11.3 端到端指标

• 用户任务完成率；
• 首次解决率；
• 人工转接率；
• P50/P95 延迟；
• 单请求成本。

🔥 P0 高频必会：最终答案错了，如何定位是检索还是生成问题？
先检查标注相关文档是否进入候选集；没进入是解析/分块/查询/召回问题。进入但排名太低是融合/Rerank 问题。正确证据已进入上下文而回答仍错，才主要是生成、指令或上下文冲突问题。

12. 常见失败诊断表

13. 一个可测试的 RAG 接口

from dataclasses import dataclass
from typing import Protocol


@dataclass
class Evidence:
    chunk_id: str
    text: str
    source_uri: str
    score: float


class Retriever(Protocol):
    async def search(self, query: str, filters: dict, top_k: int) -> list[Evidence]: ...


class Reranker(Protocol):
    async def rerank(
        self, query: str, candidates: list[Evidence], top_n: int
    ) -> list[Evidence]: ...


async def retrieve_context(
    query: str,
    filters: dict,
    retriever: Retriever,
    reranker: Reranker,
) -> list[Evidence]:
    candidates = await retriever.search(query, filters=filters, top_k=40)
    deduped = {item.chunk_id: item for item in candidates}
    ranked = await reranker.rerank(query, list(deduped.values()), top_n=6)
    return ranked

使用协议接口后，可以在单元测试中分别替换 Retriever 和 Reranker，精确定位组件问题。

14. 本章练习

练习 A：构建标注集

选择 20–50 篇文档，编写 60 个问题：

• 30 个普通问答；
• 10 个精确关键词/编号问题；
• 10 个跨段落问题；
• 5 个无答案问题；
• 5 个权限隔离问题。

为每个问题标注相关文档/Chunk 和标准答案。

练习 B：比较四条检索链

比较：Dense、BM25、Hybrid、Hybrid+Rerank。输出 Recall@5、MRR、P95 延迟和成本。

练习 C：失败分类器

将错误分为解析、Chunk、查询改写、召回、Rerank、生成、引用、权限八类。每次失败必须落到一类，并保存 Trace。

15. 面试高频问答

🔥 P0：RAG 的完整链路是什么？

数据解析清洗、结构恢复、分块与元数据、索引、查询理解、召回、融合去重、Rerank、上下文组装、生成引用、分层评测与反馈闭环。回答时应指出每层失败如何观测。

🔥 P0：如何解决 RAG 幻觉？

先提高证据质量；要求引用和无证据拒答；限制回答只能使用证据；对引用做支持性校验；把高风险事实交给确定性工具；建立无答案与冲突证据测试集。无法保证绝对消除。

🔥 P0：如何选择向量数据库？

从数据规模、延迟、过滤能力、混合检索、索引更新、一致性、多租户、备份恢复、部署运维和成本评估。不要只比较 ANN 算法名；先用真实查询集测试。

⭐ P1：什么时候用 GraphRAG？

当问题依赖实体关系、多跳路径和全局结构时可能有价值，例如组织关系、供应链、科研知识。它增加抽取、图更新和评测成本；普通 FAQ 或局部文档问答通常先用高质量 Hybrid RAG。

⭐ P1：Embedding 模型升级要注意什么？

新旧向量空间通常不兼容，需要双写或重建索引；做离线 A/B；记录模型版本；规划无停机迁移；验证维度、归一化、语言和领域效果；更新完成后再切流量。

16. 本章完成标准

• 能从数据源到最终引用画出完整 RAG 链；
• 能实现并解释 Hybrid + RRF + Rerank；
• 有带相关文档标注的查询集；
• 能分别报告检索、生成和端到端指标；
• 能从 Trace 判断错误发生在哪一层；
• 能设计权限过滤、增量索引和版本迁移。