企业官网建设流程全解析

简介：这套资源是一份基于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 字的块再按句号、分号切成子句组，保证单个检索单元在 200～600 字之间。

存储上不用引入重型组件，几万条语料用 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(encoding="utf-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", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False) 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 的默认参数k1=1.5, b=0.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", encoding="utf-8") as f: corpus = json.load(f) tokenized_corpus = [doc["tokens"] for doc in corpus] bm25 = BM25Okapi(tokenized_corpus, k1=1.5, b=0.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)), key=lambda i: scores[i], reverse=True)[: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控制文档长度归一化强度，b=0.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_embeddings=True, batch_size=32) embeddings = np.asarray(embeddings, dtype="float32") 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_embeddings=True) scores, indices = index.search(np.asarray(q_vec, dtype="float32"), top_k) return [{"score": float(scores[0][i]), "text": corpus[indices[0][i]]["text"]} for i in range(top_k)]

逻辑说明：normalize_embeddings=True和IndexFlatIP是配套的，向量归一化后内积就是余弦相似度，省去单独算模长的步骤。bge-small-zh-v1.5是国内常见的中文轻量句向量模型，对电厂规程类文本的泛化表现稳定，更重要的是它支持中文长句和相似度重排。模型编码时的batch_size按显存调，纯 CPU 环境 16 或 32 都合适。Faiss 索引保存到磁盘后，服务启动时一次性加载，不需要每次重启都重新编码全量语料。向量维度取决于模型，bge-small-zh-v1.5是 512 维，对几万条语料建索引耗时在秒级，性能完全不是瓶颈。

3.4 混合召回与 RRF 融合排序参数选择

BM25 和向量检索各有偏好，直接拼接结果会有重复，而且两种分数不在一个量纲上，不能直接相加。业界常用的融合方式是 RRF（Reciprocal Rank Fusion），它不看原始分数，只看每条结果在各自召回列表里的排名位置。公式是把排名倒数累加，常用k=60。

def rrf_fusion(bm25_results, vector_results, k=60, top_k=5): score_map = {} def add_results(results, weight=1.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, weight=1.0) add_results(vector_results, weight=1.0) merged = sorted(score_map.items(), key=lambda x: x[1], reverse=True)[:top_k] return [{"score": round(s, 4), "text": t} for t, s in merged]

逻辑说明：weight参数可以用来调节两路召回的信任度，实测在电厂问答场景里 BM25 和向量等权通常就够；如果评测发现语义近义召回不足，再把向量路权重加到 1.2～1.5。rank >= 20的截断基于一个观察：RRF 里排在第 20 名之后的结果，对最终分数的贡献不到 0.02，参与融合只会引入噪声。融合后的score只用于排序，不用于置信度展示——真正的置信度应该由最后一步的答案相关性判定给出。

这段融合逻辑可以加一个分支：当 BM25 最高分大于某个阈值（比如 30）时，直接信任词法结果，不启动向量召回。这样做的好处是省一次模型编码推理时间，对“查规程原文”这类明确字面匹配的查询尤其划算。

4. 把检索能力包装成对话系统：意图、状态与交互循环

4.1 对话系统入口处的意图识别：三类问题走三条路

检索式问答系统的“对话”二字，不是指闲聊，而是指系统能理解用户问题的类型、记住上下文状态、在信息不足时主动追问。火力发电厂知识问答场景里，用户问题大致分三类：查规程原文、问处理步骤、问设备参数。这三类问题对答案形态的期望不同——查原文要给出处，问步骤要按顺序组织，查参数要给出带单位的数值。所以在进入检索前，先用轻量规则做意图分类。

规则分类不用机器学习，一组关键词表加正则就够。比如包含“如何、怎么、步骤、操作”倾向步骤类，包含“多少、值、定值、范围”倾向参数类。分类结果用来裁剪检索范围——查到第 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_k=10) vector_results = search_vector(query, top_k=10) 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_k=3) # 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、Recall@5 与日志反馈

5.1 用 MRR 和 Recall@5 量化“这次改版有没有变好”

检索式问答的评价不能靠“感觉回答变好了”，要落到可量化的指标上。工程上最常用两个：MRR（Mean Reciprocal Rank）衡量正确答案排在结果第几，Recall@5 衡量正确答案是否出现在前 5 条结果里。评测集的构建方式是从知识库语料里挑出 100～300 个真实问题，每个问题标注 1～3 个正确答案对应的chunk_id。

5.2 一个 70 行的离线评测脚本

import json def evaluate(qa_file: str, search_func, top_k: int = 5): with open(qa_file, "r", encoding="utf-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_k=top_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 # 计算 Recall@5：命中数 / 总标注数 hit_count = len(gold.intersection(retrieved_ids)) recall_sum += hit_count / len(gold) n = len(cases) print(f"MRR@{top_k}: {mrr_sum / n:.4f}") print(f"Recall@{top_k}: {recall_sum / n:.4f}")

逻辑说明：search_func传入的是整个混合检索链路的入口函数，这样评测脚本可以复用于任意版本的回溯逻辑。gold_ids是人工标注的正确答案 id，和语料 JSON 里预先生成的 id 保持一致。MRR 只看第一个正确答案出现的位置，Recall@5 则更宽容——只要出现在前 5 就算命中。这两个指标要一起看：MRR 低而 Recall@5 高，说明答案被召回但排序不对，优先调 RRF 权重；两者都低，说明召回本身就有问题，该扩检索范围或加向量路权重。

5.3 从三种失败里定位问题：召回空、排序错、知识库缺

评测结果不好时，先看失败案例属于哪一类。第一类“召回空”，表现为 MRR 和 Recall 双双为 0，多半是查询里的核心词没进词典或在语料里不存在，处理方式是先把 query token 打印出来，确认分词是否合理，比如“MFT”被切成了“MF”和“T”。第二类“排序错”，表现为 Recall@5 不错但 MRR 很低，说明答案召回但排到了后面，优先检查是不是长文档因b=0.75被过度惩罚。第三类是知识库本身缺内容，问题里的实体在语料里压根没有，这一类任何检索算法都救不了，只能回到第 2 章补语料。日常优化时把每轮的失败 case 记到查询日志里，每周过一遍，比反复调参效率高得多。

本文还有配套的精品资源，点击获取