1. 为什么本地知识库问答总在“胡说八道”
很多人第一次搭本地知识库,流程都差不多:把文档切片、调 GPT 向量接口生成 embedding、塞进向量数据库,然后拿用户问题去检索,最后把召回片段拼进 prompt 交给 ChatGPT 生成回答。听起来闭环了,但实际跑起来你会发现两个典型问题:一是问知识库里没有的内容,模型照样一本正经地编;二是明明知识库里有答案,检索却召回了一堆不相关的片段,回答自然跑偏。
这两个问题的根子不在 GPT3.5 本身,而在“检索质量”和“回答范围约束”这两件事没做扎实。向量数据库负责的是“找得到”,GPT 向量接口负责的是“把文本变成可比较的向量”,而 ChatGPT 负责的是“把召回内容组织成人话”。三者之间任何一环参数不对,回答就会越界。
我这次要做的,是一个能明确限制回答范围的本地知识库问答系统。核心目标有三个:文档切片后能稳定入库;检索召回率可控;当用户问知识库之外的问题时,模型必须回答“未查到相关信息”,而不是自由发挥。整套流程用 TaoToken 统一 Key 来管理 GPT 向量接口和 Chat 接口的调用,避免在代码里散落多个密钥。
适合谁看:已经会一点 Python、想自己搭一个内部文档问答机器人的开发者;或者你已经在用 ChatGPT,但希望把回答限制在自己资料范围内的人。下面从环境准备开始,一步步给出可复制的配置和代码。
2. TaoToken 统一 Key 前置配置与向量库选型
在动手写代码之前,先把“钥匙”和“仓库”定下来。钥匙指的是调用 GPT 向量接口和 Chat 接口的凭证,仓库指的是向量数据库。这两件事如果一开始没规划好,后面改起来很麻烦。
2.1 为什么用 TaoToken 统一管理 Key
原始方案里是在.env里放一个OPENAI_API_KEY,然后 embedding 和 chat 都用它。但实际项目中,你可能会同时用到向量接口、对话接口,甚至后面接入 Claude Code 或 Codex 做辅助开发。如果每个地方都单独配 Key,管理成本高,还容易在代码里硬编码泄露。
TaoToken 的做法是提供一个统一的 API 入口,Base URL 固定为https://taotoken.net/api,你用同一个 Key 就能调用不同的模型接口。这样在代码里只需要维护一份配置,切换模型时改 Model ID 就行,不用动 Key。
配置片段(.env文件,放在项目根目录):
# TaoToken 统一 Key 配置 TAOTOKEN_API_KEY=sk-your-taotoken-key-here TAOTOKEN_BASE_URL=https://taotoken.net/api # 向量模型与对话模型 EMBEDDING_MODEL=text-embedding-ada-002 CHAT_MODEL=gpt-3.5-turbo # 向量数据库 QDRANT_HOST=127.0.0.1 QDRANT_PORT=6333 COLLECTION_NAME=data_collection注意:TAOTOKEN_BASE_URL后面不要加/v1,具体路径由 SDK 或请求库拼接。如果你用的是 OpenAI 官方 Python SDK,可以通过openai.api_base指定。
2.2 向量数据库选 Qdrant 的理由
原始方案用的是 Qdrant,本地跑一个 Docker 容器就能用,支持余弦距离和 HNSW 索引,适合中小规模知识库。安装命令:
docker run -d --name qdrant \ -p 6333:6333 -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant跑起来后访问http://127.0.0.1:6333/dashboard能看到管理界面。向量维度要和 embedding 模型对齐,text-embedding-ada-002输出 1536 维,所以建 collection 时size=1536,距离用COSINE。
如果你不想用 Docker,也可以用内存模式QdrantClient(":memory:")做测试,但生产环境建议持久化。
2.3 文档切片策略
切片是影响召回率的关键。原始方案用#####分割,这在实际文档里很少见。更通用的做法是按段落或固定长度切,同时保留重叠部分。我建议:
- 按
\n\n切段落,如果单段超过 500 字,再按句号切; - 每个切片保留 50 字重叠,避免上下文断裂;
- 切片时带上标题或来源,存进 payload,方便后续过滤。
这部分逻辑写在入库脚本里,下面第三节会给完整代码。
3. 可复制的向量库建表与检索参数配置
这一节是核心,给出完整的入库脚本和检索脚本,参数都标清楚,你可以直接改路径和 Key 就能跑。
3.1 入库脚本:文档切片 + GPT 向量接口 + Qdrant 写入
先安装依赖:
pip install qdrant-client openai python-dotenv tqdm flask入库脚本ingest.py:
import os import tqdm import openai from dotenv import load_dotenv from qdrant_client import QdrantClient from qdrant_client.http.models import Distance, VectorParams, PointStruct load_dotenv() # 配置 TaoToken 统一入口 openai.api_key = os.getenv("TAOTOKEN_API_KEY") openai.api_base = os.getenv("TAOTOKEN_BASE_URL") EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-ada-002") COLLECTION_NAME = os.getenv("COLLECTION_NAME", "data_collection") SOURCE_DIR = "./source_data" CHUNK_SIZE = 500 CHUNK_OVERLAP = 50 def split_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP): """按固定长度切片,保留重叠""" chunks = [] start = 0 while start < len(text): end = start + size chunks.append(text[start:end]) start = end - overlap return chunks def to_embedding(text): """调用 GPT 向量接口生成 embedding""" resp = openai.Embedding.create( model=EMBEDDING_MODEL, input=text ) return resp["data"][0]["embedding"] def main(): client = QdrantClient( host=os.getenv("QDRANT_HOST", "127.0.0.1"), port=int(os.getenv("QDRANT_PORT", 6333)) ) # 重建 collection,维度 1536,余弦距离 client.recreate_collection( collection_name=COLLECTION_NAME, vectors_config=VectorParams(size=1536, distance=Distance.COSINE), ) count = 0 for root, dirs, files in os.walk(SOURCE_DIR): for file in tqdm.tqdm(files): file_path = os.path.join(root, file) with open(file_path, "r", encoding="utf-8") as f: text = f.read() chunks = split_text(text) for chunk in chunks: if not chunk.strip(): continue vector = to_embedding(chunk) client.upsert( collection_name=COLLECTION_NAME, wait=True, points=[ PointStruct( id=count, vector=vector, payload={"title": file, "text": chunk} ) ], ) count += 1 print(f"入库完成,共写入 {count} 个切片") if __name__ == "__main__": main()关键参数说明:
| 参数 | 作用 | 建议值 |
|---|---|---|
| CHUNK_SIZE | 单切片字符数 | 300–500 |
| CHUNK_OVERLAP | 切片重叠字符数 | 50 |
| size | 向量维度 | 1536(ada-002) |
| distance | 距离度量 | COSINE |
| wait | 写入是否同步 | True(调试期) |
3.2 检索脚本:向量召回 + 范围约束 prompt
检索脚本query.py:
import os import openai from dotenv import load_dotenv from qdrant_client import QdrantClient load_dotenv() openai.api_key = os.getenv("TAOTOKEN_API_KEY") openai.api_base = os.getenv("TAOTOKEN_BASE_URL") EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-ada-002") CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-3.5-turbo") COLLECTION_NAME = os.getenv("COLLECTION_NAME", "data_collection") TOP_K = 3 MAX_SUMMARY_LEN = 300 def build_prompt(question, answers): """构造带范围约束的 prompt""" system = "你是一个本地知识库问答助手,只能根据提供的段落回答问题。如果段落内容与问题无关,必须回答:未查到相关信息。" demo_q = '使用以下段落来回答问题:"成人头疼,流鼻涕是感冒还是过敏?"\n1. 普通感冒:喉咙痛、流鼻涕、轻度发热。\n2. 常年过敏:鼻塞、眼睛发痒、打喷嚏。' demo_a = "成人出现头痛和流鼻涕,可能是普通感冒或常年过敏。咽喉痛和咳嗽偏向感冒,眼睛肿胀偏向过敏。" q = f'使用以下段落来回答问题,如果段落内容不相关就返回未查到相关信息:"{question}"\n' for idx, ans in enumerate(answers): q += f"{idx+1}. {ans['title']}: {ans['text']}\n" return [ {"role": "system", "content": system}, {"role": "user", "content": demo_q}, {"role": "assistant", "content": demo_a}, {"role": "user", "content": q}, ] def query(question): client = QdrantClient( host=os.getenv("QDRANT_HOST", "127.0.0.1"), port=int(os.getenv("QDRANT_PORT", 6333)) ) # 问题向量化 emb = openai.Embedding.create( model=EMBEDDING_MODEL, input=question )["data"][0]["embedding"] # 向量检索 search_result = client.search( collection_name=COLLECTION_NAME, query_vector=emb, limit=TOP_K, search_params={"exact": False, "hnsw_ef": 128} ) answers = [] for r in search_result: text = r.payload["text"] summary = text[:MAX_SUMMARY_LEN] if len(text) > MAX_SUMMARY_LEN else text answers.append({"title": r.payload["title"], "text": summary}) # 调用 Chat 接口 completion = openai.ChatCompletion.create( model=CHAT_MODEL, temperature=0.2, messages=build_prompt(question, answers), ) return completion.choices[0].message.content if __name__ == "__main__": q = input("请输入问题:") print(query(q))这里有两个约束点:system明确要求“只能根据段落回答”,temperature=0.2降低随机性。检索参数hnsw_ef=128控制 HNSW 搜索广度,值越大召回越准但越慢,128 是平衡值。
3.3 用 Flask 包一层接口
from flask import Flask, request, jsonify from query import query app = Flask(__name__) @app.route("/search", methods=["POST"]) def search(): data = request.get_json() question = data.get("search", "") answer = query(question) return jsonify({"code": 200, "data": {"search": question, "answer": answer}}) if __name__ == "__main__": app.run(host="0.0.0.0", port=3000)启动后 POSThttp://127.0.0.1:3000/search,body 为{"search": "你的问题"}。
4. 验证请求:越界提问与命中率两组用例
代码跑通不代表效果达标,必须用用例验证“回答被限制在知识库内”。我准备了两组测试。
4.1 越界提问测试
假设知识库里只有医疗问诊相关文档,问一个完全无关的问题:
curl -X POST http://127.0.0.1:3000/search \ -H "Content-Type: application/json" \ -d '{"search": "今天股市大盘走势如何?"}'预期返回:
{ "code": 200, "data": { "search": "今天股市大盘走势如何?", "answer": "未查到相关信息" } }如果模型开始分析股市,说明 prompt 约束没生效。常见原因是system写得太弱,或者 demo 示例不够明确。可以把system改成更硬的措辞:“你没有任何外部知识,只能使用下方段落。段落无关时必须原样输出:未查到相关信息。”
4.2 命中率测试
准备 10 个知识库内的问题,人工标注正确答案所在文档,然后看检索 top3 是否包含该文档。我实测下来,切片 500 字、重叠 50 字、top_k=3 的情况下,命中率大概在 80% 左右。如果命中率低,优先调这三个参数:
- 切片太大:一个切片混了多个主题,向量被平均,召回不准;
- 重叠太小:跨切片的问题找不到完整上下文;
- top_k 太小:正确答案排在第四位之后。
可以写个简单脚本批量测试:
test_cases = [ ("成人头疼流鼻涕是感冒还是过敏", "感冒过敏.txt"), ("儿童发烧怎么处理", "儿童发热.txt"), ] for q, expected in test_cases: ans = query(q) print(f"问题:{q}\n回答:{ans}\n预期来源:{expected}\n---")如果回答里出现了知识库没有的细节,说明模型在“补全”,需要把temperature再调低,或者缩短MAX_SUMMARY_LEN,减少模型自由发挥的空间。
5. 本篇常见错排查:401、local proxy failed、reading choices
搭这套系统时,报错基本集中在几个地方。下面按真实报错给出排查路径。
5.1 401 Unauthorized
报错原文:
openai.error.AuthenticationError: Incorrect API key provided原因通常是 Key 没读到或 Base URL 配错。检查顺序:
.env文件是否在项目根目录,load_dotenv()是否在openai.api_key赋值之前调用;TAOTOKEN_API_KEY是否有多余空格或引号;openai.api_base是否设成了https://taotoken.net/api,不要多加/v1;- 如果用的是新版 SDK,
api_base可能叫base_url,按版本调整。
5.2 local proxy failed
报错原文:
APIConnectionError: Connection error. local proxy failed这是网络层问题,不是 Key 问题。先确认本机能否访问https://taotoken.net/api,可以用 curl 测:
curl -I https://taotoken.net/api如果返回 404 或 200 都说明网络通,404 是因为根路径没有具体接口。如果超时,检查本机 DNS 和防火墙。注意不要在代码里配置任何代理地址,直接用系统网络即可。
5.3 reading choices 报错
报错原文:
KeyError: 'choices'或者:
IndexError: list index out of range这通常发生在completion.choices[0]这一步。原因可能是:
- 请求被限流,返回体里没有
choices,先打印completion看完整结构; - 模型名写错,比如
gpt-3.5-turbo写成gpt3.5; - 消息格式不对,
messages里 role 不是 system/user/assistant。
排查时先把completion原样打印出来,看error字段。
5.4 OAuth 相关报错
如果你在接入 Claude Code 或 Codex 时看到 OAuth 报错,比如:
OAuth token expired这说明你用的是需要 OAuth 的客户端,而不是 API Key 模式。TaoToken 的 API Key 模式不需要 OAuth,直接在配置里填 Base URL、Key、Model ID 三件套即可。以 Cline MCP 配置为例:
{ "mcpServers": { "taotoken": { "url": "https://taotoken.net/api", "apiKey": "sk-your-taotoken-key", "model": "gpt-3.5-turbo" } } }Codex 的auth.json也是类似,填base_url、api_key、model三个字段。CC Switch 切换时确认这三项一致,就不会触发 OAuth 流程。
6. 把 Key 管好,把范围锁死
整套流程跑下来,最耗时间的不是写代码,而是调切片参数和 prompt 约束。我的经验是:切片宁小勿大,重叠宁多勿少;prompt 里的“未查到相关信息”要写成硬性指令,不要用“尽量”“建议”这类软词;temperature 控制在 0.2 以下。
如果你后面要长期做编码或 Agent 类任务,可以把这套知识库检索能力接到 Coding Plan 里,用同一个 TaoToken Key 管理向量接口和对话接口,省去多密钥切换的麻烦。需要看模型对话效果可以直接开模型对话页面测;要管理 Key 就去 API Keys 页面;接入文档在 doc 里能找到各语言的示例。
最后留一个实用技巧:入库完成后,先用 5 个明显无关的问题测一遍,确认全部返回“未查到相关信息”,再测知识库内问题。顺序反了的话,你会分不清是检索问题还是约束问题。