Python 智能问答机器人：BM25 检索增强与 FastAPI 落地

简介这份PDF文档面向希望用Python搭建智能问答机器人或聊天机器人的开发者与学习者尤其适合刚接触自然语言处理、需要在动手前理清技术路线的人。内容围绕问题界定、技术调研与方案选型展开提醒读者先理解问题本质与现有实践再决定采用基于规则的匹配方式还是引入词嵌入、序列到序列、注意力机制及BERT、GPT等预训练模型并涉及RNN、LSTM、GRU与Transformer等序列建模思路同时提到语料收集与预处理、损失函数与优化器选择、过拟合防范以及对话状态跟踪、多轮上下文保持和交互体验设计等环节可作为入门前的框架性参考。资源包含1个PDF文件压缩包约101KB体积轻便便于随时查阅与检索要点。目前已有193人学习适合作为构建聊天机器人前梳理知识脉络的一份参考材料。1. 从零搭 python 智能问答机器人先分清「能答」和「敢答」客服群里最常被转出来的一句话是「这个问题机器人答错了」。智能问答机器人真正难的地方不在于接上模型而在于让它知道什么不该答。python 实现智能问答机器人这条路上主流做法有三条规则匹配、检索式问答、生成式问答三者的成本和可控性完全不在一个量级。这篇文章按一线落地顺序展开先把 FAQ 语料整理成可检索的结构用 jieba 配合 TF-IDF 或 BM25 做召回再把命中的片段拼进提示词交给模型生成回答最后用 FastAPI 包成接口、用坏例表持续迭代。适合已经会写 python 基础语法、想做出一个能对内交付的问答服务的读者如果你只是刚查完 python 安装教程这里的代码也足够照着跑通。2. python 智能问答机器人的路线选型与环境落地2.1 规则匹配、检索式、生成式各自的边界规则匹配是 keyword 与正则的堆叠命中即返回不命中就走默认话术。它的响应基本在毫秒级答案完全可控可解释性最强。问题在于维护成本随问题数量线性上涨用户换个说法就漏「忘记密码」的规则匹配不到「密码找不到了」一个季度下来字典能写到几百行还不敢删。检索式问答把每条标准问题向量化用户 query 也向量化算相似度取 TopK。问题空间封闭、有标准答案的场景特别合适比如产品参数、发票流程、退换货政策。它天然支持「答不出来就明确说答不出来」不会编。代价是需要一份还算干净的语料同义改写的鲁棒性中等。生成式问答直接把问题丢给模型。语言流畅度最好但没有任何知识边界问它你们公司的报销额度是多少它能一本正经地编一个数字出来。检索加生成的组合才是我一般会推荐的做法检索负责「找到依据」生成负责「把依据说成人话」同时用相似度阈值卡住没有依据的问题。2.2 一张表定路线最低可用组合选型不用纠结先看回答错了的代价有多大。答错了要赔钱、要担责任的场景规则或纯检索优先只是提升效率、用户能自行复核的场景可以上生成。维度规则匹配检索式纯生成检索 生成冷启动成本低中极低中答案可控性高高低中高同义改写鲁棒性差中好好幻觉风险无无高低单次响应延迟5ms 以内1050ms取决于模型取决于模型语料维护方式改规则改语料调提示词改语料 调提示词如果团队里没人做过类似系统我建议第一版只做 jieba BM25 召回加阈值兜底先不上模型。这样一周内就能内测且能把问题定位在语料质量上等到召回准确率稳定在八成以上再接生成层否则你分不清答错是检索没召回还是模型在编。2.3 python 环境与依赖安装的三个必查项第一版依赖不多jieba 做中文分词rank-bm25 做召回scikit-learn 提供 TF-IDF 参考实现后面接服务层再加 fastapi 和 uvicorn。用虚拟环境隔离避免和系统 python 安装包里已有的包冲突。# 1. 创建并激活虚拟环境python 3.9 及以上均可 python -m venv .venv source .venv/bin/activate # Linux / macOS # .venv\Scripts\activate.bat # Windows CMD # .venv\Scripts\Activate.ps1 # Windows PowerShell # 2. 安装依赖 pip install jieba rank-bm25 scikit-learn fastapi uvicorn[standard] requests # 3. 自检确认导入路径指向虚拟环境 python -c import jieba, sklearn, rank_bm25, fastapi; print(deps ok) python -c import sys; print(sys.executable)第二条自检命令打印出的路径里必须带.venv如果指向系统目录说明解释器选错了。用 vscode 配置 python 环境时按CtrlShiftP执行Python: Select Interpreter选中.venv下的解释器否则终端里跑得通、调试器里报 ModuleNotFoundError 是常事。提示把requirements.txt一并提交pip freeze requirements.txt换机器时pip install -r requirements.txt复现环境比口头同步包版本可靠得多。3. 语料与检索层jieba 分词加 TF-IDF 把 FAQ 变成可召回的知识库3.1 FAQ 语料的 json 字段设计语料结构决定了后面能不能做同义词扩展和多轮上下文。每条记录至少要有稳定 id、标准问题、别名列表、标准答案、分类。别名列表是最容易被忽略但收益最高的一列它把口语化问法直接收进语料等于白送召回率。[ { id: faq-001, category: 账号, question: 忘记密码怎么办, aliases: [密码找不到了, 登录密码怎么重置, 改密码], answer: 在登录页点击“忘记密码”输入注册手机号获取验证码验证通过后可设置新密码。, updated_at: 2024-05-01 }, { id: faq-002, category: 发票, question: 怎么开发票, aliases: [发票申请入口在哪, 开票流程], answer: 订单完成后 7 天内在“我的订单”详情页点击“申请开票”填写抬头与税号电子发票 24 小时内发到预留邮箱。, updated_at: 2024-05-01 } ]字段说明id用于回答时回传引用来源方便排查category用来做分类过滤比如只在「发票」类里检索aliases在入库时会和question一起展开成多条待检索文本但共享同一个idupdated_at是给运营看的政策类答案过期是问答机器人最常见的线上事故来源。语料从哪来如果官网有帮助中心用 python 爬虫把标题和正文抓下来再人工清洗比让业务方从零写快得多数量控制在 200 到 500 条覆盖八成高频问题就够第一版用了。3.2 jieba 分词、停用词与文本归一中文没有天然空格必须先分词。归一化这一步别省标点、大小写、全半角不统一会让相似度算得莫名其妙。import json import re import jieba # 常见疑问词和语气词去掉后能让相似度更聚焦在实词上 STOPWORDS set(的 了 吗 呢 吧 啊 是 我 你 请问 一下 怎么 如何 什么.split()) def normalize(text: str) - str: 转小写、去标点与空白只保留中文、字母、数字 text text.strip().lower() return re.sub(r[^\u4e00-\u9fa5a-z0-9], , text) def tokenize(text: str) - list[str]: 分词并过滤停用词单字保留数字其余至少两字 words jieba.lcut(normalize(text)) return [ w for w in words if w not in STOPWORDS and (len(w) 1 or w.isdigit()) ] def load_corpus(path: str) - list[dict]: with open(path, r, encodingutf-8) as f: return json.load(f) def build_docs(corpus: list[dict]) - tuple[list[str], list[str]]: 把问题和别名展开成检索用的文本列表同时保留 id docs, doc_ids [], [] for item in corpus: for text in [item[question], *item.get(aliases, [])]: docs.append(text) doc_ids.append(item[id]) return docs, doc_ids逻辑说明normalize处理的是字符层面把「忘记密码怎么办」和「忘记密码怎么办」抹平tokenize处理的是词层面砍掉疑问词之后「密码 找 不到」这类实词才不会被「怎么」「请问」稀释。build_docs把一条 FAQ 展开成多行检索时任何一个别名命中都算这条 FAQ 命中回答时用doc_ids反查原文不会重复输出。这是最影响最终效果的一步。如果你的语料里口语化问法多考虑加载自定义词典来修正分词边界jieba.load_userdict(dict.txt)每行写「词 词频 词性」把产品名、业务黑话塞进去避免被切碎。3.3 TF-IDF 加余弦相似度的最小可跑实现先用 scikit-learn 把整条链路跑通确认语料没问题再换 BM25。TF-IDF 的思路是一个词在当前文档里出现越多、在整个语料里出现越少它对这个文档的代表性就越强权重越高。import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity docs, doc_ids build_docs(load_corpus(data/faq.json)) # tokenizer 直接复用上面的分词函数token_patternNone 让 sklearn 不再自己切词 vectorizer TfidfVectorizer( tokenizertokenize, token_patternNone, lowercaseFalse, ) doc_matrix vectorizer.fit_transform(docs) # 形状: (文档数, 词表大小) def search(query: str, top_k: int 3): q_vec vectorizer.transform([query]) scores cosine_similarity(q_vec, doc_matrix)[0] # 每个文档一个 0~1 的分数 order np.argsort(scores)[::-1][:top_k] return [ {id: doc_ids[i], score: round(float(scores[i]), 4), text: docs[i]} for i in order if scores[i] 0 ] print(search(密码忘了))参数说明tokenizertokenize让向量化过程沿用我们自己的分词与停用词逻辑token_patternNone是必须的否则 sklearn 会用它默认的字符切分规则覆盖你的 tokenizer中文结果会很难看top_k控制返回条数一般取 3多了会干扰生成层scores[i] 0过滤掉完全不相关的文档。跑通之后先别急着接模型拿三十个真实问法测一遍看 Top1 命中多少。行业经验是纯 TF-IDF 在短问句上的 Top1 命中率大致在五到七成剩下的差异主要来自分词和语料覆盖度不是算法问题。3.4 换 BM25Okapik1 和 b 怎么调BM25 在 TF-IDF 基础上做了两处修正词频饱和和文档长度归一。短问句检索里它通常比 TF-IDF 更稳因为长文档不会因为多出现几次词就霸榜。from rank_bm25 import BM25Okapi tokenized_docs [tokenize(d) for d in docs] # k1 控制词频饱和速度b 控制文档长度归一强度 bm25 BM25Okapi(tokenized_docs, k11.5, b0.75) def search_bm25(query: str, top_k: int 3): tokens tokenize(query) scores bm25.get_scores(tokens) order np.argsort(scores)[::-1][:top_k] return [ {id: doc_ids[i], score: round(float(scores[i]), 4), text: docs[i]} for i in order if scores[i] 0 ]参数取值区间作用调法k11.2 ~ 2.0词频饱和速度越小越快饱和问题短、重复词少用 1.2长句多可提到 2.0b0.3 ~ 0.9文档长度归一强度0 完全不归一文档长度参差大用 0.75 以上长度齐整可降到 0.3注意 BM25 的分数是无界正值不像余弦相似度落在 0 到 1 之间。所以阈值兜底不能照搬 TF-IDF 的 0.5得先跑一批正负样本看命中问题的分数分布落在哪个区间取一个能挡住八成负样本的值作为阈值。这一步没有公式只有你自己的数据。4. 生成层把检索片段拼成提示词接上兼容 OpenAI 协议的模型4.1 提示词模板的结构生成层要做的事只有一件把检索到的标准答案当事实依据让模型改写成人话。模板里必须有三块——角色与约束、参考资料、用户问题缺一块模型就爱自己发挥。PROMPT_TEMPLATE 你是一个企业客服助手请严格依据【参考资料】回答【用户问题】。 规则 1. 参考资料中没有的内容直接回答“这个问题我暂时没查到建议转人工客服”不要推测。 2. 不要编造任何链接、电话、金额和时限。 3. 回答控制在三句话以内语气简洁。 【参考资料】 {context} 【用户问题】 {question} def build_prompt(question: str, hits: list[dict], corpus_map: dict) - str: # 同一 id 的多个别名命中只保留一条避免上下文重复 seen, blocks set(), [] for h in hits: if h[id] in seen or h[id] not in corpus_map: continue seen.add(h[id]) item corpus_map[h[id]] blocks.append(f[{item[id]}] {item[question]}{item[answer]}) context \n.join(blocks) if blocks else 无 return PROMPT_TEMPLATE.format(contextcontext, questionquestion)逻辑说明corpus_map是以 id 为键的语料字典用它在检索命中后取回标准答案原文而不是把相似度高的那些别名文本塞进上下文。别名是给检索用的答案是给生成用的两者不要混。参考资料里带上[faq-001]这样的编号方便回答里做引用标注也方便线上排查是哪条语料导致的错误。4.2 调用封装超时、重试与流式调用层建议统一走 OpenAI 兼容协议这样本地推理服务和云端接口的代码可以完全一致换服务方只改 base_url 和 model 两个变量。import time import requests def chat(messages, base_url, api_key, model, timeout20, max_retries2, temperature0.2, max_tokens512): 调用 /v1/chat/completions失败按指数退避重试 url f{base_url.rstrip(/)}/v1/chat/completions headers {Authorization: fBearer {api_key}, Content-Type: application/json} payload {model: model, messages: messages, temperature: temperature, max_tokens: max_tokens} for attempt in range(max_retries 1): try: resp requests.post(url, headersheaders, jsonpayload, timeouttimeout) resp.raise_for_status() return resp.json()[choices][0][message][content].strip() except requests.RequestException as e: if attempt max_retries: raise time.sleep(2 ** attempt) # 1s, 2s 退避参数说明temperature在问答场景压到 0.2 甚至 0让它老实复述资料max_tokens给 512 足够三句话设太大反而增加超时概率timeout设 20 秒超过就走兜底话术别让用户干等max_retries给 2覆盖瞬时网络抖动再多会让接口整体延迟不可控。raise_for_status别省否则 401、429 这类错误会被静默当成正常返回排查时只能靠猜。4.3 阈值兜底与会话状态兜底逻辑放在检索之后、调用模型之前Top1 分数低于阈值直接返回预设话术并给一个转人工入口不消耗模型调用。FALLBACK 这个问题我暂时没查到建议转人工客服或换个说法再问一次。 CANDIDATE 你是想问下面这些吗{options} def answer(question, corpus_map, threshold0.45, top_k3): hits search_bm25(question, top_ktop_k) if not hits or hits[0][score] threshold: # 分数略低但仍有候选时给用户反问选项比直接拒答体验好 if hits and hits[0][score] threshold * 0.6: opts 、.join(corpus_map[h[id]][question] for h in hits[:3] if h[id] in corpus_map) return CANDIDATE.format(optionsopts), None return FALLBACK, None prompt build_prompt(question, hits, corpus_map) return prompt, hits会话状态不要往模型侧塞服务端自己存。多轮对话只需要保留最近 3 到 5 轮的问答对超过就裁剪最旧的因为客服场景里用户很少追问到第五轮以上上下文越长越容易把前面的话题带偏同时 token 成本也上去了。存储用内存字典加超时淘汰即可单机服务没必要上 Redis。4.4 FastAPI 暴露 /ask 接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() corpus load_corpus(data/faq.json) corpus_map {item[id]: item for item in corpus} class AskIn(BaseModel): question: str session_id: str default app.post(/ask) def ask(body: AskIn): prompt, hits answer(body.question, corpus_map) if hits is None: # 走了兜底不调模型 return {answer: prompt, source: []} reply chat( messages[{role: user, content: prompt}], base_urlhttp://127.0.0.1:11434, # 本地推理服务地址换云端只改这里 api_keyEMPTY, modelqwen2.5:7b, ) return {answer: reply, source: [h[id] for h in hits]}启动命令uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2。--workers按 CPU 核数的一半给jieba 首次加载词典有开销进程数别开太少否则第一个请求会明显卡顿。返回体里始终带上source这是线上排查的唯一线索——用户说答错了你能立刻知道它依据的是哪条语料。注意base_url和api_key走环境变量注入别写死在代码里提交到仓库。本地推理服务通常不校验密钥填占位符即可。5. 上线后的验证与召回优化5.1 用命中率和坏例表做离线评测上线前必须有一份固定的评测集否则每次改语料都是凭感觉。准备 100 条真实问法人工标注正确答案对应的 FAQ id做成 jsonl每条一行。import json def evaluate(path: str, threshold: float 0.45): total hit1 hit3 fallback 0 bad_cases [] for line in open(path, encodingutf-8): case json.loads(line) hits search_bm25(case[query], top_k3) total 1 if not hits or hits[0][score] threshold: fallback 1 bad_cases.append({query: case[query], got: fallback, expect: case[gold_id]}) continue ids [h[id] for h in hits] hit1 ids[0] case[gold_id] hit3 case[gold_id] in ids if ids[0] ! case[gold_id]: bad_cases.append({query: case[query], got: ids[0], expect: case[gold_id]}) print(fTop1 命中率 {hit1/total:.2%} | Top3 {hit3/total:.2%} | 兜底率 {fallback/total:.2%}) return bad_cases三个指标一起看Top1 命中率反映直接体验Top3 命中率反映生成层有没有救回来的空间兜底率过高说明语料覆盖不够。把bad_cases按「语料缺失」「分词错误」「别名不足」分类前两类补语料第三类进下一步。评测集要固定每次改完语料重跑一遍别用同一批数据同时调参和验收。5.2 别名词典扩展召回分词边界错误和口语化问法是坏例的主要来源用一个别名字典在检索前做 query 改写成本最低、见效最快。SYNONYMS { 密码: [口令, passwd], 发票: [开票, 票据], 退款: [退钱, 返款], } def expand_query(query: str) - str: 把命中的同义词追加到原 query 后面不替换保留原始语义 extra [w for k, vs in SYNONYMS.items() if k in query for w in vs] return query .join(extra) if extra else query逻辑说明追加而非替换避免改写本身引入噪声SYNONYMS从坏例表里长出来每处理一批坏例加几条半年下来这张表就是最贴合你业务的分词词典。用户问「口令找不到了」扩展后变成「口令找不到了 密码」BM25 里「密码」的权重会把它拉进正确的语料。这套改写不依赖模型延迟几乎为零是纯检索和检索加生成两条路线都该先加的一层。本文还有配套的精品资源点击获取