基于 RAG 架构构建企业内网文档问答机器人

前言

前面那篇 RAG 入门讲了基本原理和最小 Demo。这篇从实际部署的角度，做一个能真正给团队用的文档问答机器人。

完整的流程包括：多种格式文档导入 → 智能切片 → 向量化存储 → 检索优化 → 大模型回答 → Web 界面。

---

一、项目架构

docs/ (PDF/Word/Markdown/TXT)
  → 文档加载器
  → 文本切片器
  → Embedding 模型
  → ChromaDB 向量库
  → 检索器 (混合检索: 向量+BM25)
  → LLM (DeepSeek/通义千问)
  → Gradio Web UI

---

二、环境与依赖

pip install langchain langchain-community langchain-openai
pip install chromadb pypdf docx2txt unstructured rank-bm25
pip install gradio              # Web UI

---

三、文档加载（支持多种格式）

import os
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    UnstructuredMarkdownLoader,
    Docx2txtLoader,       # Word
    CSVLoader,             # CSV
    UnstructuredHTMLLoader
)

def load_all_docs(docs_dir):
    """加载目录下所有支持的文档格式"""
    all_docs = []
    loaders_map = {
        ".txt": TextLoader,
        ".pdf": PyPDFLoader,
        ".md": UnstructuredMarkdownLoader,
        ".docx": Docx2txtLoader,
        ".csv": CSVLoader,
        ".html": UnstructuredHTMLLoader,
    }

    for filename in os.listdir(docs_dir):
        filepath = os.path.join(docs_dir, filename)
        ext = os.path.splitext(filename)[1].lower()
        if ext not in loaders_map:
            continue

        try:
            loader = loaders_map[ext](filepath, encoding="utf-8")
            docs = loader.load()
            # 给每个文档标记来源文件
            for doc in docs:
                doc.metadata["source"] = filename
            all_docs.extend(docs)
            print(f"[OK] {filename}: {len(docs)} 段")
        except Exception as e:
            print(f"[FAIL] {filename}: {e}")

    return all_docs

---

四、智能切片

不同的文档类型需要不同的切片策略：

from langchain.text_splitter import RecursiveCharacterTextSplitter

def create_splitters():
    """根据文档类型返回不同的切片器"""
    return {
        "default": RecursiveCharacterTextSplitter(
            chunk_size=800,
            chunk_overlap=150,
            separators=["\n\n", "\n", "。", "；", ". ", " "]
        ),
        "code": RecursiveCharacterTextSplitter(
            chunk_size=1500,    # 代码块可以大一些
            chunk_overlap=100,
            separators=["\n\n", "\n", " ", ""]
        ),
    }

def smart_split(documents):
    """智能切片：对不同类型的文档用不同策略"""
    splitters = create_splitters()
    all_chunks = []

    for doc in documents:
        source = doc.metadata.get("source", "")

        # 判断文档类型
        if any(ext in source for ext in [".py", ".js", ".java", ".go", ".ts"]):
            splitter = splitters["code"]
        else:
            splitter = splitters["default"]

        chunks = splitter.split_documents([doc])
        all_chunks.extend(chunks)

    return all_chunks

---

五、向量化 + 混合检索

纯向量检索有时会漏掉关键词精确匹配的情况。加上 BM25 做混合检索效果更好：

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 向量检索
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# BM25 关键词检索（无需 Embedding，直接用文本匹配）
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 混合检索：两种结果融合排序
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # 向量占 70%，关键词占 30%
)

---

六、生成回答

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

prompt_template = """你是一个公司内部文档助手。根据以下参考文档回答用户的问题。
如果文档中没有相关信息，请明确说"根据现有文档，我无法回答这个问题"，不要编造。

参考文档：
{context}

用户问题：{question}

回答："""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

llm = ChatOpenAI(
    model="deepseek-chat",
    temperature=0.3,
    openai_api_key="sk-xxx",
    base_url="https://api.deepseek.com"
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=ensemble_retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": PROMPT}
)

def ask(question):
    result = qa_chain.invoke({"query": question})
    answer = result["result"]
    sources = list(set(
        doc.metadata.get("source", "unknown") for doc in result["source_documents"]
    ))
    return answer, sources

---

七、Gradio Web 界面

import gradio as gr

def chat_fn(message, history):
    answer, sources = ask(message)
    # 构建历史记录格式
    history.append({"role": "user", "content": message})
    history.append({"role": "assistant", "content": answer})
    return history, ""

with gr.Blocks(title="公司文档问答") as demo:
    gr.Markdown("# 公司内部文档问答系统")
    gr.Markdown("基于公司知识库的 AI 问答助手")

    chatbot = gr.Chatbot(type="messages", height=500)
    msg = gr.Textbox(placeholder="输入你的问题...", label="提问")
    clear = gr.Button("清除对话")

    msg.submit(chat_fn, [msg, chatbot], [chatbot, msg])
    clear.click(lambda: [], None, chatbot)

demo.launch(server_name="0.0.0.0", server_port=7860)

---

八、效果调优清单

在实际项目部署后，按这个顺序调优：

1. 调切片参数：如果你的文档段落很短或者表格很多，chunk_size 需要调小
2. 调检索数量 k：太多了 prompt 会超长，太少了信息不够。从 k=4 开始调
3. 调混合检索权重：如果你的文档术语多、关键词重要，调大 BM25 的权重
4. 加 Reranker：对检索结果做二次排序（如 bge-reranker-v2-m3），能明显提升准确率
5. 调 Prompt：加上”如果不知道就说不知道”这一句能大幅减少幻觉
6. 监控反馈：记录用户点了”赞”还是”踩”，回头分析那些”踩”的问题

---

以上代码组装起来就是一个完整的文档问答机器人。把 docs/ 目录换成你自己的文档，改一下 LLM 的 API Key，就能跑起来了。