
简介这是一套面向计算机相关专业学生与开发者的知识图谱实战项目以心血管疾病领域为切入点构建可交互的智能问答系统适合用作毕业设计、课程设计或知识图谱入门进阶练习。资源包共197个文件约5.52MB其中84个json与84个csv构成疾病结构化数据与图谱实体关系素材15个txt与1个md提供说明文档6个py脚本负责图谱构建与问答逻辑另有少量png、gif及js用于界面与展示。内容覆盖心房颤动、冠心病、心力衰竭、心肌梗死、心律失常等多种心血管疾病条目可帮助读者理解从数据整理、图谱建模到问答推理的完整链路。目前已有58人学习关注。下载后建议先阅读README结合源码与数据集快速跑通流程也可在此基础上扩展疾病范围或改造问答策略用于个人学习与项目演示。1. 从十张 CSV 到能对话的问答系统这套心血管知识图谱资源到底能跑出什么拿到一个「基于知识图谱的心血管疾病问答系统」的源码包很多人第一反应是打开看有没有界面、能不能直接问答。但真正决定这套东西能不能用、值不值得拆的是数据层——这个包里给了十张按疾病拆分的 CSV心房颤动、冠心病、心力衰竭、心肌梗死、心肌炎、心律失常、心绞痛、心内膜炎、肺心病、短暂性脑缺血发作。这十类基本覆盖了心血管内科最常见的病种意味着图谱的实体和关系不是凭空造的而是有真实结构化数据打底。对做课程设计、毕设或者想入门知识图谱问答的 Python 开发者来说这套资源的定位很清晰它把「数据 → 图谱 → 问答」这条链路完整走了一遍你拿到手能直接跑也能顺着改。适合谁计算机相关专业的学生、需要交作业或立项演示的人以及想找一个带真实医疗数据练手 Neo4j 和问答逻辑的工程师。不适合指望它直接上生产的人——它是学习样本不是医疗产品。2. 数据层拆解十张疾病 CSV 怎么变成图谱的节点和关系2.1 先看清 CSV 里到底有什么这十张 CSV 是按疾病名命名的每张对应一类心血管疾病。常见做法是每张表里包含疾病本身、症状、病因、检查手段、治疗药物、并发症等字段一行一条记录或者一行一个属性。你下载后第一件事不是急着跑代码而是用 pandas 把每张表读出来看一眼列名和行数确认字段结构是否一致。因为知识图谱构建最怕的就是十张表字段对不齐后面实体抽取会直接崩。import pandas as pd import os data_dir ./data # 假设 CSV 都放在 data 目录下 files [f for f in os.listdir(data_dir) if f.endswith(.csv)] for f in files: df pd.read_csv(os.path.join(data_dir, f)) print(f {f} ) print(列名:, list(df.columns)) print(行数:, len(df)) print(df.head(2)) print()这段代码的作用是批量巡检。os.listdir拿到所有 CSV 文件名pd.read_csv逐张读入打印列名、行数和前两行。重点看两件事一是列名是否统一比如有的表叫「症状」有的叫「临床表现」那就得先做字段映射二是行数分布如果某张表只有两三行说明这类疾病的数据很稀疏图谱里对应的子图会很单薄问答时命中率低。参数上没什么要调的路径改成你自己的就行。2.2 实体和关系的抽取思路知识图谱的核心是「实体—关系—实体」三元组。在这套资源里疾病名如心房颤动是核心实体症状、药物、检查项是关联实体它们之间的连线就是关系。常见做法是遍历每张 CSV 的每一行把疾病作为头实体把该行里的症状、药物等作为尾实体关系类型用列名来定。triples [] for f in files: disease f.replace(.csv, ) # 文件名即疾病名 df pd.read_csv(os.path.join(data_dir, f)) for _, row in df.iterrows(): for col in df.columns: if col 疾病: continue value str(row[col]).strip() if value and value ! nan: # 关系类型直接用列名比如症状药物 triples.append((disease, col, value)) print(f共抽取三元组: {len(triples)}) print(triples[:5])逻辑说明外层循环遍历文件文件名去掉.csv就是疾病实体名。内层用iterrows逐行遍历再对每个非疾病列取值拼成(疾病, 列名, 值)的三元组。str(row[col]).strip()是为了去掉空格value ! nan是过滤 pandas 读进来的空值。这里的关键参数是列名——列名直接决定关系类型的命名所以前面巡检时如果发现列名不统一这里就得加一层映射字典。跑完打印三元组数量一般十张表能抽出几百到上千条具体看数据密度。2.3 导入 Neo4j 的两种方式抽完三元组下一步是写进图数据库。这套资源常见搭配是 Neo4j因为它的 Cypher 查询语言对问答系统很友好。导入方式有两种一种是用 Python 的 neo4j 驱动逐条写入适合数据量小、需要灵活控制的场景另一种是先导出成 CSV 再用LOAD CSV批量导入适合数据量大、追求速度的场景。from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, 你的密码)) def insert_triple(tx, head, rel, tail): # 用 MERGE 避免重复创建节点 query ( MERGE (a:Entity {name: $head}) MERGE (b:Entity {name: $tail}) MERGE (a)-[r:REL {type: $rel}]-(b) ) tx.run(query, headhead, relrel, tailtail) with driver.session() as session: for h, r, t in triples: session.execute_write(insert_triple, h, r, t) driver.close()这段代码用MERGE而不是CREATE是因为同一实体可能在多张表里反复出现比如「心悸」既是心房颤动的症状也可能是心律失常的症状用MERGE能保证节点唯一不会建出一堆重复节点。auth里的密码换成你本地 Neo4j 的密码默认端口 7687。execute_write是事务写入数据量大时可以改成批量提交比如每 500 条提交一次否则逐条写会慢得让你怀疑人生。跑完之后去 Neo4j Browser 里执行MATCH (n) RETURN count(n)确认节点数再MATCH ()-[r]-() RETURN count(r)看关系数两个数对得上说明导入没丢数据。3. 问答逻辑实现从自然语言问题到 Cypher 查询3.1 问题意图识别的基本套路问答系统要做的第一件事是听懂用户在问什么。用户输入「心房颤动有哪些症状」系统得识别出实体是「心房颤动」问的是「症状」这个关系。常见做法是先用分词加实体匹配把问题里的疾病名和关系词抽出来。疾病名可以直接拿十张 CSV 的文件名做词典关系词则对应 CSV 的列名。disease_list [f.replace(.csv, ) for f in files] relation_map { 症状: [症状, 表现, 有什么症状], 药物: [药, 药物, 吃什么药, 用药], 检查: [检查, 怎么查, 诊断方法], 并发症: [并发症, 会引起什么] } def parse_question(q): found_disease None for d in disease_list: if d in q: found_disease d break found_rel None for rel, keywords in relation_map.items(): if any(kw in q for kw in keywords): found_rel rel break return found_disease, found_rel逻辑说明disease_list从文件名生成是最直接的实体词典。relation_map把用户可能说的各种表达映射到标准关系名上比如「吃什么药」映射到「药物」。parse_question先匹配疾病再匹配关系返回两个值。参数上relation_map的覆盖度直接决定问答的召回率你可以根据实际测试不断往里加同义词。这个方案简单粗暴但有效适合入门如果要做更鲁棒的意图识别可以换成基于 BERT 的分类模型但那是另一个工作量了。3.2 把解析结果翻译成 Cypher拿到疾病和关系之后就要拼 Cypher 查询去图数据库里捞答案。这一步是整个问答系统的核心链路拼错了要么查不到要么查出无关的东西。def build_cypher(disease, relation): if not disease or not relation: return None query ( MATCH (a:Entity {name: $disease})-[r:REL {type: $relation}]-(b:Entity) RETURN b.name AS answer ) return query def answer_question(q): disease, relation parse_question(q) cypher build_cypher(disease, relation) if not cypher: return 抱歉我没听懂这个问题请换个说法。 with driver.session() as session: result session.run(cypher, diseasedisease, relationrelation) answers [record[answer] for record in result] if not answers: return f没有查到{disease}相关的{relation}信息。 return f{disease}的{relation}包括 、.join(answers)逻辑说明build_cypher用参数化查询$disease和$relation是占位符避免拼接字符串带来的注入风险。answer_question把解析、查询、结果组装串起来。如果parse_question返回空说明没识别出意图给个兜底回复。查到结果后用「、」拼接返回。参数上注意REL和type要和导入时保持一致导入用的什么标签和关系类型这里就得用什么否则查不到。跑通这条链路后你可以拿十类疾病逐个测一遍看哪些问题的回答是空的空的那条就是数据或映射的缺口。3.3 多跳查询并发症的并发症怎么查单跳查询只能回答「心房颤动有哪些并发症」但如果用户问「心房颤动的并发症又有哪些症状」就需要两跳甚至多跳查询。知识图谱的优势就在这里Cypher 可以用变长路径来表达。def multi_hop_query(disease, hops2): query ( MATCH path (a:Entity {name: $disease})-[r:REL*1.. str(hops) ]-(b:Entity) RETURN b.name AS answer, length(path) AS depth ORDER BY depth ) with driver.session() as session: result session.run(query, diseasedisease) return [(rec[answer], rec[depth]) for rec in result]逻辑说明[r:REL*1..2]表示沿 REL 关系走 1 到 2 跳length(path)返回路径长度ORDER BY depth让近的结果排前面。参数hops控制最大跳数一般设 2 到 3设太大查询会爆炸式增长返回一堆无关实体。这个功能在演示时很加分但实际用的时候要注意结果去重和排序否则用户会看到重复答案。常见做法是在返回前用 set 去重再按深度排序。4. 避坑与排查跑这套源码最容易翻车的五个地方4.1 中文 CSV 编码报错现象pd.read_csv直接抛UnicodeDecodeError提示utf-8 codec cant decode。原因Windows 下用 Excel 保存的 CSV 默认是 GBK 编码不是 UTF-8。解决读的时候显式指定encodinggbk或者先用记事本另存为 UTF-8。我一般会写个 try-except 自动切换编码省得每张表手动试。4.2 Neo4j 连接被拒现象Python 报ServiceUnavailable连不上 7687 端口。原因Neo4j 服务没启动或者密码不对或者防火墙拦了。解决先去 Neo4j Desktop 或命令行确认服务在跑再用neo4j status看状态。密码如果忘了可以重置。另外注意 Neo4j 4.x 和 5.x 的驱动 API 有差异execute_write在旧版本里叫write_transaction版本对不上也会报错。4.3 实体重复导致图谱膨胀现象导入后节点数远超预期同一个疾病出现好几个节点。原因用了CREATE而不是MERGE或者实体名前后有空格没 strip。解决导入前统一做strip()导入语句一律用MERGE。已经建重复的可以MATCH (n:Entity) WITH n.name AS name, collect(n) AS nodes WHERE size(nodes) 1找出来再合并。4.4 问题解析命中不了现象用户问「房颤吃什么药」系统识别不出疾病因为词典里只有「心房颤动」没有「房颤」。原因实体词典太死没有覆盖简称和别名。解决给每个疾病维护一个别名列表匹配时遍历别名。比如「心房颤动」对应[心房颤动, 房颤, AF]。这个工作琐碎但必要直接决定问答的可用性。4.5 Cypher 查询返回空但数据明明在现象图谱里明明有数据查询却返回空列表。原因关系类型或标签大小写不一致比如导入时用了REL查询时写了rel。Cypher 是大小写敏感的。解决先用MATCH ()-[r]-() RETURN DISTINCT type(r)和MATCH (n) RETURN DISTINCT labels(n)确认实际的标签和关系名再对照代码改。这个坑我踩过不止一次后来养成习惯导入和查询用同一份常量定义。5. 进阶玩法把问答系统接上 Web 界面和向量检索5.1 用 Flask 快速搭一个问答接口源码包如果自带界面最好没有的话用 Flask 包一层很快。核心就是把answer_question挂到一个路由上前端发 POST 请求带问题后端返回答案。from flask import Flask, request, jsonify app Flask(__name__) app.route(/ask, methods[POST]) def ask(): data request.get_json() question data.get(question, ) if not question: return jsonify({error: 问题不能为空}), 400 answer answer_question(question) return jsonify({question: question, answer: answer}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)逻辑说明/ask接收 JSON 格式的 POST 请求取question字段调answer_question拿结果再以 JSON 返回。debugTrue方便开发时热重载上线要关掉。参数上host0.0.0.0让局域网内其他机器也能访问只在本机测的话用127.0.0.1就行。前端随便写个输入框加按钮fetch 调这个接口就能对话了。5.2 引入向量检索补足图谱覆盖不到的问答纯图谱问答的短板是用户问法一旦超出预设的关系词就答不上来。常见做法是加一层向量检索做兜底——把 CSV 里的文本内容向量化存进 FAISS 或 Chroma图谱查不到时走向量库做语义匹配。方案适用场景优点缺点纯 Cypher 查询问题意图明确、关系词命中精确、可解释覆盖窄换个说法就挂向量检索兜底开放性问题、模糊表达召回广可能答非所问图谱向量混合实际问答系统兼顾精确和召回实现复杂度高我一般会先跑图谱查询返回空的时候再走向量检索把两边结果合并后按置信度排序。这样既保留了图谱的精确性又不会因为用户换了个说法就哑火。向量模型可以用text2vec或者sentence-transformers的中文模型具体选哪个看你的硬件和延迟要求。5.3 验证系统好不好用的三个土办法第一拿十类疾病各造五个问题覆盖症状、药物、检查、并发症、多跳一共五十个问题跑一遍统计命中率。第二故意用简称和口语化表达测比如「心梗」「心脏乱跳」看解析层扛不扛得住。第三把同一个问题换三种问法看答案是否一致不一致说明意图识别不稳定。这三个办法不需要什么工具但能快速暴露系统短板。从那以后我每次拿到问答类项目都先跑一遍这套土办法再谈优化省得在演示时翻车。希望帮到你。本文还有配套的精品资源点击获取