RAG（检索增强生成）解决的就是这个问题：把私有文档切成小块、向量化存起来，用户提问的时候先检索相关片段，再交给大模型回答。

FAISS 是 Meta 开源的向量检索库，不需要 GPU，普通电脑就能跑。

先说思路

做 RAG 一共四步：

1. 把文档切分成小段

2. 每段转成向量（embedding）

3. 用户提问，把问题也转成向量

4. 在库里找最相似的几段文本，交给大模型回答

每一步都有现成的库，用 Python 拼起来就行。

安装依赖

```bash

pip install faiss-cpu sentence-transformers

```

faiss-cpu 是 CPU 版的向量检索引擎，sentence-transformers 负责把文本转成向量。

准备知识库

先把文档拆成片段。按段落切，每个片段不要太长，200-500 字左右比较合适。

```python

# chunker.py

import re

from pathlib import Path

def load_documents(data_dir: str) -> list[str]:

    """加载目录下所有 .txt 文件，按段落切分"""

    chunks = []

    for filepath in Path(data_dir).glob("*.txt"):

        text = filepath.read_text(encoding="utf-8")

        # 按空行切段落

        paragraphs = re.split(r"\n\s*\n", text.strip())

        for para in paragraphs:

            para = para.strip()

            if len(para) < 20:

                continue  # 太短的不存

            if len(para) > 1000:

                # 超长的继续按句号切

                sentences = re.split(r"[。！？]", para)

                chunk = ""

                for s in sentences:

                    if not s.strip():

                        continue

                    if len(chunk) + len(s) < 500:

                        chunk += s + "。"

                    else:

                        if chunk.strip():

                            chunks.append(chunk.strip())

                        chunk = s + "。"

                if chunk.strip():

                    chunks.append(chunk.strip())

            else:

                chunks.append(para)

    return chunks

if __name__ == "__main__":

    chunks = load_documents("./docs")

    print(f"共切分 {len(chunks)} 个片段")

    for i, c in enumerate(chunks[:3]):

        print(f"\n--- 片段 {i+1} ({len(c)}字) ---")

        print(c[:100])

```

建向量库

把切好的片段转成向量，存进 FAISS。

```python

# build_index.py

import pickle

from pathlib import Path

from sentence_transformers import SentenceTransformer

import faiss

import numpy as np

from chunker import load_documents

def build_index(data_dir: str, model_name: str = "BAAI/bge-small-zh-v1.5"):

    """加载文档、转向量、建索引"""

    # 1. 加载并切分文档

    chunks = load_documents(data_dir)

    print(f"文档片段数: {len(chunks)}")

    # 2. 加载 embedding 模型

    print(f"加载模型: {model_name}")

    model = SentenceTransformer(model_name)

    # 3. 转向量

    print("转向量中...")

    embeddings = model.encode(chunks, show_progress_bar=True)

    dim = embeddings.shape[1]

    print(f"向量维度: {dim}")

    # 4. 建 FAISS 索引

    index = faiss.IndexFlatIP(dim)  # 内积（余弦相似度）

    index.add(embeddings.astype(np.float32))

    print(f"索引数量: {index.ntotal}")

    # 5. 保存

    output_dir = Path("./kb_index")

    output_dir.mkdir(exist_ok=True)

    faiss.write_index(index, str(output_dir / "index.faiss"))

    with open(output_dir / "chunks.pkl", "wb") as f:

        pickle.dump(chunks, f)

    print("索引已保存")

if __name__ == "__main__":

    build_index("./docs")

```

这里用了 `BAAI/bge-small-zh-v1.5`，北京智源专门为中文训练的 embedding 模型，模型文件约 30MB。第一次运行自动下载。

检索 + 回答

用户提问时：把问题转向量 → FAISS 检索最相似的片段 → 拼 prompt → 调大模型回答。

```python

# rag_query.py

import pickle

from pathlib import Path

import faiss

import numpy as np

from sentence_transformers import SentenceTransformer

class RAGEngine:

    def __init__(self, index_dir: str = "./kb_index",

                 model_name: str = "BAAI/bge-small-zh-v1.5"):

        # 加载索引

        self.index = faiss.read_index(str(Path(index_dir) / "index.faiss"))

        with open(Path(index_dir) / "chunks.pkl", "rb") as f:

            self.chunks = pickle.load(f)

        # 加载模型

        self.model = SentenceTransformer(model_name)

        self.k = 3  # 默认检索 3 个片段

    def search(self, query: str, k: int = None) -> list[tuple[str, float]]:

        """检索最相关的文档片段"""

        if k is None:

            k = self.k

        vec = self.model.encode([query]).astype(np.float32)

        scores, indices = self.index.search(vec, k)

        results = []

        for score, idx in zip(scores[0], indices[0]):

            if idx >= 0 and idx < len(self.chunks):

                results.append((self.chunks[idx], float(score)))

        return results

    def build_prompt(self, query: str, contexts: list[tuple[str, float]]) -> str:

        """拼 prompt"""

        context_text = "\n\n".join(f"[相关度 {s:.2f}] {c}" for c, s in contexts)

        return f"""请根据以下参考资料回答问题。

如果参考资料中没有相关信息，请如实说不知道，不要编造。

参考资料：

{context_text}

问题：{query}

回答："""

    def query(self, query: str, llm_chat_fn=None) -> str:

        """完整查询：检索 + 回答"""

        contexts = self.search(query)

        prompt = self.build_prompt(query, contexts)

        if llm_chat_fn:

            return llm_chat_fn(prompt)

        # 没传大模型时，打印参考内容

        print(f"问题: {query}\n")

        for i, (ctx, score) in enumerate(contexts, 1):

            print(f"--- 参考 {i} (相关度: {score:.3f}) ---")

            print(ctx[:200])

        return prompt

if __name__ == "__main__":

    engine = RAGEngine()

    engine.query("你们的退货政策是什么？")

```

连上大模型

把之前写的 LLMClient 接进来，就成完整的 RAG 系统了。

```python

# rag_demo.py

from llm_client import LLMClient, Message

from rag_query import RAGEngine

def main():

    # 初始化 RAG

    engine = RAGEngine()

    # 初始化大模型（用 deepseek 便宜）

    client = LLMClient(

        api_key="your-api-key",

        base_url="https://api.deepseek.com",

        model="deepseek-chat",

    )

    print("=== RAG 知识库问答（输入 /exit 退出）===")

    print("知识库已加载，共 {} 个片段".format(len(engine.chunks)))

    while True:

        query = input("\n问题: ").strip()

        if query.lower() in ("/exit", "/quit"):

            break

        # 检索相关片段

        contexts = engine.search(query)

        # 如果有高度相关的结果，用 RAG；否则直接问模型

        if contexts and contexts[0][1] > 0.5:

            prompt = engine.build_prompt(query, contexts)

            messages = [

                Message(role="system", content="你是一个知识库助手。"),

                Message(role="user", content=prompt),

            ]

            print("\n回答: ", end="", flush=True)

            for chunk in client.chat_stream(messages):

                print(chunk, end="", flush=True)

            print()

            # 显示参考来源

            print("\n--- 参考来源 ---")

            for i, (ctx, score) in enumerate(contexts, 1):

                print(f"  [{i}] 相关度 {score:.2f} | {ctx[:60]}...")

        else:

            # 没找到相关文档，直接问模型

            print("（未在知识库中找到相关信息，直接询问模型）")

            messages = [Message(role="user", content=query)]

            for chunk in client.chat_stream(messages):

                print(chunk, end="", flush=True)

            print()

if __name__ == "__main__":

    main()

```

一份完整的知识库文件

```txt

# docs/product.txt

产品退货政策

本店支持7天无理由退货。退货条件：商品未拆封、不影响二次销售。

运费说明：质量问题由卖家承担运费，非质量问题由买家承担。

退款时间：收到退货后3个工作日内处理。

产品保修政策

电子产品保修期为1年。保修范围包括非人为损坏的功能故障。

以下情况不在保修范围：人为损坏、私自拆修、进液、摔损。

保修期内凭购买凭证享受免费维修服务。

常见问题解答

Q：如何查询订单物流？

A：登录后进入"我的订单"，点击对应订单查看物流详情。

Q：如何联系客服？

A：工作日9:00-18:00在线客服，也可拨打400-xxx-xxxx。

```

测试效果：

```bash

# 建索引

python build_index.py

# 问问题

python rag_demo.py

```

问题"退货政策是什么？"会直接从知识库检索并回答。问题"今天天气怎么样？"因为知识库里没有，会直接交给大模型处理。

几个实际用的坑

1. embedding 模型选中文专用的。`BAAI/bge-small-zh-v1.5` 对中文效果比通用模型好很多。如果用英文模型处理中文文档，检索准确率会明显下降。

2. 文档切分粒度影响很大。切太粗，一个片段包含多个话题，检索不准。切太细，每个片段信息量不够，大模型得不到完整上下文。200-500 字比较合适，自己根据文档类型调。

3. 检索出来的片段要排序。相关度高的放前面，低的可丢弃。设置一个阈值（比如 0.5），低于阈值的不要，避免噪声干扰大模型。

4. 大模型的选择。RAG 检索出来的内容，用便宜的小模型就够了——DeepSeek、GLM-4-Flash 这些就够用，不需要 GPT-4。因为答案已经在你给的上下文里了，模型只需要做"阅读理解"。

5. 索引需要定期更新。文档变了，要重新跑 build_index.py。可以做成定时任务，每天凌晨跑一次。

6. 大规模场景用 faiss-gpu。上面用的是 faiss-cpu，几万条文档没问题。几十万条以上可以考虑 faiss-gpu，或者用 Milvus、Qdrant 这类专业向量数据库。