ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python问答系统实战:BM25与向量检索混合召回从零搭建

Python问答系统实战:BM25与向量检索混合召回从零搭建 简介这份资源是面向高校学生与Python初学者的一套问答系统课程设计完整方案围绕文本检索、问题分类、候选答案句排序与答案抽取四个核心环节展开帮助读者理解从语料预处理到答案输出的全流程实现思路。压缩包共35个文件约33.35MB以14个Python源码文件为主辅以14个JSON数据文件、2份Word设计报告与任务书、若干txt与md说明文档源码按预处理、分词、向量化、检索、分类、答案句选择与答案片段抽取等模块拆分结构清晰便于对照学习。目前已有1003人学习下载适合作为课程设计参考或NLP入门练手项目。读者可从中获得完整的设计报告与任务书、可运行的问答系统源码及配套语料数据并借助问题分类体系与逐步调优的答案抽取流程掌握检索式问答系统的搭建方法与优化思路。1. 基于Python实现的问答系统设计从零到跑通的最小闭环很多人第一次搜「Python问答系统」的时候脑子里想的其实是一个能听懂人话、能查资料、能给出靠谱答案的机器人。但真动手写起来往往卡在第一步到底该用规则匹配、检索排序还是直接上大模型我见过太多人一上来就装 transformers、下几十 G 的模型权重结果环境没配好显存先爆了最后连一个「你好」都答不出来。其实一个能用的问答系统核心就三件事把问题变成向量、把知识库变成可检索的索引、把最相关的答案拼回去。Python 在这条链路上生态最全从 jieba 分词到 sentence-transformers再到 faiss 建索引全是现成轮子。这篇笔记面向的是想自己搭一套问答系统、但不想被框架绑架的开发者我会按「最小可跑通 → 逐步加检索 → 避坑 → 进阶调优」的顺序把每一步的命令、参数和翻车点都摊开讲。你不需要 GPU 也能跟完前四章最后一章再决定要不要上向量模型。2. 问答系统的三种技术路线与选型依据2.1 规则匹配、检索式、生成式到底怎么选先别急着写代码选错路线后面全是白干。规则匹配适合 FAQ 固定、问题句式变化少的场景比如内部工单系统的「如何重置密码」用正则加关键词就能覆盖 80% 的提问响应快、零依赖。检索式问答的核心是「问题-答案」对建索引用户提问后先召回候选再排序取 top1适合知识库有明确条目、答案不需要重新组织语言的场景比如产品手册、政策条款。生成式问答则是把检索到的片段喂给语言模型让它组织成自然语言回答适合答案需要跨段落整合的场景但成本和不可控性都高。我一般会先问三个问题知识库条目是否超过 500 条用户提问是否和知识库原文措辞差异大答案是否允许直接返回原文片段如果前两个是「是」、第三个是「是」检索式就够了如果第三个是「否」才考虑生成式。很多团队一上来就上生成式结果发现检索召回率不到 60%模型再强也救不回来。2.2 最小依赖环境搭建与目录结构环境这块python安装教程网上满天飞但问答系统有几个容易漏的库。我习惯用 venv 隔离不碰系统 Python。下面这套命令在 Windows 和 Linux 下都能跑注意 faiss-cpu 在 Windows 上要用 conda 装pip 直接装容易缺 DLL。# 创建虚拟环境Python 3.9 以上都行 python -m venv qa_env # Windows 激活 qa_env\Scripts\activate # Linux/Mac 激活 source qa_env/bin/activate # 核心依赖先装这些就够跑通检索式问答 pip install jieba rank_bm25 numpy pandas flask # 如果要用向量检索再加下面两个 pip install sentence-transformers faiss-cpu装完后目录建议这样分data/放知识库原始文件index/放建好的索引app.py是服务入口retriever.py封装召回逻辑。别把所有代码塞一个文件里后面换检索器的时候你会感谢自己。2.3 知识库预处理从原始文本到问答对知识库质量直接决定召回上限。我见过有人把整本 PDF 直接丢进去结果检索出来的全是页眉页脚。正确做法是先切成问答对每条包含question、answer、source三个字段。如果原始资料是文档用滑动窗口切段每段 200 到 300 字重叠 50 字避免答案被切断。import json import re def clean_text(text): # 去掉多余空白和特殊符号保留中文、英文、数字和常用标点 text re.sub(r\s, , text) text re.sub(r[^\u4e00-\u9fa5a-zA-Z0-9。、《》], , text) return text.strip() def build_qa_pairs(raw_docs): qa_list [] for doc in raw_docs: # 假设 doc 是 dict包含 title 和 content content clean_text(doc[content]) # 按句号、问号切分每 3 句合成一个片段 sentences re.split(r[。], content) for i in range(0, len(sentences), 3): chunk 。.join(sentences[i:i3]) if len(chunk) 20: continue qa_list.append({ question: doc[title] sentences[i][:30], answer: chunk, source: doc.get(source, unknown) }) return qa_list # 保存为 jsonl每行一条方便后续流式读取 with open(data/qa_pairs.jsonl, w, encodingutf-8) as f: for item in build_qa_pairs(raw_docs): f.write(json.dumps(item, ensure_asciiFalse) \n)这段代码的关键在clean_text里的正则\u4e00-\u9fa5覆盖常用汉字但会把生僻字和 emoji 干掉如果你的知识库有特殊符号把对应范围加进去。切分粒度 3 句是我试过比较稳的太短召回噪声大太长答案冗余。question字段用标题加首句拼接是为了让检索时问题侧也有足够关键词。3. 用 BM25 和向量检索搭出可用的召回层3.1 BM25 检索零训练成本的基线方案BM25 是检索式问答的保底方案不需要任何模型建索引秒级完成。它的核心思想是词频和逆文档频率加权再对长文档做惩罚。rank_bm25这个库用起来最简单但要注意它默认按空格分词中文必须先过 jieba。import jieba import json from rank_bm25 import BM25Okapi # 读取问答对 qa_pairs [] with open(data/qa_pairs.jsonl, r, encodingutf-8) as f: for line in f: qa_pairs.append(json.loads(line)) # 对答案和问题拼接后分词检索时问题侧也分词 corpus [jieba.lcut(item[question] item[answer]) for item in qa_pairs] bm25 BM25Okapi(corpus) def search_bm25(query, top_k5): tokenized_query jieba.lcut(query) scores bm25.get_scores(tokenized_query) # 按分数降序取 top_k ranked sorted(enumerate(scores), keylambda x: x[1], reverseTrue)[:top_k] return [(qa_pairs[i], score) for i, score in ranked] # 测试 results search_bm25(怎么重置密码) for item, score in results: print(score, item[answer][:50])BM25Okapi的k1和b参数默认是 1.5 和 0.75一般不用动。如果发现短问题召回差把b调到 0.5 左右减弱长度惩罚。分词这块jieba 默认模式会把「重置密码」切成「重置」「密码」没问题但如果你有专有名词比如「星露谷物语python编程网站」这种得用jieba.add_word加自定义词典否则会被切碎导致召回失败。3.2 向量检索用 sentence-transformers 提升语义召回BM25 的硬伤是词不匹配就召回不到。用户问「忘记登录口令怎么办」知识库里写的是「密码重置流程」BM25 分数会很低。向量检索把句子映射到语义空间能解决同义替换问题。sentence-transformers里paraphrase-multilingual-MiniLM-L12-v2这个模型只有 118MCPU 上跑推理也就几十毫秒适合起步。from sentence_transformers import SentenceTransformer import numpy as np import faiss # 加载多语言小模型第一次运行会自动下载 model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) # 把所有问答对的文本编码成向量 texts [item[question] item[answer] for item in qa_pairs] embeddings model.encode(texts, normalize_embeddingsTrue, show_progress_barTrue) # 建 faiss 索引用内积相似度因为向量已归一化 dimension embeddings.shape[1] index faiss.IndexFlatIP(dimension) index.add(embeddings.astype(float32)) def search_vector(query, top_k5): query_vec model.encode([query], normalize_embeddingsTrue) scores, indices index.search(query_vec.astype(float32), top_k) return [(qa_pairs[i], float(s)) for i, s in zip(indices[0], scores[0])] # 测试语义召回 results search_vector(忘记登录口令怎么办) for item, score in results: print(score, item[answer][:50])normalize_embeddingsTrue很关键归一化后内积等价于余弦相似度faiss 的IndexFlatIP才能直接用。top_k设 5 是召回和延迟的平衡点实际线上可以设 10 再加重排序。注意模型第一次下载需要网络如果环境受限提前把模型缓存到本地用cache_folder参数指定路径。3.3 混合召回BM25 和向量分数怎么融合单用一路召回都有短板我一般用加权融合。BM25 分数范围不固定向量相似度在 0 到 1 之间直接相加会偏向量。先把 BM25 分数做 min-max 归一化再按 0.3 和 0.7 加权这个比例在多数中文问答场景下比较稳。def hybrid_search(query, top_k5, alpha0.3): # 两路各取 20 个候选 bm25_results search_bm25(query, top_k20) vector_results search_vector(query, top_k20) # 归一化 BM25 分数 bm25_scores [s for _, s in bm25_results] if max(bm25_scores) - min(bm25_scores) 1e-6: bm25_norm [(s - min(bm25_scores)) / (max(bm25_scores) - min(bm25_scores)) for s in bm25_scores] else: bm25_norm [1.0] * len(bm25_scores) # 合并到字典key 用答案文本去重 merged {} for (item, _), norm_s in zip(bm25_results, bm25_norm): key item[answer][:50] merged[key] {item: item, score: alpha * norm_s} for item, s in vector_results: key item[answer][:50] if key in merged: merged[key][score] (1 - alpha) * s else: merged[key] {item: item, score: (1 - alpha) * s} ranked sorted(merged.values(), keylambda x: x[score], reverseTrue)[:top_k] return [(r[item], r[score]) for r in ranked]alpha是 BM25 的权重知识库术语多、用户提问规范时调到 0.5用户提问口语化、同义表达多时降到 0.2。去重用的answer[:50]是个偷懒做法更稳的是给每条问答对加唯一 id。4. 避坑与排查问答系统上线前必须过的五道坎4.1 中文分词把关键词切碎导致召回为零现象用户问「python安装sklearn库」BM25 返回结果全是无关条目向量检索也排不到正确条目。原因jieba 默认词典没有「sklearn」切成「sk」「learn」和知识库里的「sklearn」对不上。解决在分词前加载自定义词典把领域词加进去。import jieba # 每行一个词UTF-8 编码 jieba.load_userdict(data/user_dict.txt) # 或者动态添加 for word in [sklearn, numpy, pandas, faiss]: jieba.add_word(word)自定义词典要覆盖你知识库里所有专有名词包括库名、产品名、缩写。这个文件我一般从知识库的 title 字段自动抽取高频词生成比手写靠谱。4.2 向量模型下载卡住或显存不足现象运行SentenceTransformer时卡在下载或者报 CUDA out of memory。原因默认从境外源拉模型网络不稳定或者模型默认加载到 GPU 但显存不够。解决指定本地缓存路径强制用 CPU。import os os.environ[HF_HUB_OFFLINE] 1 # 离线模式前提是模型已缓存 model SentenceTransformer( paraphrase-multilingual-MiniLM-L12-v2, cache_folder./model_cache, devicecpu )第一次下载时不要设HF_HUB_OFFLINE下完后把model_cache目录整个拷到部署机器再开离线模式。CPU 推理 118M 模型单条 50ms 以内QPS 要求不高的场景完全够用。4.3 索引文件没持久化每次重启重建现象服务重启后要等几分钟才能响应日志显示在重新编码。原因faiss 索引和 embeddings 只存在内存里没落盘。解决建完索引后写文件启动时先读文件。# 保存 faiss.write_index(index, index/qa_faiss.index) np.save(index/qa_embeddings.npy, embeddings) # 加载 index faiss.read_index(index/qa_faiss.index) embeddings np.load(index/qa_embeddings.npy)注意 faiss 索引文件在不同版本间可能不兼容部署环境和开发环境的 faiss 版本要锁死写进requirements.txt。4.4 返回答案太长前端展示截断现象检索返回的 answer 有 500 多字前端只显示前 100 字用户看不到关键信息。原因切分粒度太粗或者没做答案摘要。解决在返回前按句子截取和问题最相关的部分或者直接限制 answer 长度。def truncate_answer(answer, query, max_len150): if len(answer) max_len: return answer # 按句号切分找包含 query 关键词的句子优先 sentences answer.split(。) query_words set(jieba.lcut(query)) best_sent max(sentences, keylambda s: len(query_words set(jieba.lcut(s)))) return best_sent[:max_len] ...这个函数是兜底方案更好的做法是在建库阶段就把答案控制在 200 字以内从源头解决。4.5 并发请求下模型推理排队现象单用户测试正常多用户同时请求时响应时间从 50ms 涨到 2s。原因SentenceTransformer.encode默认不是线程安全的Flask 默认单线程处理。解决用 gunicorn 多 worker 启动或者把编码服务单独拆出来。# 用 gunicorn 起 4 个 worker每个 worker 独立加载模型 gunicorn -w 4 -b 0.0.0.0:5000 app:app4 个 worker 意味着 4 份模型内存118M 模型大概占 500M 内存4 份 2G小机器要算好内存再定 worker 数。更省资源的做法是用torch.set_num_threads(1)限制每个 worker 的线程数避免 CPU 争抢。5. 进阶调优用重排序和缓存把准确率再提一截召回层解决的是「别漏」重排序解决的是「别错」。我一般会在混合召回取 top20 之后加一个 cross-encoder 重排序模型把问题和候选答案拼成一对输入直接输出相关性分数。cross-encoder/ms-marco-MiniLM-L-6-v2这个模型只有 80MCPU 上对 20 个候选排序大概 200ms换来的是 top1 准确率提升 10 到 15 个百分点。from sentence_transformers import CrossEncoder reranker CrossEncoder(cross-encoder/ms-marco-MiniLM-L-6-v2, devicecpu) def rerank(query, candidates, top_k3): # candidates 是 hybrid_search 返回的 (item, score) 列表 pairs [[query, item[answer]] for item, _ in candidates] scores reranker.predict(pairs) ranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue)[:top_k] return [(item, float(s)) for (item, _), s in ranked]重排序的代价是延迟如果 QPS 要求高可以只对 top5 做重排或者用 ONNX 加速推理。另一个提准确率的技巧是缓存把高频问题的检索结果缓存起来用functools.lru_cache或者 Redis命中率能到 30% 以上。from functools import lru_cache lru_cache(maxsize1000) def cached_search(query): candidates hybrid_search(query, top_k20) return rerank(query, candidates, top_k3)lru_cache的 key 是 query 字符串完全匹配才命中。如果想做模糊缓存得用向量相似度做 key那就复杂了起步阶段不建议。验证这套系统好不好别只看 top1 准不准。我习惯准备 50 条测试问题人工标注正确答案然后算三个指标top1 命中率、top3 召回率、平均响应时间。top1 低于 70% 就回去查分词和知识库切分top3 低于 90% 就加召回候选数或者换更大的向量模型。响应时间超过 500ms 就上缓存或者减 worker。最后说个血泪教训别在知识库没清洗干净的时候调模型参数。我见过团队花两周调向量模型最后发现是 PDF 里的乱码没去掉检索出来的全是噪声。先把数据质量搞上去再谈模型和参数顺序反了就是给自己挖坑。希望帮到你。本文还有配套的精品资源点击获取
返回列表