2026 年如果你问企业里最火的 AI 落地场景是什么，不是 Agent，不是微调——是 RAG。

原因很简单：企业有一堆 PDF、Word、Confluence 文档，扔给大模型直接问，它要么胡说八道，要么回答"我无法访问这些文件"。RAG 直接把外部知识塞进 prompt，让模型的回答有了「出处」。

上个月给一个客户搭了一套 RAG 系统，用 LangChain + Chroma + BGE 技术栈，总共不到 300 行核心代码。这篇文章就是那个项目的精简版，你复制出来直接跑。

---

1. 环境搭建

# 创建项目目录
mkdir rag-qa-system && cd rag-qa-system
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装核心依赖
pip install langchain langchain-community langchain-text-splitters
pip install chromadb sentence-transformers
pip install openai python-dotenv unstructured
pip install "unstructured[pdf]" "unstructured[docx]"

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

LLM_API_KEY = os.getenv("DEEPSEEK_API_KEY")
LLM_BASE_URL = "https://api.deepseek.com/v1"
LLM_MODEL = "deepseek-chat"

EMBEDDING_MODEL = "BAAI/bge-large-zh-v1.5"  # BGE 中文嵌入模型
CHROMA_PERSIST_DIR = "./chroma_db"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
TOP_K_RETRIEVAL = 5

if not LLM_API_KEY:
    raise RuntimeError("请设置 DEEPSEEK_API_KEY")

---

2. 文档加载与智能分块

RAG 系统的第一道坎：怎么把 PDF 切成合适的块。太大模型塞不进上下文，太小丢失上下文关系。我试了 4 种分块策略，最后锁定了这个组合。

# loader.py — 文档加载与分块
from langchain_community.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    TextLoader,
    DirectoryLoader,
)
from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
)
from typing import List

class DocumentProcessor:
    """多格式文档加载 + 语义分块"""

    SUPPORTED_EXTS = {
        ".pdf": PyPDFLoader,
        ".docx": Docx2txtLoader,
        ".txt": TextLoader,
        ".md": TextLoader,
    }

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def load_documents(self, doc_dir: str) -> List:
        """加载目录下所有支持的文档"""
        all_docs = []
        for ext, loader_cls in self.SUPPORTED_EXTS.items():
            loader = DirectoryLoader(
                doc_dir,
                glob=f"**/*{ext}",
                loader_cls=loader_cls,
                loader_kwargs={"extract_images": False},
                silent_errors=True,
            )
            docs = loader.load()
            all_docs.extend(docs)
            print(f"  [{ext}] 加载 {len(docs)} 个文档")

        print(f"总计加载 {len(all_docs)} 个文档")
        return all_docs

    def split_documents(self, docs: List) -> List:
        """RecursiveCharacterTextSplitter + Markdown 感知"""
        # 主分割器：按段落和句子边界切分
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", " "],
            length_function=len,
        )
        chunks = text_splitter.split_documents(docs)

        # 元数据增强：给每个 chunk 标记来源
        for i, chunk in enumerate(chunks):
            source = chunk.metadata.get("source", "unknown")
            chunk.metadata["chunk_id"] = i
            chunk.metadata["source_file"] = source.split("/")[-1]

        print(f"分割完成：{len(docs)} 个文档 → {len(chunks)} 个文本块")
        return chunks


# 使用示例
processor = DocumentProcessor(chunk_size=500, chunk_overlap=50)
docs = processor.load_documents("./company_docs")
chunks = processor.split_documents(docs)

---

3. 向量嵌入与 Chroma 存储

BGE 中文模型的嵌入质量在中文语义检索上比 OpenAI text-embedding-3 强出一截，而且是免费的。Chroma 是轻量级向量数据库，适合中小规模（百万级以下）的知识库。

# vector_store.py — 向量嵌入与存储
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_chroma import Chroma
import os

class VectorStoreManager:
    """BGE 嵌入 + Chroma 向量存储"""

    def __init__(
        self,
        model_name: str = "BAAI/bge-large-zh-v1.5",
        persist_dir: str = "./chroma_db",
    ):
        # BGE embedding 配置
        self.embeddings = HuggingFaceBgeEmbeddings(
            model_name=model_name,
            model_kwargs={"device": "cuda"},      # 有 GPU 用 cuda，没有用 cpu
            encode_kwargs={
                "normalize_embeddings": True,       # BGE 建议开启归一化
                "batch_size": 32,
            },
        )
        self.persist_dir = persist_dir

    def create_from_documents(self, chunks: List, collection_name: str = "knowledge_base"):
        """从文档块创建向量库"""
        vector_store = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir,
            collection_name=collection_name,
        )
        print(f"向量库创建完成：{len(chunks)} 条向量 → {self.persist_dir}")
        return vector_store

    def load_existing(self, collection_name: str = "knowledge_base"):
        """加载已有向量库"""
        return Chroma(
            persist_directory=self.persist_dir,
            embedding_function=self.embeddings,
            collection_name=collection_name,
        )


# 使用示例
store_manager = VectorStoreManager()
vector_store = store_manager.create_from_documents(chunks)

---

4. 混合检索器：向量 + BM25

纯向量检索有一个盲区：它对精确关键词匹配不够敏感。比如你问「2025 年 Q3 的营收是多少」，向量检索可能返回 Q2 或 Q4 的数据，因为它理解「季度营收」这个语义但忽略了「Q3」这个精确约束。

解决方案是加一层 BM25 关键词检索，把两边的结果融合。

# hybrid_retriever.py — 混合检索
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from typing import List
import jieba

class HybridRetriever:
    """向量检索 + BM25 关键词检索，RRF 融合"""

    def __init__(self, vector_store, chunks: List[Document], top_k: int = 5):
        self.top_k = top_k

        # 向量检索器
        self.vector_retriever = vector_store.as_retriever(
            search_type="mmr",          # MMR 保证结果多样性
            search_kwargs={
                "k": top_k * 2,
                "fetch_k": top_k * 4,
                "lambda_mult": 0.7,      # 0=最大多样性, 1=最大相关性
            },
        )

        # BM25 关键词检索器（用 jieba 做中文分词）
        self.bm25_retriever = BM25Retriever.from_documents(
            chunks,
            preprocess_func=self._chinese_tokenize,
        )
        self.bm25_retriever.k = top_k * 2

        # 集成检索器：RRF (Reciprocal Rank Fusion)
        self.ensemble = EnsembleRetriever(
            retrievers=[self.vector_retriever, self.bm25_retriever],
            weights=[0.6, 0.4],          # 向量权重 60%，BM25 权重 40%
        )

    def _chinese_tokenize(self, text: str) -> str:
        """jieba 分词预处理，提升 BM25 中文匹配效果"""
        return " ".join(jieba.cut(text))

    def retrieve(self, query: str) -> List[Document]:
        """执行混合检索"""
        results = self.ensemble.invoke(query)[:self.top_k]
        return results

---

5. RAG 问答链：检索 + 生成

把检索到的文本块拼进 prompt，让 LLM 基于这些上下文回答。这里有一个容易被忽略的坑：如果不加 score_threshold 过滤，低相关度的 chunk 反而会拖累答案质量。

# rag_chain.py — 完整的 RAG 问答链
from openai import OpenAI
from typing import List, Dict, Optional

class RAGChain:
    """RAG 问答：检索 → 上下文组织 → LLM 生成"""

    SYSTEM_PROMPT = """你是一个企业知识库助手。
请严格基于以下【参考资料】回答用户问题。

规则：
1. 如果资料中有答案，直接引用并注明出处（文档名 + 段落编号）
2. 如果资料不包含相关信息，明确说"资料中未找到相关信息"
3. 不要编造资料中没有的内容
4. 回答要简洁，用中文"""

    def __init__(self, retriever: HybridRetriever):
        self.retriever = retriever
        self.client = OpenAI(
            api_key=os.getenv("DEEPSEEK_API_KEY"),
            base_url="https://api.deepseek.com/v1",
        )

    def _format_context(self, docs: List[Document]) -> str:
        """将检索结果格式化为上下文"""
        parts = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("source_file", "unknown")
            chunk_id = doc.metadata.get("chunk_id", i)
            parts.append(
                f"【资料 {i}】来源: {source} | 段落: {chunk_id}\n{doc.page_content}\n"
            )
        return "\n---\n".join(parts)

    def query(self, question: str, chat_history: Optional[List] = None) -> Dict:
        """执行一次 RAG 问答"""
        # 1. 检索
        relevant_docs = self.retriever.retrieve(question)

        if not relevant_docs:
            return {
                "answer": "未找到相关文档。请检查知识库是否包含该主题的文档。",
                "sources": [],
            }

        # 2. 组织上下文
        context = self._format_context(relevant_docs)

        # 3. 构建消息
        messages = [{"role": "system", "content": self.SYSTEM_PROMPT}]

        # 加入对话历史（最近 3 轮）
        if chat_history:
            messages.extend(chat_history[-6:])

        messages.append({
            "role": "user",
            "content": f"【参考资料】\n{context}\n\n【用户问题】\n{question}",
        })

        # 4. LLM 生成
        response = self.client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            temperature=0.3,          # 低温度保证答案一致性
            max_tokens=2048,
        )

        return {
            "answer": response.choices[0].message.content,
            "sources": [
                {
                    "file": doc.metadata.get("source_file", ""),
                    "chunk_id": doc.metadata.get("chunk_id"),
                    "content_preview": doc.page_content[:100],
                }
                for doc in relevant_docs
            ],
        }


# 完整使用示例
retriever = HybridRetriever(vector_store, chunks, top_k=5)
rag = RAGChain(retriever)

result = rag.query("公司2025年的研发投入是多少？")
print(f"回答: {result['answer']}")
print(f"引用来源: {len(result['sources'])} 篇文档")
for src in result['sources']:
    print(f"  - {src['file']} (段落 {src['chunk_id']})")

---

6. 带对话记忆的多轮问答

上面那个只能一问一答。真实场景里用户会追问，需要把上一轮检索到的上下文也带进去。

# chat_rag.py — 带记忆的多轮 RAG
class ChatRAG(RAGChain):
    """支持多轮对话的 RAG 系统"""

    def __init__(self, retriever: HybridRetriever, max_history: int = 10):
        super().__init__(retriever)
        self.sessions: Dict[str, List] = {}
        self.max_history = max_history

    def chat(self, session_id: str, question: str) -> Dict:
        """带记忆的对话接口"""
        if session_id not in self.sessions:
            self.sessions[session_id] = []

        history = self.sessions[session_id]
        result = self.query(question, chat_history=history)

        # 记录对话历史
        history.append({"role": "user", "content": question})
        history.append({"role": "assistant", "content": result["answer"]})

        # 限制历史长度
        if len(history) > self.max_history * 2:
            self.sessions[session_id] = history[-self.max_history * 2:]

        return result


# 多轮对话测试
chat_rag = ChatRAG(retriever, max_history=10)

questions = [
    "公司的研发团队有多少人？",
    "那他们的主要研究方向是什么？",
    "去年发了多少篇论文？",
]

for q in questions:
    result = chat_rag.chat(session_id="user_001", question=q)
    print(f"\nQ: {q}")
    print(f"A: {result['answer'][:200]}...")

---

7. 效果评估：用 Ragas 打分

光说「效果好」没意义。用 Ragas 框架对检索和生成质量做量化评估。

# eval_rag.py — Ragas 评估
from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy,
)
from datasets import Dataset

def evaluate_rag(rag_chain, test_questions: List[Dict]):
    """
    test_questions = [
        {"question": "研发投入多少?", "ground_truth": "2025年研发投入5.2亿"},
        ...
    ]
    """
    eval_data = {"question": [], "answer": [], "contexts": [], "ground_truth": []}

    for item in test_questions:
        result = rag_chain.query(item["question"])
        eval_data["question"].append(item["question"])
        eval_data["answer"].append(result["answer"])
        eval_data["contexts"].append([s["content_preview"] for s in result["sources"]])
        eval_data["ground_truth"].append(item["ground_truth"])

    dataset = Dataset.from_dict(eval_data)
    scores = evaluate(
        dataset,
        metrics=[context_precision, context_recall, faithfulness, answer_relevancy],
    )

    print(f"Context Precision:  {scores['context_precision']:.3f}")
    print(f"Context Recall:     {scores['context_recall']:.3f}")
    print(f"Faithfulness:       {scores['faithfulness']:.3f}")
    print(f"Answer Relevancy:   {scores['answer_relevancy']:.3f}")
    return scores

---

踩坑笔记

踩了几个坑，记下来省你时间：

1. BGE 模型首次下载 — 默认从 HuggingFace 拉模型，国内网络大概率超时。先 export HF_ENDPOINT=https://hf-mirror.com，换镜像源。
2. Chroma 的 persist 时机 — from_documents() 之后 Chroma 会自动持久化，但后续 add_documents() 必须要手动调 persist()，不然重启丢数据。
3. Chunk size 不是越大越好 — 我实测 500 字符的中文 chunk 在 BGE 上效果最好，超过 800 检索精度反而下降。
4. MMR 的 lambda_mult 要调到 0.7 — 默认值 0.5 多样性太激进，会把最相关的几个文档挤到后面去。

---

金句

"RAG 不是什么高级技术，它就是解决了一个朴素的问题——大模型没有看过你的内部文档。但朴素到极致，就是生产级。"

---

如果你也在搭 RAG 系统，或者遇到了分块/检索效果不好的问题，评论区说说你的场景——我看看能不能给点建议。