ChatGPT知识库构建实战：从数据预处理到智能问答系统搭建

背景痛点：传统检索为什么总答非所问？

去年我给公司做内部 FAQ 机器人，最早用的是 ElasticSearch 的 BM25 打分。上线一周就被吐槽“鸡同鸭讲”——明明问的是“年假几天”，却返回“年假申请流程”。根本原因是：

1. 关键词匹配无法感知语义，同义词/语序变化直接翻车
2. 知识库一旦超过 5 万条，召回 Top10 里 7 条不相关，准确率跌到 45% 以下
3. GPT-3.5 虽然懂语义，但 4k token 上限让“把整库塞进去”成了天方夜谭，成本也扛不住

于是目标很明确：让 LLM 只读“可能相关的几段”，而不是“整本书”。

技术选型：向量库 vs 直调 API 的性价比

我先后试了三种路线，结论先给：

最终线上采用“FAISS + 自建”方案，把延迟压到 100 ms 以内，成本降 60%。下文代码均以该方案为例，方便你一键迁移到 Pinecone。

核心实现：30 行代码搞定多格式解析

1. 数据层：LangChain 一把梭

LangChain 的 DocumentLoader 对常用格式都做了封装，我封装了一个统一入口：

# loader.py from pathlib import Path from langchain.document_loaders import PyPDFLoader, UnstructuredHTMLLoader from typing import List from langchain.schema import Document def load_folder(path: str) -> List[Document]: docs = [] for p in Path(path).rglob("*"): if p.suffix == ".pdf": docs.extend(PyPDFLoader(str(p)).load()) elif p.suffix == ".html": docs.extend(UnstructuredHTMLLoader(str(p)).load()) return docs

2. 切分块 + 批量 Embedding

chunk_size 不是拍脑袋，后面会实测。先写个带缓存的生成器：

# embed.py import hashlib, json, os, openai, tiktoken from typing import List from diskcache import Cache cache = Cache("embed_cache") ENC = tiktoken.encoding_for_model("text-embedding-ada-002") def get_embedding(texts: List[str]) -> List[List[float]]: key = hashlib.md5("".join(texts).encode()).hexdigest() if key in cache: return cache[key] # 每次 100 条批量，防止长度超限 embs = [] for i in range(0, len(texts), 100): resp = openai.Embedding.create( input=texts[i : i+100], model="text-embedding-ada-002" ) embs += [r["embedding"] for r in resp["data"]] cache[key] = embs return embs

3. 向量索引落盘

# build_index.py import faiss, numpy as np from loader import load_folder from embed import get_embedding from langchain.text_splitter import RecursiveCharacterTextSplitter docs = load_folder("./data") splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) texts = splitter.split_documents(docs) vectors = get_embedding([t.page_content for t in texts]) d = len(vectors[0]) index = faiss.IndexFlatIP(d) # 内积归一化后 = cosine index.add(np.array(vectors).astype(np.float32)) faiss.write_index(index, "faq.index")

4. Flask 后端：带 JWT 的 RESTful

# app.py from flask import Flask, request, jsonify from flask_jwt_extended import JWTManager, jwt_required, create_access_token import faiss, numpy, openai, os app = Flask(__name__) app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET") jwt = JWTManager(app) index = faiss.read_index("faq.index") texts = json.load(open("texts.json")) # 同步落盘 def search(query: str, k: int = 5): qvec = get_embedding([query])[0] D, I = index.search(numpy.array([qvec]), k) return [texts[i] for i in I[0]] @app.route("/login", methods=["POST"]) def login(): username = request.json.get("username") password = request.json.get("password") # 仅示例，请用真实校验 if username == "admin" and password == "pwd": return jsonify(access_token=create_access_token(identity=username)) return jsonify({"msg": "Bad creds"}), 401 @app.route("/ask", methods=["POST"]) @jwt_required() def ask(): question = request.json.get("q") chunks = search(question) context = "\n".join(chunks) prompt = f"Use the following context to answer concisely.\nContext:\n{context}\n\nQ: {question}\nA:" ans = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], max_tokens=300, temperature=0.1 ) return jsonify(answer=ans["choices"][0]["message"]["content"])

跑起来：

export OPENAI_API_KEY=sk-xx export JWT_SECRET=foo python app.py

性能优化：Chunk Size 与限流实战

1. Chunk Size 对召回率的影响

我用 200 条人工标注 FAQ 做 MRR@5 测试：

结论：500+50 是中文场景甜点值，英文可再大一点。

2. 应对 GPT-3.5 20 次/秒限流

• 后端加asyncio.Semaphore(15)做并发限速
• 对相同问题缓存 10 分钟，Key 用问题向量 128bit 量化哈希
• 压测显示缓存命中率 62%，QPS 从 8 → 24，翻三倍

避坑指南：特殊字符与权限

1. 特殊字符：PDF 常见\x0c换页符会成“不可见 token”，导致同一段落 embedding 偏差 > 0.05。统一用text = re.sub(r"\s+", " ", text)先清洗
2. 权限：知识库常含工资、人事敏感信息。
   • 向量文件放内网 MinIO，只对内网 Flask 开放
   • /ask接口按部门做行级过滤，把部门编码写进 JWT payload，检索时先过滤标签再召回
   • 日志只保存问题哈希，不保存原文，防泄密

代码规范小结

• 统一 Black 8 空格线宽，Black+isort 做 pre-commit
• 公开函数必写 Google Style docstring，并附类型标注
• 复杂业务函数拆成search()/build_prompt()/call_llm()三步，单测好写

延伸思考：LlamaIndex 混合检索

如果知识库再膨胀到千万级，纯向量召回也会“跑偏”。可以试 LlamaIndex 的BM25+Embedding混合检索：

from llama_index.retrievers import HybridRetriever retriever = HybridRetriever( vector_index=index, keyword_index=keyword_index, alpha=0.6 # 向量权重 )

实测在 100 w 条 Wiki 数据下，Top5 准确率再提 6%，延迟只加 15 ms，值得一试。

写在最后：把实验搬到“豆包”上

整套流程跑通后，我把同样思路迁移到火山引擎的豆包语音模型，发现官方已经封装好 ASR→LLM→TTS 的实时通话闭环，半小时就能在网页里跟“数字同事”聊天气。如果你想快速体验，又不打算自己踩向量库的坑，可以顺手试试这个动手实验：

从0打造个人豆包实时通话AI

我跟着文档跑了一遍，从注册到第一次语音通话大概 20 分钟，UI 也开源，改两行 JS 就能换上自己的知识库。对中级 Pythoner 来说，算是一次“语音交互”低成本入门。祝你玩得开心，早日让 AI 开口说话！