10. 综合练习：知识库问答助手
这是什么：知识库问答助手 = 一个把文件夹变成知识库、然后能对着它提问的小工具，答案还带出处。
为什么用：前面九章的零件现在该合体了：切分、嵌入、向量库、检索、生成各管一段，缺一块就答不上来。真正的项目永远不是单章作业。
如何用：本章分三步建库脚本 build_kb.py 和问答脚本 ask.py，跑起来对着自己的文档提问。
10.1 功能设计
输入：一个文档文件夹（.md / .txt）
输出：一个持久化的向量库 + 一个问答脚本
用起来：
python build_kb.py ./docs --rebuild # 建库（一次）
python ask.py "差旅住宿能报多少？" # 提问（无数次）10.2 项目结构
kb/
├── kb_common.py # 公共模块：客户端、嵌入、切分
├── build_kb.py # 建库：读文档 → 切块 → 向量化 → 入库
└── ask.py # 问答：检索 → 拼提示词 → 生成10.3 公共模块 kb_common.py
"""知识库公共模块：客户端、嵌入、切分"""
import os
from pathlib import Path
from openai import OpenAI
client = OpenAI(
api_key=os.environ["LLM_API_KEY"],
base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
)
EMBED_MODEL = os.environ.get("EMBED_MODEL", "text-embedding-3-small")
CHAT_MODEL = os.environ.get("CHAT_MODEL", "gpt-4o-mini")
CHROMA_PATH = os.environ.get("CHROMA_PATH", "./chroma_data")
COLLECTION = os.environ.get("COLLECTION", "kb")
TOP_K = int(os.environ.get("TOP_K", "3"))
def embed(texts, batch_size=64):
"""文字 → 向量。分批调用，避免一次请求太大被服务端拒绝"""
out = []
for i in range(0, len(texts), batch_size):
resp = client.embeddings.create(
model=EMBED_MODEL, input=texts[i:i + batch_size]
)
out.extend([d.embedding for d in sorted(resp.data, key=lambda d: d.index)])
return out
def split_text(text, chunk_size=400, overlap=60):
"""按语义边界递归切分（第 3 章的实现）"""
seps = ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
chunks = _split(text, chunk_size, seps)
chunks = _add_overlap(chunks, overlap)
return [c.strip() for c in chunks if c.strip()]
def _split(text, size, seps):
if len(text) <= size:
return [text]
if not seps:
return [text[i:i + size] for i in range(0, len(text), size)]
sep = seps[0]
parts = text.split(sep)
out, buf = [], ""
for part in parts:
piece = part if not buf else buf + sep + part
if len(piece) <= size:
buf = piece
continue
if buf:
out.append(buf)
buf = ""
if len(part) > size:
out.extend(_split(part, size, seps[1:]))
else:
buf = part
if buf:
out.append(buf)
return out
def _add_overlap(chunks, overlap):
if overlap <= 0 or len(chunks) <= 1:
return chunks
out = [chunks[0]]
for c in chunks[1:]:
out.append(out[-1][-overlap:] + c)
return out
def load_documents(folder):
"""读文件夹里的 .md / .txt，切块并附上元数据"""
blocks = []
for path in sorted(Path(folder).rglob("*")):
if path.suffix.lower() not in (".md", ".txt"):
continue
text = path.read_text(encoding="utf-8")
for i, chunk in enumerate(split_text(text)):
blocks.append({
"text": chunk,
"metadata": {
"source": path.name, # 来源文件名
"path": str(path),
"chunk_id": i,
},
})
return blocks
10.4 建库脚本 build_kb.py
"""建库：读文档 → 切块 → 向量化 → 写入 Chroma"""
import shutil
import sys
import chromadb
from kb_common import CHROMA_PATH, COLLECTION, embed, load_documents
def build(folder, rebuild=False):
if rebuild:
print("清空旧索引……")
shutil.rmtree(CHROMA_PATH, ignore_errors=True)
blocks = load_documents(folder)
if not blocks:
print(f"{folder} 里没有 .md / .txt 文档")
return
print(f"切出 {len(blocks)} 块，正在向量化（首次会慢一些）……")
vectors = embed([b["text"] for b in blocks])
client = chromadb.PersistentClient(path=CHROMA_PATH)
col = client.get_or_create_collection(
name=COLLECTION,
configuration={"hnsw": {"space": "cosine"}},
)
col.upsert(
ids=[f"{b['metadata']['source']}#{b['metadata']['chunk_id']}" for b in blocks],
documents=[b["text"] for b in blocks],
embeddings=vectors,
metadatas=[b["metadata"] for b in blocks],
)
print(f"✅ 完成：共 {col.count()} 块已入库 → {CHROMA_PATH}")
if __name__ == "__main__":
args = [a for a in sys.argv[1:] if not a.startswith("--")]
folder = args[0] if args else "./docs"
build(folder, rebuild="--rebuild" in sys.argv)
10.5 问答脚本 ask.py
"""问答：检索资料 → 塞进提示词 → 生成带出处的回答"""
import sys
import chromadb
from kb_common import CHAT_MODEL, CHROMA_PATH, COLLECTION, TOP_K, client, embed
def retrieve(question, top_k=TOP_K):
"""向量检索，返回带相似度的若干条"""
db = chromadb.PersistentClient(path=CHROMA_PATH)
col = db.get_or_create_collection(name=COLLECTION)
query_vec = embed([question])[0]
res = col.query(
query_embeddings=[query_vec],
n_results=top_k,
include=["documents", "metadatas", "distances"],
)
# cosine 空间下 distance = 1 - 余弦相似度
return [
{"text": doc, "metadata": meta, "score": 1 - dist}
for doc, meta, dist in zip(
res["documents"][0], res["metadatas"][0], res["distances"][0]
)
]
def answer(question):
hits = retrieve(question)
if not hits or hits[0]["score"] < 0.35: # 最高分太低 = 没找到相关资料
return "知识库里没有找到相关资料。可以换个说法，或把资料补进知识库再问。", hits
context = "\n\n".join(
f"[{i + 1}]（来源：{h['metadata']['source']}）\n{h['text']}"
for i, h in enumerate(hits)
)
prompt = (
"你是一个严谨的知识库助手。只依据下面的资料回答问题，"
"不要补充资料里没有的信息；资料不足时明确说明缺什么。\n"
"回答末尾用 [编号] 标出依据。\n\n"
f"资料：\n{context}\n\n"
f"问题：{question}"
)
resp = client.chat.completions.create(
model=CHAT_MODEL,
messages=[{"role": "user", "content": prompt}],
temperature=0,
)
return resp.choices[0].message.content, hits
if __name__ == "__main__":
q = " ".join(sys.argv[1:]) or input("问题：")
ans, hits = answer(q)
print(ans)
if hits:
print("\n--- 引用来源 ---")
for h in hits:
print(f" {h['score']:.3f} {h['metadata']['source']}")
10.6 运行与验收
# 1. 准备资料：把你的 .md / .txt 放进 docs/ 目录
# 2. 建库
python build_kb.py ./docs --rebuild
# 3. 提问
python ask.py "差旅住宿的报销上限是多少？"
你会看到：
差旅住宿的报销标准为每晚 500 元，需凭发票报销 [1]。
--- 引用来源 ---
0.812 handbook.md
0.674 handbook.md
0.401 meeting.md验收清单：
- 换个问法（换同义词）也能命中同一段
- 问一个资料里没有的问题，它会说「没找到」而不是瞎编
- 回答末尾的
[1]能在「引用来源」里对上号
10.7 常见翻车现场
| 报错 / 现象 | 原因 | 修法 |
|---|---|---|
KeyError: 'LLM_API_KEY' |
没设环境变量 | 先设好再运行 |
| 每次都答「没找到资料」 | 阈值太高，或建库时没切出块 | 打印 hits 看相似度；调低 0.35 或改 chunk_size |
DuplicateIDError |
同一批 ID add 了两次 | 用 upsert（本例已用） |
InvalidDimensionException |
换了嵌入模型但没重建库 | 加 --rebuild |
| 问答脚本报「集合不存在」 | 建库和问答的 CHROMA_PATH 不一致 |
两边用同一个环境变量 |
| 回答特别慢 | 首次要下模型 + 每次都调 API | 嵌入结果可加缓存 |
10.8 动手改
- 给你的知识库接上 PDF：把第 8 章的 MinerU 解析和第 9 章的 OCR 路线拼进
load_documents，让.pdf和.png也能入库 - 加增量更新：只对「修改过的文件」重新切块入库，其余不动
- 把第 6 章的 BM25 混合检索接进
retrieve()，用第 7 章的评测集对比 Hit Rate@3 的提升 - 加一个
--json参数，把检索结果和答案一起输出成 JSON，方便接进别的程序
10.9 小结
- 三件套：
kb_common.py（零件）+build_kb.py（造库）+ask.py（问库） - 建库流程：读文档 → 切块 → 批量向量化 → upsert 入库，换模型就
--rebuild - 问答流程：检索 → 相似度兜底 → 拼带编号的上下文 → 生成并标注出处
- 三个细节最值钱：批量调用嵌入、相似度太低就拒答、温度设 0
- 文档解析是第 8、9 章的活：MinerU 吃有文字层的 PDF，PaddleOCR 吃扫描件和图片，按扩展名与文字层自动路由
- 接下来想更进一步：接混合检索（第 6 章）、接评测（第 7 章）、换生产级向量库
RAG 完成！你已经能让大模型「开卷考试」了。下一站：智能体 Agent，让会查资料的 LLM 学会自己调工具 🐸