项目背景
某企业内部知识库沉淀了上千份产品手册、FAQ 和历史工单，员工查问题主要靠关键词搜索 + 翻文档，平均定位一个答案要 8 分钟。早期尝试直接接通用大模型，但模型对企业私有术语一问三不知，还会出现「一本正经地编造」的幻觉，根本不敢直接用。
本次目标是搭一套基于 RAG（检索增强生成）的领域问答系统：用企业自有语料做向量检索，把检索到的片段塞进 Prompt 让模型「看着答」，把通用模型 + 私有知识结合起来。技术栈选 Python + LangChain + Milvus + 国内某开源大模型，要求回答可溯源、首字延迟低于 2s、支持流式输出。
RAG 不是「调个 API」那么简单，真正的工程量在语料处理、检索质量与 Prompt 调优这三处，模型反而不是瓶颈。
架构设计
系统分「离线索引」与「在线问答」两条链路。离线把语料清洗、切片、向量化后写入 Milvus；在线接收用户问题，先检索 Top-K 相关片段，再用 Prompt 组装后调大模型流式生成，同时返回引用来源。
离线索引：文档 → 清洗 → 切片(Chunk) → Embedding → Milvus
在线问答：问题 → 改写 → 向量检索 TopK → 重排序 → Prompt 组装 → LLM 流式生成 → 引用溯源关键设计：切片按语义段落 + 重叠窗口，避免把一句话拦腰切断；检索后用交叉编码器做一次重排序（rerank），把真正相关的片段往前顶；Prompt 里强约束「只能依据给定资料回答，资料中没有就回答不知道」，从源头压制幻觉。
核心实现
1）语料切片：用 LangChain 的 RecursiveCharacterTextSplitter，按段落 → 句号 → 字符层层递进切，并保留 80 字符重叠，保证上下文不断裂。
# index/chunker.py —— 语义切片 + 重叠窗口
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=80,
separators=["\n\n", "\n", "。", "；", "，", " ", ""],
)
def chunk_docs(docs: list[str]) -> list[str]:
chunks = []
for d in docs:
chunks.extend(splitter.split_text(d))
return chunks2）向量检索：用 bge-large-zh 做 Embedding 写入 Milvus，查询时取 Top-8，再用 bge-reranker 重排序保留 Top-3，显著提升相关性。
# retrieve/search.py —— 向量检索 + 重排序
from pymilvus import Collection
def search(question: str, top_k: int = 8) -> list[dict]:
q_vec = embed(question) # query 向量化
col = Collection("kb_chunks")
col.load()
hits = col.search(
data=[q_vec], anns_field="vector",
param={"metric_type": "IP", "params": {"nprobe": 16}},
limit=top_k, output_fields=["text", "doc_id"],
)[0]
passages = [{"text": h.entity.get_value("text"),
"doc_id": h.entity.get_value("doc_id"),
"score": h.score} for h in hits]
return rerank(question, passages, top_n=3) # 交叉编码器重排3）Prompt 组装与流式生成：把检索片段编号塞进 system prompt，要求模型引用编号；用 SSE 把生成 token 流式推给前端，首字延迟控制在 2s 内。
# answer/chain.py —— Prompt 组装 + 流式生成
SYS_PROMPT = """你是企业知识助手，只能依据下方资料回答。
资料中没有的内容，必须回答"资料中未提及"。
回答末尾用 [1][2] 形式标注引用的资料编号。
资料：
{context}
"""
def answer_stream(question: str):
passages = search(question)
context = "\n\n".join(
f"[{i+1}] doc={p['doc_id']}\n{p['text']}" for i, p in enumerate(passages)
)
messages = [
{"role": "system", "content": SYS_PROMPT.format(context=context)},
{"role": "user", "content": question},
]
for chunk in llm.stream(messages): # SSE 逐 token 推送
yield chunk
yield {"sources": [{"doc_id": p["doc_id"]} for p in passages]}- 检索召回率（Top3 命中）：63% → 88%（加 rerank 后）
- 幻觉率（人工抽检）：22% → 4%
- 首字延迟：6.4s → 1.7s（流式后）
- 平均定位答案时间：8 分钟 → 12 秒
技术难点
难点一：切片粒度。切太粗，一个 chunk 塞太多无关内容稀释相关性；切太细，上下文断裂答不全。最终通过对比 256/512/1024 三种 chunk_size 在评测集上的召回率，确定 512 + 80 重叠为最佳。评测集是人工标注的 200 个真实问题，这是质量基石。
难点二：幻觉压制。即便给了资料，模型偶尔仍会「脑补」。我们在 Prompt 里加了「资料中未提及」硬约束 + 输出引用编号，并在生成后做一次「引用校验」：检查回答里的编号是否都在检索片段范围内，不在则判定为高风险并降级为「未找到相关资料」。
# answer/verify.py —— 引用校验，压制幻觉
import re
def verify(answer: str, passages: list[dict]) -> bool:
refs = set(int(x) for x in re.findall(r"\[(\d+)\]", answer))
valid = set(range(1, len(passages) + 1))
# 引用越界或全是资料外的编号 → 判定幻觉
if refs and not refs.issubset(valid):
return False
return True难点三：流式输出的中断与异常处理。SSE 连接中途断开时，已生成的内容要能落库做「未完成会话续写」。我们在每个 token 推送后异步写 Redis 缓存，连接断开时标记会话状态，用户重连可续上。
踩坑复盘
坑 1：Embedding 模型与 LLM 不匹配导致检索差。最初用某闭源 Embedding，但中文领域召回率只有 63%。换成 bge-large-zh 后召回到 81%。教训：中文场景 Embedding 模型选型比 LLM 更关键，必须用真实语料跑评测集，不能只看榜单。
坑 2：Milvus 索引参数没调。默认 nprobe=10 在百万级向量下召回率掉得明显，调到 16 后召回率 +5%，延迟仅增加 8ms。nprobe 是召回与延迟的核心旋钮，要按数据量校准。
坑 3：把检索片段一股脑塞 Prompt 导致超 token 上限。早期没做长度控制，长问题检索回 8 个片段直接超模型上下文窗口被截断。加了「片段总长度不超过 3000 token、超出按 rerank 分数裁剪」的策略后稳定。教训：Prompt 长度必须有预算控制。
RAG 的胜负手不在模型大小，而在「评测集 + 切片粒度 + rerank + Prompt 约束」这四件地基是否夯实。
项目总结
系统上线后覆盖 9 个内部知识域，员工问答满意度从原搜索方案的 54% 提到 86%，平均定位答案时间从 8 分钟降到 12 秒。整个项目最大的认知更新是：RAG 不是模型问题而是工程问题，一个扎实的评测集 + rerank + Prompt 约束，比换更大的模型收益更直接。
- 检索命中率：63% → 88%
- 幻觉率：22% → 4%
- 首字延迟：6.4s → 1.7s
- 用户满意度：54% → 86%