简介这套资源是一份基于Python构建的“火力发电厂知识问答库”检索式问答系统/对话系统实现面向电力行业信息化人员、NLP学习者及数据科学从业者用于解决垂直领域问答、信息检索与知识库对话等场景。压缩包共3个文件含2个txt文本作为问答数据、1个py主程序实现交互逻辑整体仅7KB虽体量小巧但结构清晰便于快速阅读和二次扩展。系统覆盖数据预处理、知识表示、问题理解、查询生成、答案检索与呈现等典型流程可结合TF-IDF等检索模型完成问题匹配。目前已有208人学习适合想从零理解检索式问答架构、用Python实现轻量级对话原型的学习者参考。代码量精简可直接运行查看效果并按需扩展领域问答库。1. 火力发电厂知识问答库的检索式问答系统解决的是“找到”而不是“生成”运行值班员在 DCS 报警窗里看到“MFT 动作”规程规定要在一段时间内确认火检冷却风压力他手边没有纸质规程翻 PDF 又不知道关键词该搜什么。这类场景在每个电厂都在发生信息都在厂里找不到、找不快、不敢信。用 Python 基于火力发电厂知识问答库做检索式问答系统核心思路不是训一个会说话的生成模型而是把运行规程、检修规程、事故案例、设备参数做成可检索、可溯源、可版本管理的语料库用户问一句话系统把最相关的原文段落、置信度和出处一起返回。相比直接接大模型这套方案输出可控、故障可定位也更容易通过安全合规评审。下面按实际落地顺序推进先建领域问答库再实现 BM25、向量与混合三层召回最后用对话状态包装成交互闭环。2. 知识问答库的建模从规程文档到可检索的检索单元2.1 问答库的数据来源、切分粒度与存储选型火力发电厂的知识问答库数据源通常是这几类运行规程与操作票、检修工艺卡、设备说明书、历年事故分析报告、班组技术问答台账。这些材料有一个共同点——本身已经是结构化程度较高的领域文档章节目录清晰、术语统一、参数和限值成段出现。所以建库第一件事不是分词而是设计切分粒度。常见做法是按文档原有章节结构切分而不是固定按 256 字、512 字硬切。电厂规程里“锅炉 MFT 动作处理”往往是一整节固定窗口切会把“动作条件”和“处理措施”切到两个块里检索时要么召回半截要么排序混乱。我一般用两级策略先用正则按“第X章”“X.X”这类标题把文档切成一级章节块对超过 600 字的块再按句号、分号切成子句组保证单个检索单元在 200600 字之间。存储上不用引入重型组件几万条语料用 SQLite 就足够。建表时把版本、批准日期、章节路径作为独立字段保留这直接决定后续能不能做答案溯源——检索式问答最重要的信任基础就是每条答案都能点开原文。CREATE TABLE corpus ( id INTEGER PRIMARY KEY AUTOINCREMENT, doc_name TEXT NOT NULL, -- 源文档名称如《锅炉运行规程》 chapter_path TEXT, -- 章节路径如 3.2.1 chunk_text TEXT NOT NULL, -- 切分后的检索单元 version TEXT, -- 文档版本号 approve_date TEXT, -- 批准日期 source_file TEXT -- 原始文件路径 ); CREATE INDEX idx_corpus_doc ON corpus(doc_name, chapter_path);逻辑说明chunk_text是后续分词、向量化的基本单位chapter_path用于结果展示时拼接出完整出处version和approve_date用来处理规程改版后的旧条目下线。建索引的目的是当按文档归约统计时不必全表扫描。存储选型上 SQLite 单文件即可不建议一上来就上 MySQL 或 ES语料没到百万级之前额外组件只增加部署成本。2.2 用 jieba 与自定义词典完成中文分词和文本清洗中文检索和英文最大的差异在分词。火力发电厂术语密度极高“主燃料跳闸”“汽轮机数字电液调节系统”“炉膛负压”这类词如果按通用分词切会被切成“主/燃料/跳闸”BM25 打分时“主燃料”和“跳闸”两个词的语义被割裂。所以需要在建库阶段加载领域词典。jieba 支持load_userdict每行格式是“词 词频 词性”词频可以不写让 jieba 自动调节。我维护一份火电领域的 userdict把 MFT、DEH、DCS、BMS、汽轮机轴封、火检冷却风、炉膛负压这类词全部放进去。注意 MFT 这类英文缩写要单独保留大写形式不能转成小写否则和规程原文对不上。import jieba import re # 加载领域词典userdict.txt 每行一个词可带词频和词性 jieba.load_userdict(power_plant_dict.txt) STOP_WORDS {的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个, 上, 也, 很, 到, 说, 要, 去, 你, 会, 着, 没有, 看, 好, 自己, 这} def clean_text(text: str) - str: # 去掉文档页眉页脚常见噪声保留中文、英文、数字和常用标点 text re.sub(r\s, , text) text re.sub(r[^\u4e00-\u9fa5A-Za-z0-9。、\-\%], , text) return text.strip() def tokenize(text: str) - list[str]: text clean_text(text) words jieba.lcut(text) return [w for w in words if w.strip() and w not in STOP_WORDS and len(w) 1]逻辑说明clean_text用正则白名单过滤掉 PDF 转文本时混入的乱码和控制符tokenize过滤停用词和单字词是因为检索场景里单字词对 BM25 打分几乎只有噪声。len(w) 1这个过滤在中文里尤其有效能把“的、了、在”这类高频无意义字直接挡掉。注意这里没有做词性标注词性对检索召回没有直接帮助加了反而拖慢建库速度。2.3 将语料写入 SQLite 并导出检索索引的完整代码分词和清洗是预处理环节实际建库时建议写一个独立的脚本跑完一次后把结果落到 SQLite后续检索服务只读数据库不再重复分词。下面这段代码完成读取原始文本文件按正则切分章节清洗分词写入 SQLite并额外导出一份 JSON 供向量检索使用。import sqlite3 import json import re from pathlib import Path def split_by_chapter(text: str): # 按 “第X章” 或 “X.X” 标题切分返回 (章节路径, 段落) pattern re.compile(r^(第[一二三四五六七八九十百0-9][章节]|[0-9](?:\.[0-9]){1,3})\s(.*)$, re.M) matches list(pattern.finditer(text)) chunks [] for i, m in enumerate(matches): start m.end() end matches[i 1].start() if i 1 len(matches) else len(text) section_text text[start:end].strip() if len(section_text) 50: chunks.append((m.group(1), section_text)) return chunks def build_corpus(raw_dir: str, db_path: str, json_path: str): conn sqlite3.connect(db_path) conn.execute(DELETE FROM corpus) records [] for file_path in Path(raw_dir).glob(*.txt): text file_path.read_text(encodingutf-8) for chapter, content in split_by_chapter(text): tokens tokenize(content) if len(tokens) 10: continue record (file_path.stem, chapter, content, v1.0, 2024-01-01, str(file_path)) conn.execute( INSERT INTO corpus(doc_name, chapter_path, chunk_text, version, approve_date, source_file) VALUES (?,?,?,?,?,?), record, ) records.append({ id: conn.execute(SELECT last_insert_rowid()).fetchone()[0], text: content, tokens: tokens, doc: file_path.stem, chapter: chapter, }) conn.commit() conn.close() with open(json_path, w, encodingutf-8) as f: json.dump(records, f, ensure_asciiFalse) print(f共写入 {len(records)} 个检索单元)逻辑说明split_by_chapter的正则匹配“第一章”“3.2.1”这类标题用finditer拿到所有标题位置后把相邻标题之间的内容作为独立检索单元len(section_text) 50过滤掉只有标题没有正文的空壳章节。写入 SQLite 的同时导出 JSON是因为 BM25 建索引需要的是 token 列表向量化需要的是原文两份数据分开用更清晰。last_insert_rowid取回自增主键保证 JSON 里的 id 和数据库记录能对上。3. 检索式问答系统三种召回路线词法、向量与混合排序3.1 词法检索用 BM25小语料里仍然最能打检索式问答系统的核心是一套召回和排序机制。火力发电厂知识问答库的典型规模是几千到几万条检索单元在这个量级下BM25 依然是最稳的基线。BM25 的原理可以理解为“词频 文档长度归一化 逆文档频率”的加权求和它对常见词做惩罚对在少数文档里出现的领域词给更高权重。Lucene 和 Elasticsearch 的默认参数k11.5, b0.75在大多数中文场景下直接可用不需要一上来就调参。为什么在 2025 年的技术背景下还要先讲 BM25因为向量检索在小语料上经常出现“召回结果看似相关、实则张冠李戴”的问题比如问“汽轮机轴封漏汽处理”向量召回可能把“轴封系统投运”排在“轴封漏汽处理”前面两者字面重叠不多但语义接近对生产系统来说这个错误不可接受。词法检索在排除近义干扰上更严格——它要求字面上真的有重合。所以我的基线方案永远是 BM25后续要不要加向量取决于评测结果而不是取决于技术潮流。3.2 用 rank_bm25 在本地跑通检索式问答的最小实现rank_bm25是纯 Python 实现的 BM25 库没有外部依赖适合在开发环境快速验证。下面这段代码读取上一章导出的 JSON构建索引并返回查询 TopK。from rank_bm25 import BM25Okapi import json with open(corpus_index.json, r, encodingutf-8) as f: corpus json.load(f) tokenized_corpus [doc[tokens] for doc in corpus] bm25 BM25Okapi(tokenized_corpus, k11.5, b0.75) def search_bm25(query: str, top_k: int 5): query_tokens tokenize(query) scores bm25.get_scores(query_tokens) top_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue)[:top_k] results [] for idx in top_indices: if scores[idx] 0: continue results.append({ score: round(float(scores[idx]), 4), text: corpus[idx][text], doc: corpus[idx][doc], chapter: corpus[idx][chapter], }) return results逻辑说明BM25Okapi构造时传入的是全量分词后的语料k1控制词频饱和程度值越大词频对分数的影响越线性b控制文档长度归一化强度b0.75表示对长文档做较强惩罚避免长文本靠篇幅堆词频。查询时调用get_scores拿到每个文档的分数手动做 TopK 排序而不是用get_top_n是为了能在返回结果里保留分数用于后续混合排序。scores[idx] 0的过滤很关键——BM25 允许负分生产环境不该把负分文档作为“相关答案”展示给用户。3.3 向量检索补语义用 Faiss 召回近义表达BM25 的短板是查“轴封漏汽”匹配不到只写“轴封蒸汽泄漏”的文档。要补这个短板常见做法是引入句向量检索。流程分三步用 sentence-transformers 加载中文句向量模型把每条语料的chunk_text编码成向量再用 Faiss 建索引做相似度搜索。import numpy as np import faiss from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5) texts [doc[text] for doc in corpus] embeddings model.encode(texts, normalize_embeddingsTrue, batch_size32) embeddings np.asarray(embeddings, dtypefloat32) index faiss.IndexFlatIP(embeddings.shape[1]) index.add(embeddings) faiss.write_index(index, corpus_emb.index) def search_vector(query: str, top_k: int 5): q_vec model.encode([query], normalize_embeddingsTrue) scores, indices index.search(np.asarray(q_vec, dtypefloat32), top_k) return [{score: float(scores[0][i]), text: corpus[indices[0][i]][text]} for i in range(top_k)]逻辑说明normalize_embeddingsTrue和IndexFlatIP是配套的向量归一化后内积就是余弦相似度省去单独算模长的步骤。bge-small-zh-v1.5是国内常见的中文轻量句向量模型对电厂规程类文本的泛化表现稳定更重要的是它支持中文长句和相似度重排。模型编码时的batch_size按显存调纯 CPU 环境 16 或 32 都合适。Faiss 索引保存到磁盘后服务启动时一次性加载不需要每次重启都重新编码全量语料。向量维度取决于模型bge-small-zh-v1.5是 512 维对几万条语料建索引耗时在秒级性能完全不是瓶颈。3.4 混合召回与 RRF 融合排序参数选择BM25 和向量检索各有偏好直接拼接结果会有重复而且两种分数不在一个量纲上不能直接相加。业界常用的融合方式是 RRFReciprocal Rank Fusion它不看原始分数只看每条结果在各自召回列表里的排名位置。公式是把排名倒数累加常用k60。def rrf_fusion(bm25_results, vector_results, k60, top_k5): score_map {} def add_results(results, weight1.0): for rank, item in enumerate(results): key item[text] # 只取前 20 条参与融合排名太靠后的结果贡献趋近于 0 if rank 20: break score_map[key] score_map.get(key, 0.0) weight / (k rank 1) add_results(bm25_results, weight1.0) add_results(vector_results, weight1.0) merged sorted(score_map.items(), keylambda x: x[1], reverseTrue)[:top_k] return [{score: round(s, 4), text: t} for t, s in merged]逻辑说明weight参数可以用来调节两路召回的信任度实测在电厂问答场景里 BM25 和向量等权通常就够如果评测发现语义近义召回不足再把向量路权重加到 1.21.5。rank 20的截断基于一个观察RRF 里排在第 20 名之后的结果对最终分数的贡献不到 0.02参与融合只会引入噪声。融合后的score只用于排序不用于置信度展示——真正的置信度应该由最后一步的答案相关性判定给出。这段融合逻辑可以加一个分支当 BM25 最高分大于某个阈值比如 30时直接信任词法结果不启动向量召回。这样做的好处是省一次模型编码推理时间对“查规程原文”这类明确字面匹配的查询尤其划算。4. 把检索能力包装成对话系统意图、状态与交互循环4.1 对话系统入口处的意图识别三类问题走三条路检索式问答系统的“对话”二字不是指闲聊而是指系统能理解用户问题的类型、记住上下文状态、在信息不足时主动追问。火力发电厂知识问答场景里用户问题大致分三类查规程原文、问处理步骤、问设备参数。这三类问题对答案形态的期望不同——查原文要给出处问步骤要按顺序组织查参数要给出带单位的数值。所以在进入检索前先用轻量规则做意图分类。意图类别典型问法检索策略答案形态规程查询MFT动作后如何处理BM25 向量混合召回原文段落 章节出处步骤查询如何投运轴封系统限定检修/运行规程语料按序号重排的步骤列表参数查询火检冷却风压力低报值限定设备参数表语料数值 单位 依据条款规则分类不用机器学习一组关键词表加正则就够。比如包含“如何、怎么、步骤、操作”倾向步骤类包含“多少、值、定值、范围”倾向参数类。分类结果用来裁剪检索范围——查到第 2 章建好的 SQLite 里doc_name字段做过滤条件这比全库召回再过滤效率高得多也更符合“领域知识问答”的产品预期。4.2 多轮上下文把上一轮答案组织成本轮的候选集单轮检索的最大问题是用户追问时缺少指代消解。用户先问“汽轮机轴封漏汽怎么处理”再追问“压力维持多少”如果系统不记得上一轮在讲轴封系统“压力”这个词在全库检索会得到一堆无关结果。多轮处理的常见做法是维护一个会话状态对象把上一轮 TopK 结果的 doc 和 chapter 缓存下来本轮检索时把这些限定条件作为硬过滤条件。class SessionState: def __init__(self, max_history: int 3): self.history [] self.last_context None # {docs: [...], chapters: [...]} def update(self, query: str, results: list): self.history.append(query) if len(self.history) max_history: self.history.pop(0) if results: self.last_context { docs: list({r[doc] for r in results}), chapters: list({r[chapter] for r in results}), }逻辑说明max_history控制会话长度电厂问答场景里 3 轮足够超过后丢弃最早的问题防止上下文膨胀把检索范围限制死。last_context存的是文档名和章节路径的集合而不是原始文本目的是在不泄露答案的情况下缩小下一轮检索范围。这个设计的取舍在于追问时如果过度依赖上一轮的章节过滤可能漏掉跨章节的相关内容所以实际使用中我会把last_context作为加分项而不是过滤条件——命中上下文的文档分数加 2.0而不是直接排除其他文档。4.3 一个可启动的交互式问答主循环把前三章的内容串起来就是一个能直接在终端跑起来的对话系统。这个主循环包含完整链路读 question → 意图分类 → 范围裁剪 → 混合检索 → 多轮上下文更新 → 拼装回答。def chat_loop(): state SessionState() print(火力发电厂知识问答系统已启动输入 exit 退出) while True: query input(\n问题: ).strip() if query.lower() in (exit, quit): break # 1. 意图分类决定是否启用上下文过滤 intent classify_intent(query) filter_docs None if intent step and state.last_context: filter_docs state.last_context[docs] # 2. 混合检索 bm25_results search_bm25(query, top_k10) vector_results search_vector(query, top_k10) if filter_docs: bm25_results [r for r in bm25_results if r[doc] in filter_docs] vector_results [r for r in vector_results if r[doc] in filter_docs] final rrf_fusion(bm25_results, vector_results, top_k3) # 3. 答案组装 state.update(query, final) for i, item in enumerate(final, 1): print(f\n候选 {i}置信度 {item[score]}) print(item[text][:300]) print(f出处: {item.get(doc, )} {item.get(chapter, )})逻辑说明classify_intent返回意图类别step类问题才启用上一轮的文档过滤search_bm25和search_vector都取 Top10 而不是最终展示的 Top3留出融合排序的缓冲空间state.update在每轮结束前更新上下文保证下一轮追问可用。需要说明的是print(item[text][:300])截断展示只是终端演示用实际 Web 对话里应该返回完整chunk_text由前端控制折叠展开。4.4 对话系统上线前要处理的异常与降级检索式问答系统最怕的不是检索不到而是检索到了但答非所问还硬答。生产环境必须配三层降级策略第一层混合检索分数全部低于阈值可用 5.0 作为经验值时返回“未找到相关规程请尝试换用更具体的关键词或联系技术组查阅纸质规程”第二层向量检索服务不可用时降级为纯 BM25第三层BM25 索引加载失败时直接读 SQLite 做 LIKE 匹配兜底。这里强调一个容易被忽略的点rank_bm25库要求查询 token 必须在语料中出现过否则get_scores返回全 0所以tokenize(query)后要检查是否有 token 进入过索引词典。另外如果启动时遇到ModuleNotFoundError多半是rank_bm25、faiss、sentence_transformers这几个依赖没装全建议固定 requirements.txt 并用虚拟环境安装避免在系统 Python 环境里逐个pip install把依赖搞乱。5. 问答系统上线前最值得做的评测MRR、Recall5 与日志反馈5.1 用 MRR 和 Recall5 量化“这次改版有没有变好”检索式问答的评价不能靠“感觉回答变好了”要落到可量化的指标上。工程上最常用两个MRRMean Reciprocal Rank衡量正确答案排在结果第几Recall5 衡量正确答案是否出现在前 5 条结果里。评测集的构建方式是从知识库语料里挑出 100300 个真实问题每个问题标注 13 个正确答案对应的chunk_id。5.2 一个 70 行的离线评测脚本import json def evaluate(qa_file: str, search_func, top_k: int 5): with open(qa_file, r, encodingutf-8) as f: cases json.load(f) # [{question: ..., gold_ids: [12, 34]}] mrr_sum, recall_sum 0.0, 0.0 for case in cases: results search_func(case[question], top_ktop_k) retrieved_ids [r[id] for r in results] gold set(case[gold_ids]) # 计算 MRR第一个命中 gold 的位置取倒数 rank None for i, rid in enumerate(retrieved_ids): if rid in gold: rank i 1 break if rank: mrr_sum 1.0 / rank # 计算 Recall5命中数 / 总标注数 hit_count len(gold.intersection(retrieved_ids)) recall_sum hit_count / len(gold) n len(cases) print(fMRR{top_k}: {mrr_sum / n:.4f}) print(fRecall{top_k}: {recall_sum / n:.4f})逻辑说明search_func传入的是整个混合检索链路的入口函数这样评测脚本可以复用于任意版本的回溯逻辑。gold_ids是人工标注的正确答案 id和语料 JSON 里预先生成的 id 保持一致。MRR 只看第一个正确答案出现的位置Recall5 则更宽容——只要出现在前 5 就算命中。这两个指标要一起看MRR 低而 Recall5 高说明答案被召回但排序不对优先调 RRF 权重两者都低说明召回本身就有问题该扩检索范围或加向量路权重。5.3 从三种失败里定位问题召回空、排序错、知识库缺评测结果不好时先看失败案例属于哪一类。第一类“召回空”表现为 MRR 和 Recall 双双为 0多半是查询里的核心词没进词典或在语料里不存在处理方式是先把 query token 打印出来确认分词是否合理比如“MFT”被切成了“MF”和“T”。第二类“排序错”表现为 Recall5 不错但 MRR 很低说明答案召回但排到了后面优先检查是不是长文档因b0.75被过度惩罚。第三类是知识库本身缺内容问题里的实体在语料里压根没有这一类任何检索算法都救不了只能回到第 2 章补语料。日常优化时把每轮的失败 case 记到查询日志里每周过一遍比反复调参效率高得多。本文还有配套的精品资源点击获取