====== 第十三章：项目实战 - 企业知识库问答系统 ======

前面的章节已经讲清楚了模型、提示词、消息、检索、结构化输出和状态管理。从本章开始，我们把这些能力真正拼起来，做一个最具代表性的落地项目：**企业知识库问答系统**。

之所以把它作为第一个完整项目，是因为它几乎覆盖了 LangChain 落地最核心的能力：
  * 文档接入；
  * 文本切分；
  * Embedding 与向量索引；
  * 检索增强生成；
  * 引用与防幻觉；
  * 会话上下文；
  * 基础评估与上线治理。

本章不会只讲概念，而是按“项目目标 -> 目录结构 -> 数据处理 -> 检索链 -> 接口封装 -> 评估优化”的顺序，构建一个可以继续扩展的工程骨架。

===== 13.1 项目目标 =====

一个企业知识库问答系统，至少要满足以下要求：
  * 员工提问时，不必翻大量制度文件；
  * 回答必须基于企业内部资料，而不是模型自己猜；
  * 最好能显示依据来源；
  * 如果资料不足，要明确拒答；
  * 后续可以接 Web 页面、机器人或办公门户。

===== 13.2 第一版系统范围 =====

为了避免一开始就做成“大而全”的系统，建议第一版只做这些能力：
  * 支持导入本地 Markdown / TXT / PDF 文档；
  * 支持按主题切分和向量化；
  * 支持自然语言检索问答；
  * 支持引用来源展示；
  * 支持资料不足时拒答。

第一版先不要引入 Agent。因为对知识库问答系统来说，稳定检索和引用正确，往往比“会不会调用工具”更重要。

===== 13.3 推荐目录结构 =====

<code>
enterprise-rag/
├── app/
│   ├── main.py
│   ├── chains.py
│   ├── prompts.py
│   ├── retriever.py
│   ├── ingestion.py
│   └── settings.py
├── data/
│   ├── raw/
│   └── processed/
├── vectorstore/
├── tests/
│   ├── test_ingestion.py
│   └── test_rag_chain.py
└── requirements.txt
</code>

这样的结构有几个优点：
  * 数据处理与在线问答解耦；
  * Prompt 独立管理；
  * 检索逻辑可单独替换；
  * 后期更容易接 API 服务。

===== 13.4 数据接入与预处理 =====

==== 13.4.1 一个简单的文档导入脚本 ====

<code python>
from pathlib import Path
from langchain_community.document_loaders import DirectoryLoader, TextLoader


def load_kb_documents(data_dir: str):
    loader = DirectoryLoader(
        data_dir,
        glob="**/*.md",
        loader_cls=TextLoader,
        show_progress=True,
    )
    return loader.load()


docs = load_kb_documents("./data/raw")
print(f"Loaded {len(docs)} documents")
</code>

==== 13.4.2 预处理要做什么 ====

导入之后，不要立刻切分。建议先处理：
  * 去除页眉页脚；
  * 去除重复空行；
  * 给每份文档补充 metadata；
  * 补充文档类别、版本号、生效日期。

<code python>
def normalize_document(doc):
    text = doc.page_content
    text = text.replace("

", "
")
    text = "
".join(line.strip() for line in text.splitlines() if line.strip())
    doc.page_content = text
    doc.metadata.setdefault("doc_type", "policy")
    doc.metadata.setdefault("source", "unknown")
    return doc

normalized_docs = [normalize_document(doc) for doc in docs]
</code>

===== 13.5 文本切分与索引 =====

==== 13.5.1 切分器配置 ====

<code python>
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=600,
    chunk_overlap=100,
    separators=["

", "
", "。", "，", " ", ""]
)

chunks = splitter.split_documents(normalized_docs)
print(f"Total chunks: {len(chunks)}")
</code>

==== 13.5.2 写入向量库 ====

<code python>
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./vectorstore/chroma"
)
print("Vector store created")
</code>

===== 13.6 构建检索器 =====

<code python>
def build_retriever(vectorstore):
    return vectorstore.as_retriever(
        search_kwargs={"k": 4}
    )

retriever = build_retriever(vectorstore)
</code>

如果你的文档 metadata 较完整，还可以加过滤：

<code python>
finance_retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 4,
        "filter": {"department": "finance"}
    }
)
</code>

===== 13.7 设计问答 Prompt =====

对于企业知识库问答，Prompt 不应只追求“回答自然”，而要优先强调：
  * 只依据资料回答；
  * 资料不足时拒答；
  * 输出引用；
  * 不要把通用常识当企业制度。

<code python>
from langchain_core.prompts import ChatPromptTemplate

rag_prompt = ChatPromptTemplate.from_template("""
你是企业知识库助手。
请严格依据提供的资料回答问题，不得虚构制度、日期或流程。
如果资料不足，请明确回答“未找到足够依据”。

请按以下格式回答：
1. 直接答案
2. 依据说明
3. 引用来源

问题：{question}

资料：
{context}
""")
</code>

===== 13.8 用 LCEL 构建问答链 =====

<code python>
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser


def format_docs(docs):
    return "

".join(
        f"[来源: {doc.metadata.get('source', 'unknown')}]
{doc.page_content}"
        for doc in docs
    )

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | rag_prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("差旅报销一般多久到账？")
print(answer)
</code>

===== 13.9 增加引用返回 =====

实际业务里，常见需求是不仅返回答案，还要把命中的文档片段一起带回前端。

<code python>
def ask_with_sources(question: str):
    docs = retriever.invoke(question)
    context = format_docs(docs)
    answer = (rag_prompt | model | StrOutputParser()).invoke({
        "question": question,
        "context": context,
    })
    return {
        "answer": answer,
        "sources": [
            {
                "source": doc.metadata.get("source", "unknown"),
                "preview": doc.page_content[:180],
            }
            for doc in docs
        ]
    }

result = ask_with_sources("发票抬头写错了还能报销吗？")
print(result["answer"])
print(result["sources"])
</code>

这样前端就能展示：
  * 主答案；
  * 命中文档来源；
  * 引用内容摘要。

===== 13.10 加入会话历史 =====

企业问答常常不是单轮的。比如：
  * 用户先问报销多久到账；
  * 再问如果发票抬头错误怎么办；
  * 再追问金额超标如何处理。

这时就要引入消息历史。

<code python>
from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

chat_rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是企业知识库助手，仅依据提供资料回答。"),
    MessagesPlaceholder("history"),
    ("human", "问题：{question}

资料：{context}")
])

history = [
    HumanMessage(content="差旅报销多久到账？"),
    AIMessage(content="通常在审批通过后的 3 个工作日内到账。")
]
</code>

===== 13.11 封装为函数或 API =====

一个更像样的项目，不应把链直接散落在脚本里，而应封装成服务接口。

<code python>
class EnterpriseRAGService:
    def __init__(self, retriever, model, prompt):
        self.retriever = retriever
        self.model = model
        self.prompt = prompt

    def ask(self, question: str):
        docs = self.retriever.invoke(question)
        context = format_docs(docs)
        answer = (self.prompt | self.model | StrOutputParser()).invoke({
            "question": question,
            "context": context,
        })
        return {
            "answer": answer,
            "documents": docs,
        }
</code>

之后你可以在 FastAPI、Flask 或 Streamlit 中调用这个服务。

===== 13.12 质量提升方向 =====

第一版上线后，常见优化路线包括：
  * 按文档类型做不同切分策略；
  * 引入多查询检索；
  * 引入重排；
  * 对旧文档做版本治理；
  * 增加“无答案”检测；
  * 构建评测集。

===== 13.13 一个简单的评测脚本 ====

<code python>
test_cases = [
    {
        "question": "差旅报销多久到账？",
        "must_include": ["3 个工作日"],
    },
    {
        "question": "试用期员工可以申请差旅报销吗？",
        "must_include": ["未找到足够依据"],
    },
]

for case in test_cases:
    answer = rag_chain.invoke(case["question"])
    passed = all(keyword in answer for keyword in case["must_include"])
    print(case["question"], passed)
</code>

这虽然只是最粗糙的评测，但已经比“凭感觉好像还行”强很多。

===== 13.14 本章小结 =====

企业知识库问答系统最重要的不是“说得像不像人”，而是：
  * 资料对不对；
  * 引用稳不稳；
  * 没依据时能不能拒答；
  * 能否持续维护和迭代。

===== 练习 =====

1. 为你的业务场景列出第一批最适合入库的 20 份文档。

2. 实现一个从本地目录导入、切分并写入 Chroma 的脚本。

3. 用 LCEL 构建一个最小 RAG 链，并返回引用来源。

4. 增加一个“资料不足时拒答”的测试样例。

5. 设计一个 API 输出格式，包含答案、来源、trace_id 和耗时。


===== 13.15 补充案例：用 FastAPI 暴露问答接口 =====

<code python>
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
rag_service = EnterpriseRAGService(retriever=retriever, model=model, prompt=rag_prompt)

class AskRequest(BaseModel):
    question: str

@app.post("/ask")
def ask(req: AskRequest):
    result = rag_service.ask(req.question)
    return {
        "answer": result["answer"],
        "sources": [doc.metadata for doc in result["documents"]],
    }
</code>

===== 13.16 补充案例：知识库更新脚本 =====

在实际项目中，知识库不会只建一次。你需要定期更新向量索引：

<code python>
def rebuild_index(data_dir: str, persist_dir: str):
    docs = load_kb_documents(data_dir)
    docs = [normalize_document(doc) for doc in docs]
    chunks = splitter.split_documents(docs)
    Chroma.from_documents(
        documents=chunks,
        embedding=OpenAIEmbeddings(model="text-embedding-3-large"),
        persist_directory=persist_dir,
    )
    print("Index rebuilt")
</code>

这个脚本后续可以接到：
  * 定时任务；
  * 管理后台按钮；
  * 文档上传流程。