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

前言

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

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

---

一、项目架构

---

二、环境与依赖

pip install langchain langchain-community langchain-openai
pip install chromadb pypdf docx2txt unstructured rank-bm25
pip install flashrank          # 高性能轻量 Rerank 重排序库
pip install gradio              # Web UI

---

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

避坑注意：不同文档 Loader 的初始化参数差异很大。例如 PyPDFLoader 与 Docx2txtLoader 是通过底层二进制解析库提取内容的，不能像文本加载器那样传入 encoding="utf-8"，否则会直接抛出 TypeError: __init__() got an unexpected keyword argument 'encoding'。

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

def load_all_docs(docs_dir):
    """加载目录下所有支持的文档格式（安全适配各格式 Loader）"""
    all_docs = []

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

        try:
            # 根据文件类型分别实例化，防止非法参数崩溃
            if ext == ".txt":
                loader = TextLoader(filepath, encoding="utf-8")
            elif ext == ".pdf":
                loader = PyPDFLoader(filepath)
            elif ext == ".docx":
                loader = Docx2txtLoader(filepath)
            elif ext == ".csv":
                loader = CSVLoader(filepath, encoding="utf-8")
            elif ext == ".md":
                loader = UnstructuredMarkdownLoader(filepath)
            elif ext == ".html":
                loader = UnstructuredHTMLLoader(filepath)
            else:
                continue

            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

---

五、向量化 + 混合检索 + Rerank 重排序

纯向量检索容易受语义宽泛干扰，而 BM25 只能匹配字面。工业级最佳实践是：BM25 + 向量检索初步召回（Top 10）→ Rerank 重排序模型交叉注意力精排（Top 3）：

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

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

# 1. 向量语义检索 (召回 Top 10)
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# 2. BM25 关键词精确检索 (召回 Top 10)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 10

# 3. 混合检索（RRF 倒数秩融合）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 向量占 60%，关键词占 40%
)

# 4. 加入轻量级 Cross-Encoder 进行重排序（精炼出 Top 3 最相关切片）
compressor = FlashrankRerank(top_n=3)
final_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)

---

六、生成回答（现代 LCEL 架构）

在 LangChain 0.2+ 中，官方已弃用旧版 RetrievalQA，推荐使用清晰声明式的 LCEL（LangChain Expression Language） 管道构建问答链：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain

# 1. 初始化模型
llm = ChatOpenAI(
    model="deepseek-chat",
    temperature=0.2,
    openai_api_key="sk-xxx",
    base_url="https://api.deepseek.com"
)

# 2. 严谨的系统 Prompt，杜绝幻觉
system_prompt = (
    "你是一个严谨的企业内部文档智能助手。请严格根据提供的参考文档回答用户问题。\n"
    "【规则】：\n"
    "1. 如果参考文档中没有相关事实，请直接回答：'抱歉，在现有内网知识库文档中未找到相关说明。'，严禁胡编乱造！\n"
    "2. 保持回答客观、专业、条理清晰。\n\n"
    "参考文档片段：\n{context}"
)

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
])

# 3. 组装 LCEL 问答管道
question_answer_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(final_retriever, question_answer_chain)

def ask(question: str):
    """执行检索与生成，返回回答和去重后的引用源"""
    response = rag_chain.invoke({"input": question})
    answer = response["answer"]
    
    # 提取并去重引用文档来源
    sources = list({doc.metadata.get("source", "未知文档") for doc in response["context"]})
    return answer, sources

---

七、Gradio Web 界面

import gradio as gr

def chat_fn(message, history):
    if not message.strip():
        return history, ""
    
    answer, sources = ask(message)
    
    # 在回答底部附上文档引用说明
    if sources:
        sources_str = "、".join([f"`{s}`" for s in sources])
        formatted_answer = f"{answer}\n\n> 📚 **参考知识库文档**：{sources_str}"
    else:
        formatted_answer = answer

    history.append({"role": "user", "content": message})
    history.append({"role": "assistant", "content": formatted_answer})
    return history, ""

with gr.Blocks(title="企业内网知识库问答") as demo:
    gr.Markdown("# 🏢 企业内网知识库智能问答机器人")
    gr.Markdown("基于混合检索（BM25 + 向量）与 Flashrank Rerank 重排序引擎构建。")

    chatbot = gr.Chatbot(type="messages", height=520)
    msg = gr.Textbox(placeholder="请输入您的问题（支持制度查询、技术方案、产品规格等）...", label="输入问题")
    
    with gr.Row():
        submit_btn = gr.Button("发送提问", variant="primary")
        clear_btn = gr.Button("清空对话")

    # 绑定提交事件与清空逻辑
    msg.submit(chat_fn, [msg, chatbot], [chatbot, msg])
    submit_btn.click(chat_fn, [msg, chatbot], [chatbot, msg])
    clear_btn.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，就能跑起来了。