RAG 从原理到落地：让答疑机器人读懂公司制度文件

上一篇用纯提示词搭了一个答疑机器人，它口才不错，但有个致命短板：公司内部的制度、岗位职责、工具规范，模型训练时根本没见过，问起来它只能靠猜。这一篇解决这个问题，思路叫 RAG（Retrieval Augmented Generation，检索增强生成）。

全文基于 LlamaIndex + 阿里云百炼（DashScope）实现，从原理讲到可复用的代码模块，最后处理 RAG 最容易翻车的场景：多轮对话。

一、为什么是 RAG：开卷考试的逻辑

先想清楚问题出在哪。大模型的知识来自训练数据，截止在某个时间点，而且全是公开语料。你们公司的《考勤管理制度》不在里面，直接问"年假怎么算"，模型要么编一个，要么给一堆通用废话。

RAG 的解法可以用开卷考试来理解：闭卷靠记忆，记不住就丢分；开卷靠翻资料，找到相关段落，加上自己的理解作答。大模型也一样——回答前先把相关知识检索出来塞进上下文窗口，让它"看着资料答题"。

这就是上下文工程（Context Engineering）的核心：给模型的上下文窗口填充恰到好处的信息。信息太少，模型"不知道"；信息太多或跑题，性能下降、成本上涨。RAG 是这件事里最成熟的一套方法论，整个流程分两个阶段：

阶段	干什么	类比
建立索引（离线，做一次）	把知识库文档加工成可检索的形式	考前给参考资料划重点、贴标签
检索生成（在线，每次提问）	找到相关段落，交给大模型作答	考场上翻到对应页，照着理解答题

建立索引的四个步骤

1. 文档解析：把 PDF、Word、Markdown 等格式的知识库文档加载成纯文本。格式五花八门，这一步是所有后续处理的地基。
2. 文本分段：把长文档切成一个个小段落（chunk）。没人答题时翻整本书，检索也是一样——按段落找，比按整本找又快又准。
3. 文本向量化：用 embedding 模型把每个段落变成一个高维向量。两段文字语义越接近，向量越相似，"意思相近"这件事就变成了可计算的数学问题。
4. 存储索引：把向量和原文段的对应关系存进向量数据库。不然每次提问都重新解析、向量化一遍，响应速度没法看。

检索生成的两个阶段

• 检索（Retrieval）：用户提问后，同样用 embedding 模型把问题变成向量，去向量数据库里找语义最相近的几个段落。这是 RAG 最重要的环节——考场上翻错了资料，后面答得再流畅也是错的。
• 生成（Generation）：把检索到的段落和用户问题一起填进提示词模板，交给大模型生成回答。典型模板长这样：请根据以下信息回答用户的问题：{召回文本段}。用户的问题是：{question}。注意这个阶段用的是模型的总结能力，不是它自己的知识——资料给对，它照着说就行。

小结：RAG = 离线建索引 + 在线检索生成。检索质量决定上限，生成只是照着资料作答。后面所有优化手段（重排、句子窗口检索等）都是在折腾这两个环节。

二、最小可用的 RAG：三十行跑通全流程

原理清楚了，上手写。依赖先装好：

# 终端执行
pip install llama-index llama-index-embeddings-dashscope llama-index-llms-openai-like
pip install llama-index-readers-file PyMuPDFReader docx2txt

跑代码前还有一个坑要提前填：LlamaIndex 处理文本时默认用 NLTK 做句子切分，首次使用会自动联网下载分词语料，离线或网络受限的环境会卡死或报 LookupError。最省事的做法是设一个环境变量禁用它，LlamaIndex 会退回用正则表达式切分，效果对中文场景够用。

# ch02/Test02_rag.py
from llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.readers.file import PyMuPDFReader
from llama_index.llms.openai_like import OpenAILike

import os
import logging
logging.basicConfig(level=logging.ERROR)

# 禁用 NLTK，避免首次运行时联网下载分词语料失败
os.environ["LLAMA_INDEX_DISABLE_NLTK"] = "1"

if __name__ == '__main__':
    # ---- 步骤1：加载文档 ----
    # SimpleDirectoryReader 遍历目录下所有文件，按扩展名自动选择解析器
    # 这里单独指定 .pdf 用 PyMuPDF 解析，比默认的 pypdf 更稳
    print("正在解析文件...")
    documents = SimpleDirectoryReader('docs',
                                      file_extractor={".pdf": PyMuPDFReader()}).load_data()

    # ---- 步骤2：构建向量索引 ----
    # from_documents 内部完成：切分文本块 -> 逐块向量化 -> 建立检索索引
    print("正在创建索引...")
    index = VectorStoreIndex.from_documents(documents,
                                            embed_model=DashScopeEmbedding(
                                                model_name='text-embedding-v4',
                                                embed_batch_size=10  # v3 以上版本批量必须 <= 10
                                            ))

    # ---- 步骤3：创建查询引擎 ----
    # as_query_engine 封装了"检索 + 生成"：先按问题向量检索文本块，
    # 再把文本块作为上下文连同问题发给 LLM 生成回答
    print("正在创建提问引擎...")
    query_engine = index.as_query_engine(
        streaming=True,
        llm = OpenAILike(
            model="qwen-plus",
            api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
            api_key=os.environ['DASHSCOPE_API_KEY'],
            is_chat_model=True
        ),
    )

    # ---- 步骤4：执行查询并流式打印 ----
    question = '我们公司内容管理工程师进行项目管理应该用什么工具'
    print("正在生成回复...")
    streaming_response = query_engine.query(question)

    print("回答是：")
    streaming_response.print_response_stream()

几处关键选择说一下为什么：

• Embedding 模型用 text-embedding-v4：百炼的向量模型，把文本映射成高维向量做语义检索。注意 embed_batch_size=10 这个参数——v3 以上的版本单批最多 10 条，不设的话默认批量会直接报错，这是官方文档里不太起眼但一跑就撞上的坑。
• LLM 用 OpenAILike 而不是专门的 DashScope 封装：百炼提供了兼容 OpenAI API 的端点，任何兼容 OpenAI 格式的服务都能这样接。以后换模型供应商，改一个 api_base 和 model 就行，代码不动。
• streaming=True：流式输出，回答逐字蹦出来。RAG 检索本身很快，慢的是生成，流式让用户第一秒就看到反馈，体验差很多。

提问"我们公司内容管理工程师进行项目管理应该用什么工具"，机器人能从制度文档里翻出答案——这些内容模型训练时从没见过，全靠检索喂给它的。

三、索引持久化：别每次提问都重建一遍

上面的代码每次运行都要重新解析、重新向量化，文档一多要等好几分钟。而知识库文档并不会每分钟都变——建立索引本来就该是离线做一次的事。LlamaIndex 的解法是把索引存成本地文件，下次直接加载：

# ch02/Test03_vec_db.py
from llama_index.core.indices.base import BaseIndex
from llama_index.embeddings.dashscope import DashScopeEmbedding
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.readers.file import PyMuPDFReader
from llama_index.core import StorageContext, load_index_from_storage

import os
import logging
logging.basicConfig(level=logging.ERROR)

os.environ["LLAMA_INDEX_DISABLE_NLTK"] = "1"

# 1. 解析文档并建立索引
def parse_docs(dir = 'docs'):
    print("正在解析文件...")
    documents = SimpleDirectoryReader(dir,
                                      file_extractor={".pdf": PyMuPDFReader()}).load_data()
    print("正在创建索引...")
    index = VectorStoreIndex.from_documents(documents,
                                            embed_model=DashScopeEmbedding(
                                                model_name='text-embedding-v4',
                                                embed_batch_size=10
                                            ))
    return index

# 2. 索引持久化到本地目录
def store_index(index : BaseIndex , dir = 'knowledge_base/test'):
    index.storage_context.persist(dir)
    print(f"索引文件保存到了{dir}")

# 3. 从本地目录加载索引
def load_index(dir = 'knowledge_base/test'):
    storage_context = StorageContext.from_defaults(persist_dir=dir)
    index = load_index_from_storage(storage_context,
                                    embed_model=DashScopeEmbedding(
                                        model_name='text-embedding-v4',
                                        embed_batch_size=10
                                    ))
    print(f"成功从{dir}路径加载索引")
    return index

三步走的逻辑很直白：

• parse_docs 建索引，和上一节完全一样；
• store_index 调用 storage_context.persist()，把切分后的文本块、向量、索引结构整体写到本地目录；
• load_index 是反操作，用 StorageContext.from_defaults(persist_dir=...) 指定目录，load_index_from_storage 还原出索引对象。

有一个容易忽略的细节：加载时必须传入和建索引时相同的 embedding 模型。因为检索阶段要把用户问题向量化，再和库里存的向量算相似度——两次用了不同的 embedding 模型，向量根本不在同一个语义空间里，相似度算出来就是垃圾。这也是为什么 load_index 里还要再写一遍 DashScopeEmbedding(...)。

使用时两个入口各干各的：

# ch02/Test03_vec_db.py（续）
if __name__ == '__main__':
    # 首次运行：建索引并保存（跑一次就够了）
    # index = parse_docs()
    # store_index(index)

    # 之后每次运行：直接加载，秒级启动
    index = load_index('knowledge_base/test')

把首次运行的注释解开跑一遍，之后注释回去——索引已经躺在本地了，启动从几分钟降到几秒。文档更新时再重跑一次 parse_docs + store_index 覆盖旧的即可。

四、封装成可复用的 RAG 模块

三个脚本各自能跑之后，把公共逻辑抽成一个模块，后续迭代（换模型、调提示词、加评测）都基于它进行，不用每次复制粘贴：

# util/rag.py
from llama_index.core import (SimpleDirectoryReader, VectorStoreIndex,
                              StorageContext, load_index_from_storage)
from llama_index.embeddings.dashscope import DashScopeEmbedding
from llama_index.llms.openai_like import OpenAILike
import os
import logging
logging.basicConfig(level=logging.ERROR)

# DashScope 兼容 OpenAI 协议的端点，固定不变
DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"

def _embed_model():
    # 统一收口 embedding 模型配置，建索引和加载索引必须用同一个
    return DashScopeEmbedding(
        model_name='text-embedding-v4',
        embed_batch_size=10
    )

def indexing(document_path="docs", persist_path="knowledge_base/test"):
    """建立索引并持久化。文档更新后调用。"""
    documents = SimpleDirectoryReader(document_path).load_data()
    index = VectorStoreIndex.from_documents(documents, embed_model=_embed_model())
    index.storage_context.persist(persist_path)

def load_index(persist_path="knowledge_base/test"):
    """加载已持久化的索引。日常启动走这条路径。"""
    storage_context = StorageContext.from_defaults(persist_dir=persist_path)
    return load_index_from_storage(storage_context, embed_model=_embed_model())

def create_query_engine(index):
    """基于索引创建查询引擎，封装检索+生成全流程。"""
    return index.as_query_engine(
        streaming=True,
        llm=OpenAILike(
            model="qwen-plus",
            api_base=DASHSCOPE_BASE_URL,
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            is_chat_model=True
        ))

def ask(question, query_engine):
    """向答疑机器人提问，流式打印回答。"""
    streaming_response = query_engine.query(question)
    streaming_response.print_response_stream()

业务代码被压到几行：

# ch02/Test03_chatbot.py
import os
import sys
sys.path.append('util')
import rag

if __name__ == '__main__':
    # 索引已在之前步骤建好并持久化，这里直接加载
    # 若文档更新需要重建，先执行：rag.indexing()
    index = rag.load_index(persist_path='knowledge_base/test')
    query_engine = rag.create_query_engine(index)
    rag.ask('我们公司内容管理工程师进行项目管理应该用什么工具', query_engine)

结构上只有一点值得强调：_embed_model() 把 embedding 配置收口到一个函数里。第三节提过建索引和加载索引必须用同一个模型，收口之后这件事由代码结构保证，而不是靠开发者记得在两处写一样的配置。

这份 rag.py 依赖环境变量 DASHSCOPE_API_KEY。如果项目里没有现成的配置模块，可以补一个最小实现：

# util/tools.py（最小实现）
import os
from pathlib import Path
from dotenv import load_dotenv

def load_key():
    """从项目根目录的 .env 文件加载 API Key 等环境变量"""
    load_dotenv(Path(__file__).resolve().parent.parent / ".env")
    assert os.getenv("DASHSCOPE_API_KEY"), "请在 .env 中配置 DASHSCOPE_API_KEY"

.env 文件里写一行 DASHSCOPE_API_KEY=sk-xxxx，并把 .env 加进 .gitignore——Key 永远不进代码仓库。

五、RAG 的多轮对话难题：指代词让检索失效

单轮问答跑通了，接上多轮对话就暴露一个 RAG 特有的问题。

回忆检索的过程：系统拿用户问题去向量库里匹配。但如果问题依赖上一轮的上下文呢？看这个例子：

• 第一轮：用户问「张三的工位在哪里？」
• 第二轮：用户接着问「他的主管是谁？」

检索系统拿到的只有「他的主管是谁」这几个字。「他」是谁，检索系统完全不知道，embedding 一算，召回的段落和张三八竿子打不着，回答自然错。

那把完整对话历史拼进查询里一起检索行不行？也不行。历史一长，embedding 模型的注意力被稀释，检索效果反而下降——历史里一堆和当前问题无关的句子，把语义搅浑了。

业界的标准解法是问题改写（question rewrite）：先让大模型结合对话历史，把当前问题改写成一个自包含的独立问题，再用改写后的问题去检索。「他的主管是谁？」会被改写成「张三的主管是谁？」——指代补全了，检索就能命中。改写只花一次轻量的 LLM 调用，换来检索准确率，这笔账很划算。

LlamaIndex 内置了这套机制，对应 CondenseQuestionChatEngine：

# ch02/Test04_chat_history.py
from llama_index.core import PromptTemplate
from llama_index.core.llms import ChatMessage, MessageRole
from llama_index.core.chat_engine import CondenseQuestionChatEngine
from llama_index.llms.openai_like import OpenAILike

import os
import sys
sys.path.append('util')
import rag

# 1. 问题改写的提示词模板：把对话历史 + 后续问题，改写成独立问题
prompt_template = PromptTemplate(
    """
    给定一段对话历史（人类与助手之间）和人类的后续问题，
    请将该问题改写为一个独立的问题，包含对话中所有相关的上下文信息。

    <对话历史>
    {chat_history}

    <后续问题>
    {question}

    <改写后的独立问题>
"""
)

if __name__ == '__main__':
    # 2. 模拟一轮历史对话
    chat_history = [
        ChatMessage(role=MessageRole.USER, content="内容开发工程师有哪些细分类型？"),
        ChatMessage(role=MessageRole.ASSISTANT, content="内容开发工程师是综合性技术岗位。"),
    ]

    # 3. 复用前面封装好的 RAG 查询引擎
    rag_engine = rag.create_engine()

    # 4. 创建支持多轮对话的聊天引擎
    chat_engine = CondenseQuestionChatEngine.from_defaults(
        query_engine=rag_engine,
        condense_question_prompt=prompt_template,
        chat_history=chat_history,
        llm=OpenAILike(
            model="qwen-plus",
            api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
            api_key=os.environ['DASHSCOPE_API_KEY'],
            is_chat_model=True
        ),
        verbose=True  # 打印改写后的问题，调试时非常有用
    )

    # 5. 提问时故意不提"内容开发工程师"，只说"核心职责是什么？"
    streaming_response = chat_engine.stream_chat("核心职责是什么？")
    for token in streaming_response.response_gen:
        print(token, end="")

运行时开着 verbose=True，能清楚看到整个过程：模型先把「核心职责是什么？」改写成「内容开发工程师的核心职责是什么？」，改写后的问题才送去检索，召回的段落自然是对的。

这个引擎的工作流拆开看是两步循环：

1. 每轮用户输入进来，先由 LLM 结合 chat_history 做一次问题改写；
2. 改写后的问题交给普通 RAG 查询引擎（检索 + 生成），结果回给用户，同时这轮对话自动追加进 chat_history。

对话历史不用自己管，chat_engine.chat_history 随时可查。改写模板也可以按业务调——比如客服场景可以在模板里加一句"若问题已完整则原样返回"，避免不必要的改写引入偏差。

六、踩坑记录

现象	真相
text-embedding-v3 / v4 报批量超限错误	v3 以上版本单批最多 10 条，embed_batch_size=10 必须显式设置
启动时卡在下载 NLTK 语料，或报 LookupError	LlamaIndex 默认用 NLTK 分句且要联网下载语料；设 LLAMA_INDEX_DISABLE_NLTK=1 回退正则切分
每次提问都要等几分钟才有响应	每次运行都在重新建索引；用 storage_context.persist() 持久化，之后 load_index_from_storage 直接加载
加载索引后检索结果全部不对劲	加载时没传（或传了不同的）embedding 模型，问题向量和库内向量不在同一语义空间
第二轮问"他的主管是谁"，召回全错	问题带指代词，embedding 匹配不上；先用 LLM 结合历史改写成独立问题再检索（CondenseQuestionChatEngine）
把完整对话历史直接拼进查询，检索反而变差	长历史稀释语义；只把"改写后的独立问题"送去检索，历史留在聊天引擎里管理

七、拓展阅读：文本向量化是怎么回事

计算机没法直接理解"我喜欢吃苹果"和"我爱吃苹果"有多相似，但能计算两个向量的余弦相似度。文本向量化就是用 embedding 模型把自然语言变成向量，让"语义相近"变成"向量夹角小"。

embedding 模型的训练通常包含对比学习：输入大量标注了是否相关的文本对，让相关文本对的向量相似度变高，不相关的变低。训练完成后，模型就具备了把语义"压"进向量空间的能力。

对应到 RAG 流程里：建索引时，n 个 chunk 变成 n 个向量存进向量数据库；检索时，问题 q 变成向量 vq，在库里找与它最相似的 k 个向量，再通过向量与文本段的映射关系取回原文段。整个过程就是一次高维空间里的"找最近邻"。

小结：这篇把答疑机器人从"全靠模型自己想"升级成了"照着公司文档答"。核心动作三件：理解 RAG 两阶段流程（建索引 + 检索生成）；用 LlamaIndex 落地最小 RAG，并把索引持久化、封装成可复用模块；用问题改写解决多轮对话里的指代难题。机器人现在答得对了，但"答得好不好"还没有标准——下一节做提示词优化，让回答的风格和质量可控。