简介一套基于故障诊断知识图谱的问答系统完整源码与说明文档面向设备故障智能问答领域融合知识图谱与自然语言处理技术适用于机械工程、电子科技等专业学生的毕业设计或实践也可作为设备维护技术人员提升故障排查效率的参考。压缩包共75个文件约1.57MB主要包含Python源码、JavaScript与HTML前端页面、CSS样式、CSV数据文件以及字体图标等静态资源其中Python代码实现问句解析、实体识别、图谱查询与答案生成等核心逻辑CSV数据用于构建设备故障知识图谱前端页面则提供可视化交互问答界面。内容覆盖知识图谱建模、语义匹配、答案检索的完整流程并附带测试脚本和readme说明帮助读者理清从数据预处理到问答响应的实现链路。已有72人浏览学习适合作为故障诊断智能化学习或毕设选题的参考资料。1. 故障诊断知识图谱问答系统从跑通代码到跑通业务中间隔着什么如果只是想把毕业设计演示跑起来这套源码解压后改几条路径就能看到问答界面。但工业场景里故障诊断问答的难点从来不在界面而在“电机异响怎么回事”这种口语化问句怎么拆成知识图谱里的实体和关系再翻译成一条 Cypher 查询。这套以故障诊断为主轴的知识图谱问答系统源码把整条链路做通了本体建模、语料清洗、Neo4j 导入、意图识别、实体抽取、查询生成。它适合两类人正在做智能问答系统毕业设计的学生以及想把维修工单整理成知识库的工程师。你拿到的不是算法孤岛而是一个能跑通的“自然语言→知识图谱→答案”闭环。2. 故障知识图谱构建本体、清洗与 Neo4j 导入的完整落地2.1 本体设计五种实体、六类关系为什么这样切故障诊断知识图谱的第一步不是写代码而是定本体。很多初版图把所有信息塞在“故障”一个点上结果是每个节点挂几十个属性问答模板根本写不动。这套源码包的本体取了五种实体设备Device、故障模式Fault、症状Symptom、原因Cause、对策Measure。设计逻辑很直接用户问“电机噪声异常怎么办”这句话里就包含了设备“电机”、故障模式“噪声异常”、意图“怎么办”。如果缺少设备节点就分不清故障发生在电机还是齿轮箱如果缺少症状节点就没法用“出现什么现象”这种柔性方式做检索。实体类型、标签和主要属性如下表实体类型标签主要属性设备Devicename, code, location故障模式Faultname, severity, frequency症状Symptomname, measurement_point, threshold原因Causename, mechanism, remark对策Measurename, steps, responsible关系方面源码包里默认建了六类关系设备到故障模式用OCCURS故障模式到症状用SHOWS故障模式到原因用CAUSED_BY故障模式到对策用HANDLED_BY原因到故障模式用LEAD_TO故障模式之间用SIMILAR_TO。关系不是越多越好。像“设备-安装于-位置”这种关系如果问答模板里根本不涉及就不要塞进第一版。每增加一类关系清洗和导入逻辑都要跟着动问答模板也要维护否则后面一定会翻车。2.2 故障语料清洗把工单表格拆成标准三元组故障诊断数据大多来自设备维修记录、轴承故障统计表、专家经验文档。这些数据有一个共性字段不规整。故障描述列里可能同时出现中文逗号、英文逗号、顿号、分号同一行里写了好几个故障。清洗阶段必须做三件事字段规整、多值拆分、术语标准化。源码包里build/目录下的清洗脚本核心逻辑是这样一段import pandas as pd import re from collections import OrderedDict # 读取原始维修工单dtypestr 防止“设备编号”这种列被读成数字 df pd.read_excel(data/repair_orders.xlsx, dtypestr) rows [] for _, record in df.iterrows(): device record[设备名称].strip() # 故障描述里中英文标点混用的情况非常常见用正则统一拆分 fault_list re.split(r[,、;\n], record[故障描述] or ) symptom record[故障现象] or for fault in fault_list: if not fault: continue rows.append(OrderedDict( devicedevice, faultfault.strip(), symptomsymptom, )) out pd.DataFrame(rows).drop_duplicates(subset[device, fault, symptom]) out.to_csv(build/import_nodes.csv, indexFalse, encodingutf-8-sig)这段代码里三个关键点。第一dtypestr防止设备编号被 Excel 自动转成科学计数法。第二re.split里的字符集覆盖了中英文逗号、顿号、分号和换行实际工单里还可能混入全角空格所以后面加了strip()。第三utf-8-sig导出是为了让 CSV 被 Neo4j 读取时不带 BOM也方便后续用 Excel 打开检查。接下来要把症状从一列里拆开生成关系文件。常见做法是用explode把多值字段展开成一行一条关系# 把“设备-故障-症状”展开成一行一条关系 rels ( out.assign(symptomout[symptom].str.split(r[,;])) .explode(symptom) ) rels[symptom] rels[symptom].str.strip() rels rels.dropna(subset[symptom]).query(symptom ! ) rels[[device, fault, symptom]].to_csv( build/import_rels.csv, indexFalse, header[device, fault, symptom], encodingutf-8-sig, )assign先把症状列按标点打散成列表explode把一行变成多行。如果一行工单里有三个故障、两个症状这一步会产生六条关系记录。这里不做去重是刻意的后面导入 Neo4j 时用MERGE去重比在清洗阶段去掉更安全因为 CSV 里的重复可能来自不同工单合并后反而保留了完整的语义。如果语料里出现了“由于润滑不足导致轴承磨损”这种因果句想抽 Cause 节点不要急着上实体关系抽取模型。毕设数据量几百条时用因果句式正则更划算。源码包里维护了一组由于(.?)导致(.?)这类句式抽不准的情况比预想少得多。2.3 Neo4j 批量导入约束、索引、LOAD CSV为什么用 Neo4j 而不是 MySQL故障问答的高频查询是“设备→故障模式→原因”“故障模式→对策”这是典型多跳关系遍历。MySQL 做两跳就要写多个 JOIN三跳以后 SQL 基本不可维护。图数据库里直接匹配路径就行。导入前先建约束和索引。源码包kg_build/下的导入脚本开头是这一段CREATE CONSTRAINT device_name IF NOT EXISTS FOR (n:Device) REQUIRE n.name IS UNIQUE; CREATE CONSTRAINT fault_name IF NOT EXISTS FOR (n:Fault) REQUIRE n.name IS UNIQUE; CREATE CONSTRAINT cause_name IF NOT EXISTS FOR (n:Cause) REQUIRE n.name IS UNIQUE; CREATE INDEX symptom_name_idx IF NOT EXISTS FOR (n:Symptom) ON (n.name);这里三个约束分别保证 Device、Fault、Cause 的名称唯一重复执行导入脚本不会产生重复节点。Symptom 只加普通索引不加唯一约束。原因是同一症状会出现在多台不同设备上它们应该是多个节点而不是被强行合并成一个。节点导入用LOAD CSV:auto USING PERIODIC COMMIT 500 LOAD CSV WITH HEADERS FROM file:///import_nodes.csv AS row MERGE (d:Device {name: trim(row.device)}) MERGE (f:Fault {name: trim(row.fault)}) FOREACH (s IN split(coalesce(row.symptom, ), |) | MERGE (sm:Symptom {name: trim(s)}) )MERGE不是CREATE重复导入不会生成重复节点。这里的split(coalesce(row.symptom, ), |)用管道符拆多值coalesce处理空值。实际执行时你要先把 CSV 放进 Neo4j 的 import 目录文件路径写file:///import_nodes.csv。关系导入同样走LOAD CSV:auto USING PERIODIC COMMIT 500 LOAD CSV WITH HEADERS FROM file:///import_rels.csv AS row MATCH (d:Device {name: row.device}) MATCH (f:Fault {name: row.fault}) MERGE (d)-[:OCCURS]-(f) WITH d, f, row MATCH (s:Symptom {name: row.symptom}) MERGE (f)-[:SHOWS]-(s)这里先 MATCH 节点再 MERGE 关系因为关系依赖节点存在。如果某个节点没匹配上这一行会被静默跳过所以导入完成后一定要做统计检查MATCH (n) RETURN labels(n)[0] AS entity_type, count(*) AS cnt ORDER BY entity_type;看到设备、故障、症状的数量和原始数据对得上再继续下一步。3. 问答系统核心链路意图识别、实体抽取与动态查询生成3.1 意图识别先把问题分成五类再谈其他问答系统的第一层不是实体抽取而是意图识别。因为“轴承过热是什么原因”和“轴承过热怎么办”的 Cypher 结构完全不同前者走CAUSED_BY后者走HANDLED_BY。如果意图不分就只能把所有关系查出来再猜效果会非常差。源码包里的qa/intent.py用规则关键词分类把问题分成五类询问原因、询问对策、询问症状、询问故障、寒暄。核心代码如下INTENT_RULES { ask_cause: [为什么, 什么原因, 咋回事, 怎么回事, 导致], ask_measure: [怎么办, 如何处理, 怎么维修, 措施, 解决], ask_symptom: [什么现象, 表现, 症状, 会有哪些], ask_fault: [什么故障, 有没有故障, 会坏], } def classify_intent(question): for intent, keywords in INTENT_RULES.items(): for kw in keywords: if kw in question: return intent, kw return chat, question这段代码的先后顺序就是优先级顺序。问句“为什么轴承会过热该怎么办”同时命中原因和对策默认返回先命中的ask_cause。如果业务上更希望优先给对策就把ask_measure放到循环前面。源码里这种做法是刻意保持简单的毕设问答系统的意图集合有限规则匹配完全够用而且每一条命中逻辑都能回溯调试成本远低于跑一个分类模型。3.2 实体识别与标准化用词典加最长匹配不为小语料硬上 BERT实体识别是问答系统里最容易过度设计的地方。几百条故障语料上去微调 BERT效果往往不如一个维护良好的词典因为故障知识图谱的本质是受控词表领域术语就那些变化不大。源码包走了词典加同义词映射的路线先做文本标准化再做实体匹配。先说标准化。用户说“电动机过热”图谱里的标准节点名可能叫“电机”“温度过高”中间隔着一层同义词。代码是这样处理的SYNONYMS { 电动机: 电机, 马达: 电机, 发热: 温度过高, 过热: 温度过高, 异响: 噪声异常, 转不动: 卡死, } def normalize_text(text): text text.replace( , ) for alias, std in SYNONYMS.items(): if alias in text: text text.replace(alias, std) return text这段代码要求SYNONYMS里的词按长度从长到短排序否则“电动机”会被“电机”先替换成“电机动”反而制造脏数据。实际项目中我一般直接把同义词表独立成 CSV挂在qa/synonyms.csv里新增词不用改代码。实体抽取部分源码包用的是词典匹配加最长优先def extract_entities(question, entity_dict): norm_q normalize_text(question) hits [] for entity in entity_dict: if entity and entity in norm_q: hits.append(entity) # 最长匹配优先避免“电机轴承”被拆成“电机”“轴承” hits sorted(hits, keylen, reverseTrue) return hits这里排序是关键。如果设备词典里同时有“电机”和“电机轴承”用户问“电机轴承过热怎么办”不做最长匹配就会把“电机”当成设备实体“轴承”反而找不到对应故障。实体数量少时循环足够实体超过几百个后建议换成pyahocorasick自动机匹配复杂度从 O(n) 降到 O(词长)源码包注释里也写了替换方案。3.3 查询模板把意图和实体翻译成 Cypher意图和实体都有了下一步是拼接 Cypher。这里有一个必须坚持的原则所有查询变量用参数化传参不要拼字符串。源码包qa/query_builder.py里的模板大致是CYPHER_TEMPLATES { ask_cause: MATCH (d:Device {name: $device})-[:OCCURS]-(f:Fault {name: $fault}) MATCH (f)-[:CAUSED_BY]-(c:Cause) RETURN c.name AS cause, collect(DISTINCT f.name) AS faults LIMIT $limit , ask_measure: MATCH (d:Device {name: $device})-[:OCCURS]-(f:Fault {name: $fault}) MATCH (f)-[:HANDLED_BY]-(m:Measure) RETURN m.name AS measure, m.steps AS steps LIMIT $limit , } def build_query(intent, entities, limit3): device entities.get(Device) fault entities.get(Fault) if not device or not fault: return None template CYPHER_TEMPLATES[intent] return template, {device: device, fault: fault, limit: limit}参数$device和$fault直接绑定既能防止 Cypher 注入也能让 Neo4j 复用查询计划。$limit控制返回条数默认 3 条避免一次回答糊上十几条对策。这里要强调关系方向。本体设计里是“故障模式→识别出→症状”也就是(fault)-[:SHOWS]-(symptom)。但用户问“什么原因”时走的是(fault)-[:CAUSED_BY]-(cause)。方向一旦在导入阶段搞反查询模板怎么调都查不出数据这也是后面第四章里最常见的坑之一。3.4 答案包装与兜底回答不得当用户换一种问法才是最大的坑Cypher 查出来的数据是图结构用户要看自然语言。答案包装要做两件事去重排序转成中文句子。源码包里的qa/answer.py做了简化处理def format_answer(records, intent): if not records: return 知识库里暂时没有匹配到记录可以换个说法试试 if intent ask_measure: result [] for r in records: step r.get(steps) or 对照维修手册操作 result.append(f针对{r[measure]}处理方式{step}) return .join(result) # 其他意图取查询的第一个返回字段 return .join(str(r[list(r.keys())[0]]) for r in records)兜底逻辑分两层。第一层实体识别有结果但 Cypher 查不到答案。这多半是语料里没有直接关系此时不要直接报错而是用共享症状找相似故障MATCH (f:Fault {name: $fault})-[:SHOWS]-(s:Symptom)-[:SHOWS]-(other:Fault) WHERE other.name $fault RETURN other.name AS similar_fault, count(DISTINCT s) AS shared_symptoms ORDER BY shared_symptoms DESC LIMIT 3两个故障共享的症状越多说明它们越接近。这条查询在毕设场景里非常实用不需要任何模型就能给出有解释性的相似推荐。第二层兜底实体完全没有命中。这时不要去猜直接提示用户换一种说法并给出一个示例问句比如“你可以试试电机异响的原因”。实测下来这种引导比强行给答案的体验好得多。4. 避坑排查故障问答系统最常踩的五个现场4.1 现象Neo4j 服务正常Python 连接 7687 一直超时原因Neo4j 桌面版默认只监听127.0.0.1如果你把代码部署到服务器或者 Neo4j 跑在 Docker 容器里连接串还写 localhost就会超时。另一类情况是只映射了 7474 浏览器端口忘了映射 7687 Bolt 端口。解决修改 Neo4j 配置文件neo4j.conf把监听地址改成0.0.0.0重启服务。Docker 启动时-p 7474:7474 -p 7687:7687两个端口都要映射。Python 连接串用bolt://服务器IP:7687不要再用 localhost。我在源码包里第一版就栽过这个跟头后来每次写部署文档都会特别标注端口。4.2 现象CSV 导入后统计节点数为零或重复节点翻倍原因CSV 文件是在 Windows 记事本里编辑保存的带 BOM 表头LOAD CSV WITH HEADERS读到的列名变成了\ufeffdevice和脚本里的row.device对不上于是每行都被跳过。重复计数翻倍则是没有用MERGE用了CREATE同一行数据执行两次导入就生成两个节点。解决清洗导出时用utf-8-sig编码或者对已有文件执行sed -i 1s/^\xEF\xBB\xBF// import_nodes.csv导入节点时坚持用MERGE关系也用MERGE这样才能保证脚本可重复执行。导入完成后跑统计 Cypher 确认数量再往下走。4.3 现象换了个叫法就查不到“电机过热”能出答案“电动机过热”查不到原因同义词映射只写进了SYNONYMS字典但没有在问答入口调用。或者映射方向反了把“电机”映射成“电动机”而图谱里的标准节点名是“电机”。解决全项目只保留一个入口函数answer(question)入口第一行强制走normalize_text()不要在每个模块里单独归一化。同义词表的维护要遵守一条纪律图谱里已有的标准名放在右边用户口语词放在左边。提交前跑一遍回归问句把“电动机过热”“电机发热”“马达温度高”都试一遍确认都映射到同一条答案路径。4.4 现象实体抽取把“电机轴承故障”拆成“电机”加“轴承”原因设备词典里同时存在“电机”和“电机轴承”匹配时没有做区间去重。系统先匹配到“电机”再从剩余文本里找到“轴承”结果把完整部件拆成了两个独立实体后续查询既查不到Device电机轴承也查不到Fault故障。解决匹配结果先按长度降序排列再用区间去重逻辑过滤重叠部分。简化代码如下def dedup_entities(hits, question): selected [] for ent in sorted(hits, keylen, reverseTrue): start question.find(ent) if any(s start len(ent) and start s len(e) for s, e in selected): continue selected.append((start, ent)) return [ent for _, ent in selected]这个函数的核心是短实体命中的区间如果和长实体重叠就丢弃短实体。实际操作中我还把“电机轴承”这类复合部件单独加进设备词典并把匹配顺序从普通词典匹配改成了先匹配复合词。4.5 现象回答耗时从几十毫秒突然涨到两秒最后直接超时原因故障语料小的时候Neo4j 全库扫描也没问题节点到五千以上就开始暴露。查询条件里的Device.name、Fault.name如果没有约束或索引Neo4j 会逐节点扫描标签下所有节点。另一个原因是代码里用字符串拼接的方式生成查询导致 Cypher 查询计划无法缓存。解决把上一章的CREATE CONSTRAINT和CREATE INDEX全部执行一遍并给查询加上参数绑定。遇到慢查询用PROFILE看执行计划重点关注有没有出现NodeIndexSeek还是NodeByLabelScan。如果是后者就是索引没生效。从那以后我每次上线问答系统前都会强制检查一遍索引、参数化和超时配置确认这三个点没问题再让查询接真实数据。5. 进阶落地验证集、HTTP 接口与新知识接入5.1 建一个三十条的离线验证集问答系统最怕东改一个词、西改一个模板改完只能手工点几句话验证。我建议在qa/test_cases.csv里维护三十条问句覆盖四类意图和典型同义词。表格结构很简单问句、预期意图、预期设备实体、预期故障实体。每次改完词典或模板跑一遍脚本统计意图准确率和实体准确率。实体准确率按完全匹配才算对别用模糊匹配放水。5.2 用 FastAPI 把问答封装成 HTTP 接口毕设答辩时总不能每次都在命令行里敲 Python。源码包提供了一个 FastAPI 接口封装核心代码如下from fastapi import FastAPI from pydantic import BaseModel app FastAPI() qa QASystem() class QARequest(BaseModel): question: str limit: int 3 app.post(/qa) def answer(req: QARequest): return {answer: qa.answer(req.question, req.limit)}这个接口返回 JSON前端可以拿它对接对话框。limit字段让调用方控制答案条数默认 3 条避免一次返回太多。5.3 新知识接入的三步法真实故障场景里每个月都会有新故障类型。每次新增知识我都强制走三步先在SYNONYMS表里加口语词再把新故障和原因、对策的关系写入图谱最后在验证集里加两条问句。三步缺一个后续问答就会在某一天突然翻车。从那以后我每次交付问答系统都先把这三十条验证集跑一遍确认四类意图的边界用例全部通过再放出去给业务方用。希望帮到你。本文还有配套的精品资源点击获取