简介:一个基于智谱清言大模型API与RAG检索增强生成技术的智能问答系统,专门服务计算机专业考研408统考科目复习。系统将数据结构、计算机网络、操作系统、计算机组成原理等科目资料构建为向量数据库,支持语义检索与精准知识问答,可针对难点即时返回考点与解析,帮助查漏补缺,还能根据学习历史推荐复习内容。压缩包共28个文件、约25.97MB,含6个Python脚本、2个SQLite数据库、8个二进制数据文件、2个PDF与2个TXT文档,以及README、配置和说明文档,覆盖数据库构建、文本向量化、大模型调用、Streamlit交互界面等完整工程链路。附赠历年真题、考点分析、术语解释、示例代码等结构化资料,目录清晰、便于复现部署。目前已有67人学习,适合考研学生系统备考,也适合开发者参考RAG工程落地。
1. 基于智谱清言大模型API与RAG的408智能问答:资料越多越答不准?根因不在模型,在检索
考研408一共四门课,考生手头的资料通常有几十MB甚至更多,但真到做题卡壳时,面对几十个PDF反而翻不到想要的结论。另一个痛点是通用大模型聊起来头头是道,涉及具体教材的表述、早年真题的答案,它经常一本正经地编。这个标题指向的方案,就是用智谱清言大模型API接RAG检索增强生成:先把408教材、真题、笔记切片并向量化,建成一个可语义检索的向量数据库;每次提问先召回最相关的几个知识片段,再让大模型基于这些片段作答。真正解决的,是大模型没读过你资料时答不准、以及资料太多时人翻不到这两个问题。适合正在备考408的学生,也适合想用最小成本搭垂直问答服务的开发者。下面按可复现路径拆解。
2. 构建408向量数据库:清洗、切分与向量化的落地路径
2.1 为什么直接问大模型不够:RAG先解决“看过资料再作答”
大模型的知识止步于训练数据截止时间,408又是个每年考纲和命题风格都可能微调的科目。直接问模型“某年408真题中某题选什么”,它大概率会给出一个上下文合理、但基于错误记忆的答案。RAG的思路是把外部知识库提前准备好,每次提问都先从库里检索相关片段,再把片段作为上下文丢给大模型,让它“看着资料说人话”。
这个方案里,最容易被低估的是知识库构建环节。很多人以为RAG就是装个向量库、调个API就跑通了,结果做出来一问三不知,才回头发现:原始资料没清洗、切分粒度不对、embedding模型选得随意。408资料的特点是结构强、术语密、伪代码多、表格多,这些问题会逐个引爆。
这里还要区分一下常被拿来比较的KG知识库和RAG知识库。知识图谱(KG)适合关系固定的领域,比如“进程状态之间有哪些转换边”,需要人工维护schema和实体关系,408考点虽然关系清晰,但覆盖四门课上千个知识点时,KG构建成本会高到一个人维护不过来。RAG知识库则不需要显式建模,把段落丢向量库,靠语义检索命中上下文,属于“先存起来,查询时再隐式关联”。所以个人做408问答系统,RAG是更务实的起点。
2.2 语料清洗:把PDF、Word变成可切分的干净文本
我一般先建一个raw/目录,按科目归档:数据结构、计算机组成原理、操作系统、计算机网络。资料主要来自教材扫描版、教辅PDF、自己整理的一轮笔记。扫描版PDF直接用pdfplumber提取会得到一堆乱码或空文本,这种情况要么先用OCR工具转成文本层,要么放弃该来源换排版清晰的电子版。
清洗这一步很多人跳过,但向量模型对噪音很敏感。页眉页脚里的书名、页码、水印、章节标题重复出现在正文中,会导致检索时把来源信息当成正文内容;全角半角符号混用、空格和换行混乱,也会让切分器把本应连续的句子拦腰截断。
import pdfplumber import re def extract_pdf(path: str) -> str: text = "" with pdfplumber.open(path) as pdf: for page in pdf.pages: page_text = page.extract_text() or "" text += page_text + "\n" # 清理页眉页脚:常见规律是每页顶部两行和底部两行是书名/页码 lines = [line.strip() for line in text.splitlines() if len(line.strip()) > 2] text = "\n".join(lines) # 合并多余空白,保留段落结构 text = re.sub(r"[ \t]+", " ", text) # 去掉独立成行的页码 text = re.sub(r"\n\d{1,3}\n", "\n", text) # 修复行尾连字符分词,例如 "net-work" -> "network" text = re.sub(r"(\w+)-\n(\w+)", r"\1\2", text) return text逻辑说明:pdfplumber 按页面提取文本,页眉页脚因为长度和正文差异被过滤;正则修的是两种最高频噪音——孤立的页码和PDF换行导致的断词。参数说明:len(line.strip()) > 2是我在408资料上试出来的经验值,正文行几乎没有短于3个字符的,标题、页码、页眉基本会被滤掉;如果你的资料里有大量单字成行的伪代码,可以把这个阈值降为1。
清洗后的文本我会手动抽查几页,重点看表格是否被提取成行列错乱的文本。PDF里的表格是408资料重灾区:算法复杂度对比表、操作系统调度算法表、TCP状态转换表,一旦表格单元格顺序被打乱,后面切分和检索都会一塌糊涂。遇到表格,我通常会单独用表格提取工具另存为Markdown,或者接受原始排版并把每个表格作为不可切分的整体处理。
2.3 文本切分:chunk_size、overlap与结构化优先级
切分是RAG里最玄学的一步,也是翻车率最高的地方。固定按500字切,最大的问题是把知识点拦腰截断。比如“B+树索引的插入过程”可能从第470字开始,到第510字结束,前半段在旧chunk,后半段在新chunk,检索时谁也答不全。
更合理的策略是先按文档结构切,再在段落内做精细切分。408电子书大多有清晰的章节层级,我会先用正则把##、###标题找出来作为切片边界,保证每个chunk对应一个相对完整的知识子主题。子主题仍太长时,再用字符切分器兜底。
from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, # seperators 从优先到兜底,中文里“。”和“;”可以保住句子完整 separators=["\n## ", "\n### ", "\n\n", "\n", "。", ";", ""], keep_separator=True, ) chunks = splitter.split_text(clean_text)逻辑说明:separators决定切分器优先用哪个分隔符切。这里把一级/二级标题放在最前面,让标题和下面内容尽量留在同一chunk;其次按空行切段落,再按句号/分号切句子,最后才逐字符强行切。keep_separator=True会把分隔符保留在上一块末尾,避免标题丢失。
参数说明:chunk_size=500按字符数计算,中文500字大约是250到300个token,适配多数embedding模型的输入上限。chunk_overlap=80是相邻chunk的重叠长度,作用是让跨边界的上下文被两边同时覆盖。不要小看这个80,没有重叠时,一个知识点正好卡在边界上的概率会高得多。
另外要专门处理伪代码块。408的四门课都有大量伪代码、代码片段和算法步骤,比如快速排序的递归实现、银行家算法的安全检查流程。它们用空行分隔往往不充分,我会在预处理时检测“def ”、“while (”、“for ”等特征,给这些代码块前后加上```围栏,防止切分器把半行代码切到两个chunk里。
2.4 向量化:用智谱清言API做Embedding的取舍
清洗和切分完成后,要把每个chunk变成向量。这里有两种选择:自己部署开源的embedding模型(如bge系列),或者调用智谱清言开放平台提供的embedding接口。408项目规模不大,通常只有几千到几万个chunk,调用API的计算成本极低,而且不需要自己维护模型和GPU,所以我一般会选后者。
调用API时需要注意一个实际问题:文本内容会发送给第三方服务。408考研资料基本都是公开的教材和真题整理,没有隐私问题,可以放心用。但如果你未来要做企业内部知识库,得先确认数据合规要求。
import os from openai import OpenAI client = OpenAI( api_key=os.getenv("ZHIPU_API_KEY"), base_url="https://open.bigmodel.cn/api/paas/v4/" ) def embed_texts(texts: list[str]) -> list[list[float]]: # 单次请求批量传入,减少API调用次数 resp = client.embeddings.create( model="embedding-2", # 以智谱开放平台当前支持的embedding模型名为准 input=texts ) return [d.embedding for d in resp.data]逻辑说明:智谱清言API兼容OpenAI SDK格式,只是base_url指向自己的端点,方便复用现有代码。批量传入文本可以省掉多次网络往返,但单次batch不要超过平台的输入长度限制。
参数说明:model="embedding-2"是我在本地跑通时的模型名,你使用时以智谱官方控制台能看到的当前模型名为准。embedding模型不同,向量维度和语义表达能力也不同,常见的输出维度是1024或2048,这个维度会在后面建向量库时自动对齐,不需要手动处理。
向量化之后,我建议把chunk文本、向量、来源元数据一并保存成一个中间文件,比如embedding_cache.jsonl。这样做的好处是,后续调整检索逻辑时可以跳过embedding重复计算,重建向量库只要重新读一遍文件。408资料更新后,也能只跑增量部分,而不是从头再调一遍API。
3. 向量数据库选型与语义检索:从Chroma到混合召回
3.1 向量库怎么选:Chroma、FAISS、Milvus的边界
知识库建好后,下一步是把向量和原文存进能支撑语义检索的存储。市面上的向量数据库很多,但408问答系统用到的量级通常是几万到几十万chunk,这个规模下Chroma是最省心的选择。
| 方案 | 部署复杂度 | 适合规模 | 元数据过滤 | 适合场景 |
|---|---|---|---|---|
| Chroma | 嵌入式,纯本地文件 | 10万chunk以内 | 支持基本等值/范围过滤 | 个人项目、轻量Demo |
| FAISS | 库形态,需要自己管理索引 | 百万级 | 需配合外部倒排 | 单机离线检索 |
| Milvus | 独立服务,需要运维 | 千万级 | 功能完整 | 大规模生产环境 |
Chroma是嵌入式向量库,一个PersistentClient就解决存储和检索,不需要启动服务端。FAISS是元数据库,只管理向量检索,原文和元数据还要你自己存,对408这种带大量科目、章节信息的项目并不方便。Milvus功能强大,但对一个人做学习辅助系统来说运维成本过高,不划算。
3.2 元数据设计:科目、章节、来源决定你能过滤多细
很多人建向量库时只存文本和向量,忽略元数据,这是后面检索质量上不去的根源。408四门课的考点粒度差异很大,如果把整本操作系统教材混在一个collection里,检索“进程调度算法”时,很可能召回文件系统章节里包含“调度”字样的chunk。元数据可以让你在检索前先把范围关进某个科目或章节里。
import chromadb from chromadb.config import Settings client = chromadb.PersistentClient( path="./408_kb", settings=Settings(anonymized_telemetry=False) ) collection = client.get_or_create_collection( name="kaoyan_408", metadata={"hnsw:space": "cosine"} ) ids = [f"data_struct_chunk_{i}" for i in range(len(chunks))] metadatas = [ { "subject": "data_struct", "chapter": "查找", "section": "B+树", "source": "王道_数据结构", "year": 2024, } for _ in chunks ] collection.add( embeddings=embeds, documents=chunks, metadatas=metadatas, ids=ids, )逻辑说明:ids用 “科目 + 序号” 而不是纯数字,后续删除和更新时能按前缀定位到某个文档。metadatas里的subject、chapter、section通过where条件在查询时过滤,source和year则用于答案中标注出处。
参数说明:metadata={"hnsw:space": "cosine"}指定向量检索的距离空间。408的embedding向量没有特别强的模长语义,用余弦相似度比欧氏距离更稳定。如果你用的embedding模型官方说已经归一化,那么内积和余弦等价,这里仍然建议显式写cosine,避免不同代码库之间的默认值不一致。
3.3 语义检索参数:Top-K、距离算法与相似度阈值
检索这一步是整个系统中用户感知最明显的部分。Top-K设太大,无关片段会混进来,大模型会被干扰;设太小,正确答案落在第10个chunk里却没被召回。对408这种知识密度高的场景,我一般先取8到10个候选,再做后续过滤。
def semantic_search(query: str, top_k: int = 8, subject: str | None = None): query_embedding = embed_texts([query])[0] filter_dict = None if subject: filter_dict = {"subject": subject} results = collection.query( query_embeddings=[query_embedding], n_results=top_k, where=filter_dict, include=["documents", "metadatas", "distances"], ) scored_chunks = [] for doc, meta, dist in zip( results["documents"][0], results["metadatas"][0], results["distances"][0], ): scored_chunks.append({ "doc": doc, "metadata": meta, "score": 1 - dist, # 余弦距离转相似度 }) return scored_chunks逻辑说明:where参数支持等于过滤,n_results可以传大于top_k的数,比如10,然后在代码里再按阈值截断到5个。返回的distances在cosine空间下是距离,我用1 - dist转成相似度,方便统一理解。
参数说明:相似度阈值没有绝对标准,取决于embedding模型和语料分布。在408资料上,我通常先打印一次检索的相似度分布,看正确chunk落在什么区间。常见情况是相关片段在0.78到0.95之间,不相关的在0.6以下。阈值定在0.72到0.78左右比较稳妥,如果发现错误召回多,就往上调。注意:不同embedding模型之间的相似度分数不可横向对比,换模型就得重新标定。
3.4 混合检索:用BM25补齐向量检索的细节漏召回
只做纯语义检索,很快会遇到RAG瓶颈:向量检索擅长“意思相近”的匹配,但不擅长“字面精确”的匹配。408里有大量需要精确命中的东西——inode、PAV、TLB、RSA、三次握手、川大智驾这类缩写和专有名词。问题描述如果和教材原话措辞不同,向量还能召回;但问“paas和iaas区别”,如果原始资料里写的是“平台即服务、基础设施即服务”,向量可能把很多讲云服务的chunk都带出来。
常见的补救方案是混合检索:向量召回一批语义相近结果,BM25通过词频召回一批字面命中的结果,合并去重后再重排。408规模小,用rank_bm25就够。
from rank_bm25 import BM25Okapi import jieba tokenized_corpus = [list(jieba.cut(doc)) for doc in all_docs] bm25 = BM25Okapi(tokenized_corpus) def bm25_search(query: str, top_k: int = 5): query_tokens = list(jieba.cut(query)) scores = bm25.get_scores(query_tokens) top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k] return [all_docs[i] for i in top_indices]逻辑说明:BM25 是经典的词频-逆文档频率算法,对精确关键词非常敏感。用jieba.cut做中英文混合切分,是为了让中文词和英文缩写都能被正确分词。这里all_docs是原始chunk列表,检索后拿到的是索引,再映射回文本。
实际操作中,我会让向量召回取6个、BM25取4个,合并后按member分数重新排序。初始权重可以设为向量0.7、BM25 0.3,但这个比例需要根据测试集调,后面第6章会讲具体评估方法。混合检索的本质是给“语义模糊题”和“术语精确题”都留一条路,不要指望单一检索器通吃。
4. 问答链路:把检索结果交给智谱清言API,Prompt划边界
4.1 系统提示词:告诉模型“不知道就承认”
检索做得好,只解决了一半问题。另一半是生成环节,也就是如何让智谱清言大模型API严格依据检索片段作答,而不是把RAG结果当成背景参考后自由发挥。这一层靠系统提示词划边界。
很多人在用户问题前拼接一段“以下是从知识库中找到的资料……”,但忘记在系统层声明数据来源和回答边界。结果模型仍然按自己的知识惯性回答,幻觉没有被抑制。正确的做法是把边界写进系统提示词,因为系统提示词发生在每一轮对话之前,优先级比用户内容更高。
SYSTEM_PROMPT = """你是一个计算机考研408科目的答疑助手。 请严格依据用户消息中提供的“知识库片段”回答。 规则: 1. 只能使用知识库片段中出现的事实和表述; 2. 片段不足以回答时,直接回答“知识库中未找到相关依据”,不要根据常识补充; 3. 回答要面向考研答题场景,先给结论,再给关键过程或理由; 4. 如果使用了某个片段,在句末标注片段来源,例如(来源:王道_操作系统); 5. 多个片段结论冲突时,报告冲突,不要自行调和。"""这里面最关键的是第2条和第5条。第2条约束幻觉,第5条处理知识库本身的矛盾——408的不同教材在个别术语定义上有差异(比如进程状态划分有的分三态有的分五态),让模型暴露冲突,而不是强行编一个统一答案。
4.2 构造注入上下文:引用来源与控制Token
检索结果不能全部塞进prompt。Top-K拿到6到10个chunk,每个chunk几百字,全部拼接后可能达到2000到3000字,再加上历史消息和系统提示词,离模型上下文窗口上限很近。所以需要裁剪和结构化。
def build_user_context(query: str, chunks: list[dict], max_chars: int = 1500) -> str: parts = [] total = 0 for i, chunk in enumerate(chunks, 1): source = chunk["metadata"].get("source", "未知") doc = chunk["doc"] if total + len(doc) > max_chars: break parts.append(f"[{i}](来源:{source})\n{doc}") total += len(doc) context = "\n\n".join(parts) user_content = f"知识库片段:\n{context}\n\n问题:{query}" return user_content逻辑说明:按顺序拼chunk,一旦总长度超过max_chars就截断。max_chars要根据你最终调用的模型token上限来定,一般对话模型能接收几千到几十k token,但给RAG用的时候,我习惯把注入内容控制在1500到2500字符,留下空间给系统提示词和历史会话。
这背后有个容易被忽略的细节:向量库召回的chunk是按相似度排序的,但不代表每个chunk都值得进入prompt。如果第3个chunk的相似度已经低于阈值,正确做法是直接丢弃,而不是硬拼进去。注入的垃圾片段越多,模型越容易被带偏。
4.3 调用智谱清言API:请求参数与限流重试
组装好消息后,就进入模型生成环节。智谱清言API兼容OpenAI SDK风格,调用方式非常标准。
def ask_rag(query: str, chunks: list[dict], history: list[dict] | None = None): history = history or [] user_content = build_user_context(query, chunks) messages = ( [{"role": "system", "content": SYSTEM_PROMPT}] + history[-4:] # 只保留最近四轮会话 + [{"role": "user", "content": user_content}] ) resp = client.chat.completions.create( model="glm-4", # 以智谱开放平台当前提供的对话模型名为准 messages=messages, temperature=0.2, top_p=0.7, max_tokens=1024, ) return resp.choices[0].message.content逻辑说明:history[-4:]把多轮会话截到最近两轮完整问答(一问一答算一轮)。这个截断很重要,下面第5章会展开讲。
参数说明:temperature=0.2是问答场景下偏保守的值,低温度能减少模型自由发挥,适合考试题目这种有标准答案的领域。top_p=0.7配合温度共同控制采样分布,我对408问答常用这个组合。max_tokens=1024限制答案长度,考研题目通常几百字足够,太长反而容易带入编造内容。
如果要处理高频请求,建议做限流重试。智谱API这类开放接口通常有每分钟请求数限制,超限会返回429或类似错误。直接暴露给用户会造成“答不上来”的体验。
import time import random def call_with_retry(func, retries=4, base_wait=0.6): for attempt in range(retries): try: return func() except Exception: wait = base_wait * (2 ** attempt) + random.uniform(0, 0.3) time.sleep(wait) raise RuntimeError("API调用多次失败,请稍后再试")逻辑说明:指数退避让每次重试等待时间翻倍,同时加入随机抖动,避免多个请求在同一瞬间集体重试。base_wait=0.6是个人常用值,如果API文档给出了限流窗口,可以直接按窗口大小调。
4.4 多轮会话:别让历史消息撑爆上下文
RAG本身是单轮问答,但408学习场景往往是连续的:先问“什么是分页存储”,再追问“分页和分段有什么区别”。第二问需要上文里的“分页”指代。这时候需要把历史问题也传给模型。
最省事的方式是直接把整段历史消息拼进messages。但这样做有一个隐藏问题:上一轮检索出的10个chunk已经消耗了大量token,如果多轮场景每轮都保留,上下文很快会超限。正确做法是历史消息只传“问答压缩后的文本”,不传历史检索chunk。
def compress_history(history: list[dict], max_rounds: int = 2) -> list[dict]: # 历史消息只保留最近max_rounds轮,每轮只存用户问题和助手回答 compressed = [] for msg in history[-max_rounds * 2:]: if msg["role"] == "user": # 用户消息里只取“问题:”后面的部分,丢弃“知识库片段”前缀 text = msg["content"].split("问题:")[-1] else: text = msg["content"] compressed.append({"role": msg["role"], "content": text}) return compressed逻辑说明:history里的用户消息是在build_user_context之后生成的,包含了知识库片段,不能再作为历史传给下一轮。压缩函数用split("问题:")截取真正的提问,把冗长的上下文剥掉,只保留精简版。max_rounds=2表示只看最近两轮,既维持对话连贯,又控制token增长。
如果项目会放大规模,可以用一个大模型对历史做摘要,替换原始消息。个人做408学习系统,先压缩历史就够用了,费用也更省。
5. 408问答系统避坑:五个真正让人返工的问题
5.1 切分切断知识点,答案总是差半句
现象:问“进程调度算法的评价指标”,答案只给了吞吐量、周转时间,丢失了等待时间和响应时间。人工检查发现,知识库里完整内容被分成了两个chunk,检索只命中了前半段。
原因:固定chunk_size不管文档结构,知识点的收尾刚好落到chunk边界。重叠参数也救不了,因为重叠只是边缘的冗余,知识点本身被硬切成了独立的两段。
解决:我后来把切分策略改成“先按章节标题粗分,再在粗分块内部设定chunk_size”。用第2章提到的RecursiveCharacterTextSplitter,但separators的第一优先级是\n###,让每个###三级标题带的内容尽量单独成块。对于标题长度超过500字的节,再按句号切。这样主知识点整体命中概率高很多。改完切分参数后,需要回看每个chunk的前20字和后20字,确认没有在半句话上断掉。
5.2 表格被拆成碎片,检索到却读不通
现象:问“TCP状态转换中的TIME_WAIT出现在哪”,模型说“知识库未找到”,但资料里明明有完整的TCP状态图表格。
原因:PDF提取时表格被按行拍平,每行一个chunk。“SYN_SENT → ESTABLISHED”这类行的语义,脱离表头“状态转换”就没意义。切分器再把表格行拆开,检索时命中的chunk里全是孤立行。
解决:表格不能当普通正文切。我在预处理里增加规则:识别连续多行以“|”或空格分隔的文本,将其前后加上```table标记,并且允许切分器把整个表格当作一个不可再分的整体——也就是临时把表格的chunk_size设得足够大。同时我给表格chunk添加元数据table=True,检索时如果问题里包含“转换”“状态”“对比”等词,可以只用where={"table": True}过滤到表格型内容。图片型表格没法直接处理,向量数据库本身不存图片,需要先把图片OCR成文字再进库。
5.3 语义召回一堆“像但不对”的内容
现象:问“死锁的四个必要条件”,系统召回来的是“死锁的预防、避免、检测与解除”那一章的内容,相似度也排在前面。答案虽然没答错,但引用来源方向错了,模型自己会被绕晕。
原因:向量检索按语义接近度排名,“死锁的必要条件”和“死锁的避免”在语义空间里距离很近,尤其是同一个文档里章节相邻,上下文向量互相污染。
解决:第一个动作是加元数据过滤,把where限定到“操作系统 / 死锁”章。第二个动作是提高top_k后做重排序:先召回10个,再用一个轻量本地交叉编码器或者关键词规则打分。最简单的规则是,把用户问题里的核心名词和每个chunk文本做字面匹配,命中数多的chunk排名提到前面。混合检索章节里提的BM25实际上也能缓解,但真正的稳定方案是重排序。对一个几十万chunk的项目,重排序只用跑10个候选,成本很低。
5.4 API限流和偶发超时,多问几次就崩
现象:本地测试前几十个问题都正常,连续提问到第50个左右,请求开始偶发报错,再往后频繁失败。
原因:开放API的速率限制不是按“用户请求一次”粒度处理,而是按每分钟请求数和token消耗量双重限制。个人本地测试没有做任何防抖,一下就打满配额。
解决:同步做两件事。第一,对生成调用包上指数退避重试,也就是第4章里call_with_retry的代码;第二,在前端入口加一个简单的串行化,保证同一时刻最多只有一个生成请求在跑。408问答是低频学习场景,不需要并发。如果确实要多会话并行,可以维护一个线程池,最大并发3到5,配合令牌桶限制每分钟请求数。出现429时,重试间隔至少大于API返回的retry_after字段,没有该字段就用2秒兜底。
5.5 改一条资料,被迫重建整个向量库
现象:发现某教材一个知识点表述有误,修正后想重跑,结果还要把全部几千个chunk重新embedding再写入,耗时十几分钟,还把原有的检索链路搞乱了。
原因:向量库没有按文档维度做覆盖管理,而是全量删除全量重加。更新一条,整个collection里的id全部变了,旧引用全部失效。
解决:每个文档在入库时分配固定ID前缀,比如os_wangdao_2024_。更新时先按该前缀查询并删除旧chunk,再把新文档切片后写入相同的ID空间,保证关联不会乱。
def upsert_document(doc_prefix: str, chunks: list[str], metadata: dict, embeddings: list): existing = collection.get(where={"doc_id": doc_prefix}) if existing["ids"]: collection.delete(ids=existing["ids"]) new_ids = [f"{doc_prefix}_c{i}" for i in range(len(chunks))] collection.add( documents=chunks, metadatas=[{**metadata, "doc_id": doc_prefix} for _ in chunks], embeddings=embeddings, ids=new_ids, )逻辑说明:where={"doc_id": doc_prefix}先定位旧chunk,删除后重新添加。每次更新只影响一个文档,不会动其它内容。doc_id字段必须在入库时就写好,这也是第3章强调元数据设计的原因之一。
如果你用的是Chroma之外的FAISS或Elasticsearch方案,思路一样:维护一个“文档到chunk id”的映射表,实现文档级原子替换。这个坑越早处理越好,资料更新是408备考过程中必然发生的。
6. 让系统真正可用:评测集、缓存与混合检索调优
走到这一步,Demo已经能跑,但要判断“能不能投入日常使用”,必须做定量验证。我见过太多RAG项目处于“看着不错,一问具体就露馅”的状态,所以请准备一份覆盖四门课、约50到100题的评测集。
评测集不用复杂,每一条包含:问题、期望命中的chunk id、标准答案要点。跑一遍检索,统计期望chunk是否出现在top 5里,这就是命中率;再跑一遍问答,人工判断答案是否覆盖标准答案要点。混合检索的权重调整要基于这个结果,而不是拍脑袋。
def evaluate_hit_rate(qa_pairs, retriever, top_k=5): hit = 0 for q, gold_id in qa_pairs: results = retriever(q, top_k=top_k) if any(r["metadata"]["chunk_id"] == gold_id for r in results): hit += 1 return hit / len(qa_pairs)逻辑说明:retriever可以是纯向量、纯BM25或混合检索,把它抽象成函数后,就能快速对比不同配置。chunk_id需要在入库时写在元数据里,否则无法定位期望命中对象。
混合检索的调优,我通常的做法是:把向量相似度和BM25分数分别归一化到0到1,然后枚举vector_weight从0.3到0.8,在评测集上选命中率最高的组合。如果发现很多问题靠向量召回、BM25反而引入噪音,就把向量权重提高;如果发现很多术语问题靠BM25才能救回来,就保底保留前2个BM25结果,不给BM25太多票数。
真正投入前,我还会加一个答案缓存层。408的问题重复度不低,“什么是死锁”“进程和线程的区别”这类问题被反复问,每次都调API既慢又烧钱。最简单的是用SQLite缓存:问句的向量或hash作为key,答案、来源、时间作为value。
import sqlite3 import hashlib def get_cached_answer(question: str): key = hashlib.md5(question.encode("utf-8")).hexdigest() row = conn.execute("SELECT answer FROM cache WHERE query_key=?", (key,)).fetchone() return row[0] if row else None def set_cached_answer(question: str, answer: str): key = hashlib.md5(question.encode("utf-8")).hexdigest() conn.execute("INSERT OR REPLACE INTO cache (query_key, answer) VALUES (?,?)", (key, answer)) conn.commit()逻辑说明:缓存只针对完全相同的问句,不做语义匹配,避免错缓存。如果后续想更智能,可以用向量检索在缓存库中找相似问句,但会引入额外复杂度,408场景用精确hash缓存已经能降不少成本。
说一个我自己的教训:第一次搭这套系统时,我迷信向量检索,把BM25和重排序当成“老派”技术忽略掉,结果几个常见概念题翻车。后来加了混合召回、重排序和评测集,系统才从“能聊天”变成“能答题”。另一个习惯是,每改一次切分或检索策略,先跑一遍评测集再上线,不要凭一次样例问答的感觉做判断。这套系统真正常态使用后,我会固定每周刷新一次知识库,把新错题和新笔记增量入库,让向量库跟着复习进度走。希望对你也有用。
本文还有配套的精品资源,点击获取