简介：这份资源是面向计算机、人工智能及相关专业学生与开发者的知识图谱实战项目，以心血管疾病领域为切入点，构建可交互的智能问答系统，适合作为毕业设计、课程设计、项目立项演示或知识图谱入门进阶的学习案例。压缩包共197个文件，约5.52MB，其中84个json与84个csv构成疾病实体、关系及属性数据，15个txt与1个md提供说明文档，6个py脚本负责图谱构建与问答逻辑，另有png、gif等图片辅助展示。内容覆盖心房颤动、冠心病、心力衰竭、心肌梗死、心律失常等常见心血管疾病的语料整理，读者可借此掌握从数据清洗、图谱建模到问答推理的完整链路，并在此基础上修改扩展其他病种或功能。目前已有58人学习关注，代码经测试可运行，下载后建议先阅读README说明，仅供学习参考，切勿用于商业用途。

1. 从一份心血管问答源码说起：知识图谱到底解决了什么

心血管疾病是门诊里最高频的主诉方向之一，患者问来问去无非那几类：这个药能不能和他汀一起吃、放了支架以后多久复查、血压降到多少算达标。如果只靠关键词匹配做问答，用户换个说法系统就答不上来；如果只靠大模型硬答，遇到具体药物相互作用又容易一本正经地胡说。这就是「基于知识图谱的心血管疾病问答系统」要解决的核心矛盾——把散落在指南、药品说明书、临床路径里的实体和关系抽出来，存进图数据库，再用自然语言问句去图里精确检索，最后组织成人能看懂的回答。它适合两类人：一类是正在做课程设计或毕业设计、需要一套能跑通全流程的 Python 项目源码和数据集的同学；另一类是已经会写 Python、想搞清楚知识图谱构建到智能问答系统落地链路的工程师。整套方案的技术栈很清晰：Python 做数据处理和 Web 服务，Neo4j 存图谱，前端做一个对话页面，数据集负责提供心血管领域的实体、关系和问答对。下面我按「先建图、再问答、最后排错」的顺序，把这条链路拆开讲。

2. 心血管知识图谱怎么建：从原始语料到 Neo4j 可查询的图

2.1 先定本体，再谈抽取，否则后面全是返工

很多人一上来就写爬虫抓指南文本，抓完发现实体类型五花八门，疾病、症状、药品、检查项混在一起，关系也没法统一。知识图谱构建的第一步永远是本体建模，也就是先想清楚这个领域里有哪些类型的节点、哪些类型的关系。心血管这个方向，我一般会收敛成六类实体和五类关系，够用且不至于失控。

关系类型对应为：疾病「有症状」症状、疾病「用药物」药物、疾病「需检查」检查、疾病「治疗方式」治疗、疾病「危险因素」危险因素。本体定完再去做实体识别和关系抽取，标注规范才不会来回改。这一步偷懒，后面图里会出现同一个药三种写法、同一个症状挂在不同疾病下语义打架的情况，清洗成本远高于前期设计。

2.2 用 Python 把结构化数据写进 Neo4j

数据集通常已经给了实体表和关系表，常见格式是 CSV 或 JSON。真正要写的是导入脚本，把行数据变成图里的节点和边。下面这段是核心逻辑，用官方 neo4j 驱动批量写入，关键是 MERGE 而不是 CREATE，避免重复导入时产生重复节点。

from neo4j import GraphDatabase import csv driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "your_password")) # 实体导入：按类型打标签，name 作为唯一键 def load_entities(tx, label, rows): query = f""" UNWIND $rows AS row MERGE (n:{label} {{name: row.name}}) SET n.desc = row.desc """ tx.run(query, rows=rows) # 关系导入：先匹配两端节点，再建边 def load_relations(tx, rel_type, rows): query = f""" UNWIND $rows AS row MATCH (a {{name: row.head}}) MATCH (b {{name: row.tail}}) MERGE (a)-[:{rel_type}]->(b) """ tx.run(query, rows=rows) with driver.session() as session: with open("disease.csv", encoding="utf-8") as f: session.execute_write(load_entities, "Disease", list(csv.DictReader(f))) with open("drug.csv", encoding="utf-8") as f: session.execute_write(load_entities, "Drug", list(csv.DictReader(f))) with open("disease_drug.csv", encoding="utf-8") as f: session.execute_write(load_relations, "用药物", list(csv.DictReader(f)))

逻辑说明：UNWIND把一批行展开成逐行处理，MERGE保证节点或边已存在时不重复创建，SET用来补属性。参数上，label和rel_type是拼进 Cypher 的，所以只能来自你代码里写死的白名单，绝不能直接接用户输入，否则就是注入漏洞。批量大小建议一次 500 到 1000 行，太大内存吃紧，太小网络往返次数多。导入完用一条查询验证：MATCH (n) RETURN labels(n), count(*)，看每类节点数量是否和源数据对得上。

2.3 图谱质量自检：三个必须跑的校验查询

图建完不代表能用，得先自检。第一，查孤立节点：MATCH (n) WHERE NOT (n)--() RETURN n.name，孤立节点在问答里永远检索不到，要么补关系要么删。第二，查重复实体：MATCH (n) WITH n.name AS name, count(*) AS c WHERE c > 1 RETURN name, c，同名多节点会让答案分裂。第三，查关系方向是否一致，比如「用药物」应该是疾病指向药物，如果出现反向边，问答模板匹配就会错。这三条查询花不了几分钟，但能挡掉后面一大半玄学 bug。

3. 问答系统怎么把自然语言变成 Cypher：意图识别与模板匹配

3.1 意图分类决定走哪条查询路径

用户输入「高血压吃什么药」和「高血压有哪些症状」，字面结构几乎一样，但要走完全不同的查询。所以问答系统的第一层是意图识别，把问句归到预定义的几类意图上，比如「疾病-药物」「疾病-症状」「疾病-检查」「药物-禁忌」。数据集里一般会带问答对，可以直接拿来训练一个文本分类器。轻量做法是用 scikit-learn 的 TF-IDF 加朴素贝叶斯，够快够稳；数据量大再上 BERT 微调。下面是最小可用的分类流程。

from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.naive_bayes import MultinomialNB from sklearn.pipeline import make_pipeline # questions 是问句列表，labels 是对应意图标签 model = make_pipeline(TfidfVectorizer(analyzer="char", ngram_range=(1, 2)), MultinomialNB()) model.fit(questions, labels) def predict_intent(text): return model.predict([text])[0]

逻辑说明：中文用char级别的 n-gram 比按词切更省事，不用额外装分词器，ngram_range=(1,2)兼顾单字和双字组合。参数上，如果意图类别不均衡，MultinomialNB可以换ComplementNB，对小类别更友好。分类置信度低于阈值时，不要硬答，直接回「没太理解，你可以换个说法」，比答错强。

3.2 实体链接：把「降压药」对上图里的「阿司匹林」

意图识别完，还要从问句里抽出实体，并映射到图里真实存在的节点名。用户说「降压药」，图里存的是具体药名，这中间需要一层实体链接。常见做法是维护一个别名词典，把口语说法映射到标准名，再用字符串相似度兜底。下面这段演示用编辑距离做模糊匹配。

from difflib import SequenceMatcher def link_entity(mention, candidates, threshold=0.6): best, score = None, 0 for c in candidates: s = SequenceMatcher(None, mention, c).ratio() if s > score: best, score = c, s return best if score >= threshold else None

逻辑说明：candidates是图里某一类节点的全部名称，threshold控制匹配松紧。设太高会漏匹配，设太低会把「阿司匹林」错配到「阿托伐他汀」。我的经验是 0.6 起步，再根据实际问句调。更稳的做法是别名词典优先，模糊匹配只做兜底，因为药品名差一个字就是两种药，容错空间很小。

3.3 模板生成 Cypher 并组织答案

意图和实体都齐了，就可以套查询模板。每个意图对应一段 Cypher，把实体名填进去执行，拿回结果再拼成自然语言。下面是「疾病-药物」意图的模板。

TEMPLATES = { "disease_drug": """ MATCH (d:Disease {{name: $name}})-[:用药物]->(drug:Drug) RETURN drug.name AS answer """ } def answer(intent, entity, session): cypher = TEMPLATES.get(intent) if not cypher: return "暂不支持这类问题" records = session.run(cypher, name=entity) names = [r["answer"] for r in records] return "、".join(names) if names else "图谱中没有查到相关信息"

逻辑说明：Cypher 里用$name参数化传值，这是防注入的关键，别用字符串拼接。返回结果为空时给明确提示，不要返回空字符串让前端显示成白屏。多个答案用顿号连接，前端再按需渲染成列表。到这里，一条从问句到答案的链路就通了，剩下的就是把它包成 Web 服务。

4. 把问答系统跑起来：Flask 接口与前端对话页

4.1 用 Flask 暴露一个 /ask 接口

后端不需要多复杂，一个接收 POST 的接口就够。请求体带用户问句，返回意图、实体和答案。下面是最小实现。

from flask import Flask, request, jsonify from neo4j import GraphDatabase app = Flask(__name__) driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "your_password")) @app.route("/ask", methods=["POST"]) def ask(): text = request.json.get("question", "").strip() if not text: return jsonify({"error": "问题不能为空"}), 400 intent = predict_intent(text) entity = link_entity(extract_mention(text), all_disease_names) with driver.session() as session: ans = answer(intent, entity, session) return jsonify({"intent": intent, "entity": entity, "answer": ans}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000)

逻辑说明：request.json要求前端发Content-Type: application/json，否则取不到值。host="0.0.0.0"是为了容器或局域网访问，本地调试用127.0.0.1更安全。生产环境别用 Flask 自带服务器，换 gunicorn 起多 worker。接口返回结构里带上意图和实体，方便前端调试和用户纠错。

4.2 前端对话页与联调要点

前端一个输入框加一个消息列表就能用，关键是处理加载态和错误态。用户点发送后先显示「正在查询」，拿到结果再替换。跨域问题在开发阶段最常见，Flask 侧装 flask-cors 加一行CORS(app)即可。联调时先拿 curl 测接口，确认后端通了再调前端，能省掉大量「到底是前端还是后端」的扯皮。

curl -X POST http://127.0.0.1:5000/ask \ -H "Content-Type: application/json" \ -d '{"question": "高血压吃什么药"}'

这条命令返回的 JSON 里如果 answer 有内容，说明整条链路是通的。如果返回空，先查意图识别对不对，再查实体链接有没有对上，最后查 Cypher 在 Neo4j Browser 里手动跑一遍有没有结果，逐段定位。

5. 避坑与排查：这套系统最容易翻车的五个地方

5.1 现象：导入后节点数量翻倍。原因：用了 CREATE 而不是 MERGE，或者重复执行了导入脚本。解决：把建节点语句统一改成 MERGE，导入前先MATCH (n) DETACH DELETE n清空，或者给节点加唯一约束CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.name IS UNIQUE，从数据库层面挡住重复。

5.2 现象：问句明明有对应数据，系统却答「没查到」。原因：实体链接没对上，用户说法和图里标准名不一致。解决：先打印出抽取到的 mention 和链接结果，确认是抽取错了还是匹配阈值太高，补别名词典比调阈值更可靠。

5.3 现象：意图分类把「高血压吃什么药」判成「疾病-症状」。原因：训练数据里两类问句句式太像，特征区分度不够。解决：补充易混样本，或者把 char n-gram 范围扩到 (1,3)，再不行就上预训练模型。别指望调分类器参数能解决数据本身的问题。

5.4 现象：接口返回 500，日志报 Cypher 语法错误。原因：模板里的标签或关系类型是动态拼的，或者参数名写错。解决：Cypher 里所有变量值走$param，标签和关系类型只从白名单取，改完在 Neo4j Browser 里先验证再上线。

5.5 现象：前端一直转圈拿不到结果。原因：跨域被拦，或者后端没起在预期端口。解决：浏览器 F12 看 Network 里请求状态，CORS 问题加 flask-cors，端口问题核对启动命令和前端请求地址是否一致。

6. 让答案更可信：给问答结果加上来源与置信度

到这一步系统能答了，但医疗方向的问答，光有答案不够，用户会问「你凭什么这么说」。进阶做法是给每个答案附上来源和置信度。来源可以存在关系属性里，比如疾病和药物之间的边上加一个source字段记录指南名称，查询时一并返回。置信度则来自意图分类的概率和实体链接的相似度分数，两者相乘做一个粗略打分，低于阈值的答案标注「仅供参考」。

def answer_with_confidence(intent, entity, session, intent_prob, link_score): cypher = """ MATCH (d:Disease {name: $name})-[r:用药物]->(drug:Drug) RETURN drug.name AS answer, r.source AS source """ records = session.run(cypher, name=entity) results = [{"answer": r["answer"], "source": r["source"]} for r in records] confidence = round(intent_prob * link_score, 2) return {"results": results, "confidence": confidence}

逻辑说明：intent_prob从分类器predict_proba取最大值，link_score是实体链接的相似度，两者相乘只是工程上的粗略估计，不是严格概率，但足够用来做「高置信直接答、低置信加提示」的分流。source字段需要在建图时就从数据里带进来，事后补很麻烦，所以导入阶段就要规划好属性。

验证这套机制是否有效，我的习惯是准备二十条真实问句做回归测试，覆盖每个意图和几类边界情况，每次改完代码跑一遍，看答案和置信度有没有异常波动。这个测试集不用大，但要稳定，它是我改代码时的后悔药。踩过的坑告诉我，医疗类问答最怕的不是答不出，而是答得理直气壮却是错的，所以宁可让置信度低一点、提示多一点，也别让系统显得无所不知。希望帮到你。

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