第九章：核心能力详解 - Retrieval 与 RAG

如果说 Agent 解决的是“模型如何行动”，那么 Retrieval 与 RAG 解决的则是“模型如何基于外部知识可靠回答”。在企业场景中，大部分高价值应用都离不开 RAG，因为企业最重要的信息通常不在模型训练语料里，而在内部制度、合同、产品资料、项目文档、知识库、工单记录和业务数据库中。

本章将从 RAG 的原理、文档加载、切分、Embedding、向量索引、检索链构建、提示词设计、重排、多查询检索、评估与防幻觉等方面展开，并加入更偏工程实战的具体代码案例。

大模型再强，也无法天然满足企业问答的 4 个核心诉求：

• 实时性：模型并不知道公司昨天刚更新的制度；
• 私有性：模型训练时没有你的内部数据；
• 可追溯性：模型自然语言回答很难给出清晰出处；
• 可控性：没有外部证据时，模型仍可能自信地幻觉。

RAG 的核心思想就是：先检索，再生成。

很多人把 RAG 理解为：

• 文档切一下；
• 存进向量库；
• top-k 检索；
• 把结果塞进 prompt。

这只是最初级版本。一个可靠的 RAG 系统往往至少包含：

• 文档接入；
• 清洗标准化；
• 分块；
• Embedding；
• 向量索引；
• 查询改写；
• 检索；
• 重排；
• 生成；
• 引用；
• 评估与监控。

原始文档 -> 清洗 -> 切分 -> Embedding -> 写入向量库
用户问题 -> 查询理解 -> 检索 -> 重排 -> 生成 -> 返回引用

这个流程中的任何一步变差，都会拖累最终效果。

RAG 的数据源并不限于 PDF：

• Word / PDF / Markdown / HTML；
• FAQ 网页；
• DokuWiki / Confluence / 飞书文档；
• 工单系统和 CRM 备注；
• 数据库导出；
• API 响应；
• 邮件归档；
• 表格和报表。

在 LangChain 生态中，这一步通常由 Document Loaders 完成。

from langchain_community.document_loaders import DirectoryLoader, TextLoader
 
loader = DirectoryLoader(
    "./kb",
    glob="**/*.md",
    loader_cls=TextLoader,
    show_progress=True,
)
documents = loader.load()
print(len(documents))
print(documents[0].page_content[:200])
print(documents[0].metadata)

很多低质量 RAG 的问题根源并不在模型，而在脏数据：

• 页眉页脚重复；
• OCR 错乱；
• 目录和正文混在一起；
• 表格列断裂；
• 版本冲突；
• 失效文件未下架；
• 一个文档出现多个互相矛盾副本。

因此，在生产实践中，你往往要先做一轮文档规范化，例如：

• 去重；
• 补充来源与日期 metadata；
• 标记版本号；
• 删除“草稿”“废弃”“测试文档”；
• 将图片 OCR 成可检索文本。

整篇文档直接入库的问题包括：

• 语义粒度太粗；
• 容易命中大量无关上下文；
• 不利于引用定位；
• 生成阶段 token 成本太高。

因此，RAG 的一个核心工作就是切 chunk。

from langchain_text_splitters import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=80,
    separators=["
 
", "
", "。", "，", " ", ""]
)
chunks = splitter.split_documents(documents)
print(len(chunks))
print(chunks[0].page_content)

通常没有一组万能参数，必须结合文档类型调优：

• FAQ 适合较小 chunk；
• 制度文档往往需要按标题层级切分；
• 合同类文档适合按条款切；
• 技术文档可以适度保留更大上下文。

经验上可以先从：

• `chunk_size=400~800`
• `chunk_overlap=50~120`

开始试验。

例如：

• 制度文档按“章节 → 条款”切；
• FAQ 按“问题 + 答案”切；
• API 文档按“接口 + 参数 + 示例”切；
• 合同按“条款编号”切。

因为这些切分方式更接近用户真实提问的语义单位。

Embedding 模型会把文本映射到向量空间，让“语义相近”的文本靠得更近。这样即使用户问的是“差旅报销需要哪些材料”，文档里写的是“出差费用报销应提交票据和审批单”，系统仍有机会通过语义相似度命中。

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
 
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

写入向量库时，应尽量保留 metadata，例如：

• `source`：来源文件；
• `doc_type`：制度、FAQ、合同、公告；
• `department`：人事、财务、行政；
• `version`：版本号；
• `effective_date`：生效日期。

因为很多时候精准检索不仅靠语义相似度，还要靠 metadata 过滤。

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
results = retriever.invoke("差旅报销需要提交什么材料？")
for doc in results:
    print(doc.metadata)
    print(doc.page_content[:200])
    print("-" * 30)

retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 4,
        "filter": {"department": "finance"}
    }
)

这类过滤非常适合：

• 财务类问题只检索财务制度；
• 某个项目只检索该项目资料；
• 只检索最新版本文档。

除了使用高阶 helper，LangChain 也非常适合用 LCEL（LangChain Expression Language）显式搭建 RAG 链。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
 
prompt = ChatPromptTemplate.from_template("""
你是企业知识助手。请仅依据下面提供的资料回答问题。
如果资料不足，请明确回答“未找到足够依据”。
 
问题：{question}
 
资料：
{context}
""")
 
def format_docs(docs):
    return "
 
".join(doc.page_content for doc in docs)
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | model
    | StrOutputParser()
)
 
answer = rag_chain.invoke("差旅报销需要提交什么材料？")
print(answer)

这个例子非常适合教学，因为它完整展示了：

• 输入问题；
• 检索文档；
• 格式化上下文；
• 拼入 Prompt；
• 调用模型；
• 输出最终答案。

如果你希望使用更高层的封装，也可以使用官方常见的 retrieval helper：

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是企业知识助手。仅依据提供资料回答；若资料不足，请明确说明。"),
    ("human", "问题：{input}
 
资料：{context}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
doc_chain = create_stuff_documents_chain(llm, prompt)
retrieval_chain = create_retrieval_chain(retriever, doc_chain)
 
result = retrieval_chain.invoke({"input": "报销审批一般需要多久？"})
print(result)

基础 top-k 检索只是开始。很多实际项目还会加：

• 查询改写；
• 多查询检索；
• 重排模型；
• 父子块检索；
• 混合检索（语义 + 关键词）。

query_variants = [
    "差旅报销需要提交什么材料？",
    "出差费用报销要附哪些单据？",
    "报销差旅费时需要哪些凭证？"
]
 
all_docs = []
seen = set()
for query in query_variants:
    for doc in retriever.invoke(query):
        key = (doc.metadata.get("source"), doc.page_content[:100])
        if key not in seen:
            seen.add(key)
            all_docs.append(doc)

多查询的价值在于：

• 用户提问不一定使用文档原词；
• 一个问题可能包含多个角度；
• 某些术语在不同部门有不同叫法。

向量检索擅长“召回”，但不一定擅长“最终排序”。在要求较高的系统里，往往会：

• 先用向量库召回 10~20 个候选；
• 再用重排模型对候选做精排；
• 只把前 3~5 个高质量 chunk 交给生成模型。

一个可靠的 RAG 提示词至少应包含：

• 仅依据资料回答；
• 若资料不足，明确承认不知道；
• 对流程类问题尽量步骤化输出；
• 尽可能附带引用来源；
• 不要把未命中的常识混成企业事实。

prompt = ChatPromptTemplate.from_template("""
你是公司内部知识库助手。
请严格依据提供的资料回答问题，不得虚构制度、日期或流程。
如果资料不足，请回答“未找到足够依据”，并指出还需要什么信息。
 
请按以下格式回答：
1. 直接答案
2. 依据说明
3. 引用来源
 
问题：{question}
 
资料：
{context}
""")

RAG 不会自动消灭幻觉。RAG 仍可能出错，原因包括：

• 检索没命中真正相关文档；
• chunk 中混入无关内容；
• 文档本身过时或冲突；
• 生成提示词没有强约束；
• 模型把自身常识和检索内容混在一起。

常见防护手段包括：

• 输出引用片段；
• 显式要求资料不足时拒答；
• 对关键场景加入答案验证步骤；
• 对文档引入版本号与生效日期。

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
 
loader = DirectoryLoader("./kb", glob="**/*.md", loader_cls=TextLoader)
docs = loader.load()
 
splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=100)
chunks = splitter.split_documents(docs)
 
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(model="text-embedding-3-large"),
    persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
 
prompt = ChatPromptTemplate.from_template("""
你是企业知识助手，仅依据资料回答。
若资料不足，请直接回答“未找到足够依据”。
 
问题：{question}
 
资料：
{context}
""")
 
def format_docs(docs):
    return "
 
".join(
        f"[来源: {doc.metadata.get('source', 'unknown')}]
{doc.page_content}"
        for doc in docs
    )
 
chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)
 
print(chain.invoke("试用期员工可以申请差旅报销吗？"))

一个可上线的 RAG 系统，不能只靠“感觉还行”。至少要评估：

• 命中率：相关资料是否被检出；
• 精准率：检出的资料噪声是否过大；
• 回答正确率；
• 引用准确率；
• 平均延迟；
• 单次成本；
• 失败类型分布。

建议你建立一个小规模评测集，例如 50~200 个真实问题，记录：

• 标准答案；
• 标准引用；
• 所属文档；
• 难度等级；
• 是否多跳推理。

• RAG 的核心是“先检索，再生成”；
• 构建高质量 RAG，重点不只是向量库，而是整条知识链路；
• 文档清洗、切分、Embedding、metadata、检索和提示词缺一不可；
• LCEL 是理解 RAG 链路最清晰的方式之一；
• 企业级 RAG 必须重视引用、版本控制、防幻觉和评估。

1. 使用一批公司制度文档，设计一个从清洗、切分到索引的 RAG 数据处理流程。

2. 比较两种切分策略：固定长度切分与按标题层级切分，分析它们对检索质量的影响。

3. 用 LCEL 手写一个最小 RAG 链，并在答案中输出引用来源。

4. 设计一个查询改写模块，为同一个问题生成 3 个语义等价查询并合并检索结果。

5. 为你的业务场景建立一个 20 条问题的 RAG 评测集。

• LangChain 官方文档：Retrieval