RAG 知识库实战——从文档到能问答的 Agent

你有一堆公司文档——产品手册、规章制度、历史案例——你想让 AI 能回答关于这些内容的问题。第一反应是把文档全塞进 system prompt？试过就知道，几万字的内容塞进去，模型会开始糊弄你，而且 token 费用直接炸了。

RAG（检索增强生成）就是解这道题的标准方法：不是让模型背下全部文档，而是考试前先翻书，找到相关段落再回答。 开卷考试——这一个比喻，基本说清了 RAG 的本质。

这一节我们不只讲原理，直接带你走完从"文档文件夹"到"能问答的 Agent"的完整路径，每一步给代码，给"你应该看到什么"，给坑在哪。

---

RAG 是什么：开卷考试比喻

闭卷考试 = 让模型"背住"所有知识，考试时靠记忆回答。问题：模型背了一堆通用知识，但没背你公司的内部文档；就算把文档全塞进上下文，也会超限、变贵、变糊。

开卷考试 = RAG：考试开始，先查书，找到相关的几段，再写答案。

具体来说：

1. 考试前先把书"索引好"——把文档切成段落，把每段转成向量，存进向量数据库
2. 问题来了，先查索引——把问题也转成向量，找出最相关的几段
3. 把找到的段落 + 原始问题一起交给模型——让模型"看着这几段"写答案

这样：不超限（只把相关片段塞进去，不是全文）、不幻觉（模型有明确依据可引用）、成本可控（不是每次都发整本书）。

RAG 的正式英文名是 Retrieval-Augmented Generation，"检索增强生成"。不熟悉这个术语的，可以先看 RAG 是什么。

---

五步流程全景图

把一堆文档变成可问答的知识库，五步缺一不可：

文档文件夹
    ↓ 第一步：切分（chunking）
    ↓ 第二步：embedding（转向量）
    ↓ 第三步：存向量库
          ↑
      第四步：检索（用问题向量找最相关 chunk）
          ↓
      第五步：生成（把 chunk + 问题交给模型）
                    ↓
                 回答

每一步都有细节，下面逐一讲。

---

第一步：切分文档（chunking）

为什么要切

不切行不行？不行。理由：

1. embedding 模型有长度限制：大多数 embedding 模型单次最多几百到几千个 token，一篇几千字的文章直接丢进去，尾部内容被截断或失真
2. 检索粒度太粗：你想找"退货流程"，但整篇文章 embedding 是"用户手册总体语义"，这个向量未必能精准命中
3. 塞进上下文太贵：就算能检索到，把整篇文章塞给模型，token 开销是按 chunk 塞的 10 倍起

切多大合适

没有万能答案，但有经验参考：

场景	推荐 chunk 大小	理由
FAQ、短问答文档	150～300 字	每段本身就是独立问答单元
产品手册、规章	400～800 字	每段包含一个完整的功能或条款
长篇报告、研究文章	600～1200 字	需要一定上下文才能理解

通用原则：切分后的每个 chunk，要能"独立被读懂"，不能切断一个完整的语义单元。 切断了一个条款的前后，检索出来模型也没法用。

另外，相邻 chunk 之间建议留 20% 的重叠（overlap），避免关键信息刚好落在切割边界被一刀两断。

代码：切分一个文本文件

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 100) -> list[str]:
    """
    按字符数切分文本，相邻 chunk 留 overlap 字符重叠。
    生产环境可换成按句子/段落边界切，语义更完整。
    """
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk)
        start = end - overlap  # 留重叠
    return chunks

# 读一个文档
with open("company_docs.txt", "r", encoding="utf-8") as f:
    raw_text = f.read()

chunks = chunk_text(raw_text, chunk_size=500, overlap=100)
print(f"切分出 {len(chunks)} 个 chunk，第一个：\n{chunks[0][:100]}...")

你应该看到：终端打印 chunk 数量，第一个 chunk 的前 100 字。

---

第二步：embedding（把文字转成向量）

什么是 embedding

向量就是一串数字，比如 [0.12, -0.34, 0.78, ...]，通常有几百到上千个维度。语义相近的文字，对应的向量在空间里也相近——这是检索能工作的数学基础。

把"退货政策"转成向量，把"我能退款吗"也转成向量，两个向量方向相近，搜索时就能找到。

代码：批量 embedding

import anthropic

client = anthropic.Anthropic()  # 自动读取 ANTHROPIC_API_KEY

def embed_chunks(chunks: list[str]) -> list[list[float]]:
    """
    批量获取 chunks 的 embedding 向量。
    embedding 模型名以官方文档为准（截稿 2026-06）：
    https://docs.anthropic.com/en/docs/build-with-claude/embeddings
    """
    embeddings = []
    for i, chunk in enumerate(chunks):
        response = client.messages.create(
            model="claude-opus-4-5",  # 以官方文档为准（截稿 2026-06）
            max_tokens=10,
            messages=[{"role": "user", "content": chunk}],
        )
        # ⚠️ 注意：Anthropic 的 embedding API 与 messages API 分开。
        # 实际使用请查阅官方 embedding 端点文档（截稿 2026-06）。
        # 下方 embed_texts() 展示了使用专用 embedding 库的范式。
        _ = response  # 占位，实际 embedding 请用专用端点
        embeddings.append([0.0] * 768)  # 占位向量，替换为真实 embedding
        if (i + 1) % 10 == 0:
            print(f"已处理 {i+1}/{len(chunks)} 个 chunk")
    return embeddings

更推荐的方式：用专门的 embedding 库（比如 sentence-transformers 本地跑，或 OpenAI / Cohere 的 embedding API），专门为语义检索优化，成本也更低。以下是使用 sentence-transformers 本地 embedding 的范式：

# pip install sentence-transformers
from sentence_transformers import SentenceTransformer

# 模型名以 Hugging Face 官方为准（截稿 2026-06）
model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")

def embed_texts(texts: list[str]) -> list[list[float]]:
    """返回每段文本的 embedding 向量列表"""
    embeddings = model.encode(texts, show_progress_bar=True)
    return embeddings.tolist()

chunk_embeddings = embed_texts(chunks)
print(f"向量维度：{len(chunk_embeddings[0])}，共 {len(chunk_embeddings)} 条")

你应该看到：进度条跑完，打印向量维度（取决于模型，以官方为准）和条数。

---

第三步：存向量库

向量库的职责：存下（chunk 文本 + 对应向量），后续支持快速相似度检索。

入门阶段最轻量的选择是 FAISS（Meta 开源，纯本地，无需部署服务）；生产环境可以换 Milvus / Qdrant / Weaviate 等，API 结构类似。

# pip install faiss-cpu numpy
import faiss
import numpy as np
import json

def build_vector_store(chunks: list[str], embeddings: list[list[float]]) -> tuple:
    """
    构建 FAISS 索引，返回 (index, chunks) 以便检索时取回原文。
    以 faiss 官方文档为准（截稿 2026-06）：https://faiss.ai
    """
    vectors = np.array(embeddings, dtype="float32")
    dimension = vectors.shape[1]

    # IndexFlatL2：按欧氏距离（L2）做精确检索，适合 10 万条以内
    index = faiss.IndexFlatL2(dimension)
    index.add(vectors)

    print(f"向量库构建完成：{index.ntotal} 条向量，维度 {dimension}")
    return index, chunks

index, chunk_store = build_vector_store(chunk_embeddings, chunk_embeddings)

---

第四步：检索（用问题向量找最相关 chunk）

问题来了，先把问题转成向量，再在向量库里找"最近邻"。

def retrieve(query: str, index, chunk_store: list[str],
             top_k: int = 3) -> list[str]:
    """
    把问题转向量，检索向量库，返回最相关的 top_k 个 chunk。
    """
    query_embedding = embed_texts([query])
    query_vec = np.array(query_embedding, dtype="float32")

    distances, indices = index.search(query_vec, top_k)

    results = []
    for i, idx in enumerate(indices[0]):
        if idx != -1:
            results.append(chunk_store[idx])
            print(f"  第{i+1}条（距离 {distances[0][i]:.4f}）：{chunk_store[idx][:60]}...")
    return results

relevant_chunks = retrieve("我们的退货政策是什么？", index, chunk_store, top_k=3)

你应该看到：打印出 3 条 chunk 的前 60 字，以及各自的 L2 距离（距离越小越相关）。

---

第五步：把检索到的内容塞进上下文，让模型回答

import anthropic

client = anthropic.Anthropic()

def rag_answer(query: str, context_chunks: list[str]) -> str:
    """
    把检索到的 chunk 拼成 context，连同问题一起发给模型，让它基于 context 回答。
    """
    context = "\n\n---\n\n".join(context_chunks)

    system_prompt = (
        "你是一个知识库问答助手。请根据下方提供的【参考资料】回答用户问题。\n"
        "如果参考资料中没有足够信息，请直接说'我在知识库中没有找到相关内容'，不要编造。\n\n"
        f"【参考资料】\n{context}"
    )

    response = client.messages.create(
        model="claude-opus-4-5",  # 以官方文档为准（截稿 2026-06）
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": query}],
    )

    answer = response.content[0].text
    return answer

answer = rag_answer("我们的退货政策是什么？", relevant_chunks)
print(f"\n问题：我们的退货政策是什么？")
print(f"回答：{answer}")

你应该看到：模型基于检索到的 chunk 内容，给出有据可查的回答——而不是凭空捏造。

---

最小可跑 RAG 骨架（完整拼装版）

把上面五步拼在一起，就是一个端到端能跑的最小 RAG 系统：

"""
最小可跑 RAG 骨架
依赖：pip install anthropic sentence-transformers faiss-cpu numpy
embedding 模型名以 Hugging Face 官方为准（截稿 2026-06）
"""

import anthropic
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

# ─── 配置 ───────────────────────────────────────────────────────────────────
EMBED_MODEL = "paraphrase-multilingual-MiniLM-L12-v2"  # 以官方为准（截稿 2026-06）
CHUNK_SIZE = 500
OVERLAP = 100
TOP_K = 3

# ─── 初始化 ──────────────────────────────────────────────────────────────────
embed_model = SentenceTransformer(EMBED_MODEL)
llm_client = anthropic.Anthropic()

# ─── 步骤一：切分 ────────────────────────────────────────────────────────────
def chunk_text(text: str, size: int = CHUNK_SIZE, overlap: int = OVERLAP) -> list[str]:
    chunks, start = [], 0
    while start < len(text):
        chunk = text[start: start + size]
        if chunk.strip():
            chunks.append(chunk)
        start += size - overlap
    return chunks

# ─── 步骤二+三：embedding & 建索引 ──────────────────────────────────────────
def build_index(docs: list[str]) -> tuple:
    """接收原始文档列表，切分→embedding→建 FAISS 索引"""
    all_chunks = []
    for doc in docs:
        all_chunks.extend(chunk_text(doc))

    print(f"[1/3] 切分完成：{len(all_chunks)} 个 chunk")

    vectors = embed_model.encode(all_chunks, show_progress_bar=True).astype("float32")
    print(f"[2/3] Embedding 完成：维度 {vectors.shape[1]}")

    index = faiss.IndexFlatL2(vectors.shape[1])
    index.add(vectors)
    print(f"[3/3] 向量库构建完成：{index.ntotal} 条")

    return index, all_chunks

# ─── 步骤四：检索 ────────────────────────────────────────────────────────────
def retrieve(query: str, index, chunks: list[str]) -> list[str]:
    query_vec = embed_model.encode([query]).astype("float32")
    distances, indices = index.search(query_vec, TOP_K)
    return [chunks[i] for i in indices[0] if i != -1]

# ─── 步骤五：生成 ────────────────────────────────────────────────────────────
def answer(query: str, context_chunks: list[str]) -> str:
    context = "\n\n---\n\n".join(context_chunks)
    resp = llm_client.messages.create(
        model="claude-opus-4-5",  # 以官方文档为准（截稿 2026-06）
        max_tokens=1024,
        system=(
            "你是一个知识库问答助手。根据下方【参考资料】回答问题。"
            "资料中没有的内容请说'知识库中未找到'，不要编造。\n\n"
            f"【参考资料】\n{context}"
        ),
        messages=[{"role": "user", "content": query}],
    )
    return resp.content[0].text

# ─── 跑起来 ──────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    # 示例文档（替换成你自己的文档内容）
    sample_docs = [
        "公司退货政策：自购买之日起 30 天内可无理由退货，超过 30 天不予受理。"
        "退货须携带原始凭证和完好包装，否则扣除 20% 损耗费。",
        "员工假期制度：正式员工每年享有 15 天带薪年假，工龄 5 年以上增加至 20 天。"
        "年假须提前 3 个工作日申请，经直属主管审批后方可生效。",
        "产品质量保证：所有产品提供一年有限保修，人为损坏和非正常使用不在保修范围内。"
        "保修期内免费维修，保修期外按成本收费。",
    ]

    index, chunks = build_index(sample_docs)

    queries = [
        "我买了东西，两个月后想退货，可以吗？",
        "工作三年的员工有几天年假？",
        "产品坏了保修吗？",
    ]

    for q in queries:
        relevant = retrieve(q, index, chunks)
        ans = answer(q, relevant)
        print(f"\n❓ {q}")
        print(f"💬 {ans}")

你应该看到什么

[1/3] 切分完成：X 个 chunk
[2/3] Embedding 完成：维度 384
[3/3] 向量库构建完成：X 条

❓ 我买了东西，两个月后想退货，可以吗？
💬 根据公司退货政策，退货须在购买之日起 30 天内提出，您提到的两个月（约60天）已超过退货期限，无法受理退货申请。

❓ 工作三年的员工有几天年假？
💬 工龄三年的员工属于正式员工范畴，每年享有 15 天带薪年假（工龄 5 年以上才增加至 20 天）。

❓ 产品坏了保修吗？
💬 如属正常使用损坏，在购买后一年内可享受免费保修服务；若为人为损坏或非正常使用，则不在保修范围。

关键特征：模型的回答有明确依据，不是编造的——如果你把 sample_docs 里没有的内容拿来问，它会说"知识库中未找到"而不是瞎猜。这正是 RAG 和纯 LLM 问答的本质区别。

---

常见坑与故障排查表

症状	根因	解法
检索出来的 chunk 跟问题毫不相关	embedding 模型语言不匹配（用了英文模型处理中文）或 chunk 太大（语义被稀释）	换多语言 embedding 模型；把 chunk_size 减小到 300～500
模型回答"知识库中未找到"，但文档明明有	chunk 切在了关键信息的两侧；或 TOP_K 太小，相关 chunk 排在第 4、5 条以后	增大 overlap；把 TOP_K 从 3 调到 5；检查切分后的 chunk 人眼扫一遍
模型回答包含错误信息，但 chunk 里明明写对了	context 里有多个矛盾的 chunk，模型在拼凑	加去重逻辑（余弦相似度 > 0.95 的 chunk 只保留一条）；优化 system prompt 让模型遇到矛盾时标注出来
每次问同样的问题，回答不稳定	检索结果顺序每次略有不同，导致进入 context 的内容不同	对检索结果按相似度评分排序后固定顺序；或改用 MMR（最大边际相关性）检索策略
问题是"请列出所有退货条款"，回答只给了一条	TOP_K 太小，相关 chunk 分散在多个片段，只捞到了一条	增大 TOP_K；或在 query 前加扩充指令："展开所有相关细节"
Embedding 速度极慢，批量处理卡住	模型在 CPU 上逐条处理，没批处理	encode() 传入整个列表而不是逐条调用；加 batch_size 参数

---

常见问题

Q：RAG 和直接把文档塞进 system prompt 有什么区别？

直接塞的方式叫"全文注入"，问题有三：① token 上限（大文档超限直接报错）② 成本（每次请求都要发完整文档）③ 质量（模型面对几万字的上下文，注意力会分散，容易答偏）。RAG 只把"跟这个问题最相关的几段"塞进去，三个问题都解决了。代价是需要提前建索引、引入向量库。如果文档总字数在 2000 字以内，直接塞也未尝不可；超过这个量，RAG 是标准答案。

Q：向量库我必须用 FAISS 吗？有没有更简单的？

FAISS 是入门首选，零服务依赖、本地跑、够快。文档量在几十万条以内完全够用。如果你想要持久化、多用户、分布式，可以换 Qdrant（开源，Docker 一行起服务）或 Chroma（Python 原生，适合实验）。结构是一样的，只是客户端 API 写法不同，以各自官方文档为准（截稿 2026-06）。

Q：用户问了知识库里没有的问题，模型会乱说吗？

取决于你的 system prompt。示例里我们写了"资料中没有的内容请说'知识库中未找到'，不要编造"——这条指令对 Claude 系列模型很有效。但没有 100% 保证，建议在关键场景加一层校验：如果检索到的 chunk 相似度都低于某个阈值，直接回复"未在知识库中找到相关内容"，不走模型生成。

Q：RAG 和向量记忆（第 4.2 节讲的）是同一个东西吗？

结构上非常相似，都是"向量检索+上下文注入"，但用途不同。向量记忆（见 三层记忆架构：短期、向量、结构化）是让 Agent 记住历史对话，检索的是过去的交互记录。RAG 知识库检索的是外部文档。本质机制相同，只是索引里存的内容不同——明白了这一点，两种场景切换起来很顺。

---

增量：给 RAG 加"引用来源"让回答可溯源

上面的示例模型直接给答案，用户不知道是哪段文档里说的。加一个小改动，让模型把引用的来源也标出来：

修改 system_prompt：

"……【参考资料】\n"
"[段落1]：{chunk_1}\n[段落2]：{chunk_2}\n[段落3]：{chunk_3}\n\n"
"回答时，请在结尾标注你引用了哪几个段落（如：引用：[段落1][段落3]）。"

这样用户能追溯回答依据，发现答错时也知道去改哪段文档——这是企业级 RAG 的标配。

关于如何在客服场景落地带 RAG 的 Agent，可以参考 客服数字员工实战落地（L6 篇）中的完整链路设计。

---

小结

• RAG = 开卷考试：先检索相关文档片段，再让模型基于它回答——不是让模型背全文
• 五步缺一不可：切分 → embedding → 存向量库 → 检索 → 生成
• 切分是最容易踩坑的一步：太大语义稀释，太小上下文缺失，关键词落在边界；overlap 是标配
• embedding 模型选多语言版，否则中文检索不准
• system prompt 里明确"找不到就说找不到"，是防幻觉的第一道门

跑通了这个骨架，后续可以往两个方向走：更准（改进切分策略、换更好的 embedding 模型、加 reranker）和更大（接真实向量数据库、支持增量更新、多文档来源混合检索）。

想了解 Agent 怎么把 RAG 用在记忆层，看 三层记忆架构：短期、向量、结构化（4.2 节）；想看 RAG 在客服 Agent 里的完整落地流程，看 客服数字员工实战落地；先了解 RAG 概念层，看 RAG 是什么；或者直接回到 AI Agent 智能体阶梯 看整体路径。

👉 看看 AI 数字员工落地指南，或了解 数字员工搭建实战课。需要为企业落地方案，欢迎找我们聊 企业服务。