医疗知识图谱对话系统（基于 Neo4j + Python 实现）
本文展示我在专业综合实践课程中的第一次作业完整项目实现流程，包括知识图谱构建、Neo4j 操作、数据导入、文本到 Cypher 语句解析，以及命令行对话系统开发。
任务目标
本任务要求基于 Neo4j 图数据库 构建一个「简易医疗知识图谱」，并在此基础上实现一个能理解自然语言、自动生成 Cypher 查询语句并回答问题的 图谱问答系统。
一、整体设计思路
- 知识构建层
使用disease1.csv医疗数据集，通过 Python 脚本读取并导入到 Neo4j 中，自动创建 疾病（Disease）、症状（symptom）、药物（drug）、治疗（treatment）、科室（department） 等节点及其关系（HAS_SYMPTOM、HAS_Drug、IS_OF_DEPARTMENT 等）。 - 知识存储层
利用 Neo4j 的图结构特性进行节点存储与关系建模。Neo4j 提供可视化界面，支持快速查询与可视化展示。 - 语义解析层
编写 Python 程序解析自然语言问题，通过关键词匹配和模糊清理规则，将文本自动转换为 Cypher 查询语句。 - 交互层（对话系统）
用户在命令行输入问题，系统根据语义生成查询语句，访问 Neo4j 数据库，返回匹配结果并打印输出。系统支持上下文承接，可连续对话。
系统架构图
自然语言输入 → 文本解析 → Cypher 查询生成 → Neo4j 查询 → 答案输出 二、Neo4j 的安装、启动与基础操作
由于Neo4j是基于Java的图数据库，运行Neo4j需要启动JVM进程，因此必须安装JAVA SE的JDK。
1.jdk的下载安装配置
jdk下载网址：https://www.oracle.com/java/technologies/downloads/
建议下载稳定且较新版本jdk，以防止版本不兼容等后续问题，根据自己电脑环境安装适应版本，安装后需要配置环境变量。通过win+R进入运行窗口，输入sysdm.cpl，进入系统属性界面，点击高级，点击环境变量，进入配置环境。
需要新建两个环境变量：
第一个：变量名 ：JAVA_HOME
变量值：安装JDK的地址
第二个：变量名：CLASSPATH
变量值：
.;%JAVA_HOME%\lib\dt.jar;%JAVA_HOME%\lib\tools.jar;然后点击变量名：Path ，选择编辑，然后新建，分别新添两个变量值
%JAVA_HOME%\bin; %JAVA_HOME%\jre\bin;
配置结束
2.Neo4j下载、解压、配置、运行
（1）neo4j下载官方地址：https://neo4j.com/download-center/
（2）Neo4j解压
解压到自定义英文文件夹
（3）Neo4j配置
步骤与配置jdk相似，新建环境变量
变量名：NEO4J_HOME
变量值：刚才解压下载文件的地址
点击系统变量区的Path，点击编辑，然后新建
新建变量：（注意保存）
%NEO4J_HOME%\bin（4）neo4j运行
打开命令提示符，进入powershell，输入neo4j.bat console,出现start即成功。
打开网址 http://localhost:7474/ 登录用户名密码：默认均为：neo4j。首次进入后需要修改默认密码，按需修改即可，进入后界面如下。
3.基础操作
增加 虚构一个病症，药物和症状，并建立关系
CREATE (d:Disease {name: '蓝皮龙热', type: '幻想类疾病'}); CREATE (s:Symptom {name: '耳朵发光'}); CREATE (m:Medicine {name: '光抑素胶囊', category: '幻想类抑制药'}); CREATE：无条件创建图元素（节点/关系）。(d:Disease {...})：创建一个带别名d的节点，标签为Disease。{name: '蓝皮龙热', type: '幻想类疾病'}：节点属性字典（property map）。-
CREATE会重复插入，如果重复跑多次就会多出同名节点。若要“去重/幂等”，用MERGE。
MATCH (d:Disease {name: '蓝皮龙热'}), (s:Symptom {name: '耳朵发光'}) CREATE (d)-[:HAS_SYMPTOM]->(s); MATCH (d:Disease {name: '蓝皮龙热'}), (m:Medicine {name: '光抑素胶囊'}) CREATE (d)-[:TREATED_BY]->(m); 关系连接解析
MATCH (d:Disease {name:'蓝皮龙热'})：匹配已有的疾病节点，赋名为d。MATCH (s:Symptom {name:'耳朵发光'})：匹配已有的症状节点，赋名为s。CREATE (d)-[:HAS_SYMPTOM]->(s)：在d到s之间创建一条有向关系，类型为HAS_SYMPTOM- 语法
(起点)-[:关系类型 {关系属性…}]->(终点)。箭头方向决定“谁指向谁”。查询时方向也会影响匹配。
查找
MATCH (d:Disease {name: '蓝皮龙热'})-[r]->(x) RETURN d, r, x; MATCH (d)-[r]->(x)：匹配从d出发、任意类型（没写类型名）且任意方向为出边（箭头->）的关系r，到达任意节点x。
RETURN d, r, x：图形结果面板会显示一个“星型”结构，中心是 d，连到 s 和 m。
修改
MATCH (m:Medicine {name: '光抑素胶囊'}) SET m.effect = '可抑制耳部光能暴走，缓解蓝皮龙热症状' RETURN m; MATCH找到药物m。SET m.effect = ...：给节点m新增或覆盖属性effect。RETURN m：方便你在结果面板确认属性改动。- SET 关键点：
SET n.prop = value：新增/覆盖某个属性。SET n += {a:1,b:2}：批量追加/覆盖多个属性（常用于参数化）。REMOVE n.prop：删除某个属性。- 若是关系属性同理：
MATCH (d)-[r:TREATED_BY]->(m) SET r.method='口服' RETURN r;
删除
MATCH (d:Disease {name: '蓝皮龙热'}) DETACH DELETE d; DETACH DELETE：删除节点 以及与其相连的所有关系。
三、neo4j-数据导入
数据源：使用医疗数据集 disease1.csv，包含字段：
name, alias, part, age, infection, insurance, department, checklist, symptom, complication, treatment, drug, period, rate, money 通过使用py2neo进行数据导入，需要预先下载py2neo包以及pandas包，在python中读入数据，连接neo4j数据库，接着使用Node()等函数初始化节点和边的对象，再使graph.merge()函数将节点和边导入到neo4j中。具体操作如下：
测试连接是否成功
#connection.py from py2neo import Graph # 连接本地 Neo4j 数据库 graph = Graph("bolt://localhost:7687", auth=("neo4j", "12345678")) # 测试查询 try: result = graph.run("RETURN '连接成功！' AS message").data() print(result[0]["message"]) except Exception as e: print("连接失败：", e) 测试结果：
核心代码：（为防止过程出错，我设定了每处理一千行数据便会输出一条结果，便于观察，非最终版）
""" #build_medical_graph.py """ import chardet with open("disease1.csv", "rb") as f: print(chardet.detect(f.read(5)))#查看原数据集编码 import pandas as pd from py2neo import Graph, Node, Relationship # 1. 连接数据库 graph = Graph("bolt://localhost:7687", auth=("neo4j", "12345678")) graph.run("MATCH (n) DETACH DELETE n") # 清空数据库 print("数据库连接成功，旧数据已清空") # 2. 读取 CSV 数据 df = pd.read_csv("disease1.csv", encoding="ANSI") print(f"成功加载 CSV，共 {len(df)} 条记录") print("CSV 表头：", list(df.columns)) # 3. 定义节点类型与对应关系 node_rel_map = { "alias": "HAS_ALIAS", "part": "IS_OF_PART", "age": "IS_OF_AGE", "infection": "IS_INFECTIOUS", "insurance": "In_Insurance", "department": "IS_OF_Department", "checklist": "HAS_Checklist", "symptom": "HAS_SYMPTOM", "complication": "HAS_Complication", "treatment": "HAS_Treatment", "drug": "HAS_Drug", "period": "Cure_Period", "rate": "Cure_Rate", "money": "NEED_Money" } # 4. 逐行创建节点与关系 for idx, row in df.iterrows(): disease_name = str(row["name"]).strip() if not disease_name: continue # 创建疾病节点 disease_node = Node("Disease", name=disease_name) graph.merge(disease_node, "Disease", "name") # 遍历各列属性 for col_name, rel_type in node_rel_map.items(): if col_name not in row: continue raw_value = str(row[col_name]).strip() if raw_value == "nan" or raw_value == "": continue # 多值字段：空格、逗号、顿号 values = [v.strip() for v in raw_value.replace("，", " ").replace(",", " ").split()] for v in values: if not v: continue target_node = Node(col_name, name=v) graph.merge(target_node, col_name, "name") rel = Relationship(disease_node, rel_type, target_node) graph.merge(rel) if (idx + 1) % 1000 == 0: print(f"已处理 {idx + 1} 条记录...") print("图谱构建完成！") # 5. 查看节点统计 stats = graph.run(""" MATCH (n) RETURN labels(n)[0] AS 类型, count(*) AS 数量 ORDER BY 数量 DESC """).data() print("\n=== 节点统计 ===") for s in stats: print(s) 数据导入后，在neo4j上会出现对应结果
注：若pycharm下载在D盘等其他盘，在c盘下载安装包后运行代码可能会出现无法找到包等报错，需要进行调试，具体步骤为：点击file-settings-add interpreter-add localinterpreter-systerm interpreter，将框中的地址改为python下载地址，点击apply，保存，即可解决。
四、文本 → Cypher 语句解析
1. 解析设计思路
自然语言解析的核心是：
“关键词识别 + 实体抽取 + 模糊匹配 + 模板生成”
系统会识别问题中包含的关键词（如“症状”“药物”“科室”等），
再将疾病名或症状名提取出来并清理多余词，如：
| 输入问题 | 清理后实体 | 生成意图 |
|---|---|---|
| 癫痫的症状有哪些？ | 癫痫 | disease_symptoms |
| 痢疾吃什么药？ | 痢疾 | disease_drugs |
| 腹泻是什么病？ | 腹泻 | symptom_diseases |
系统的文本解析模块通过“关键词识别—实体抽取—模糊匹配—模板生成”四个步骤实现自然语言到 Cypher 语句的自动转换。程序首先识别用户问题中的关键动词或名词（如“症状”“药物”“科室”等），判断问题意图；随后对输入文本进行分词与清理，去除“是什么病”“为什么”“如何治疗”等功能性词语，提取出核心实体（如疾病名或症状名）。再利用模糊匹配机制在图谱中检索最接近的节点名称，最终根据意图类型自动选择相应的 Cypher 模板生成查询语句。例如，“癫痫的症状有哪些？”识别为疾病症状查询；“痢疾吃什么药？”识别为疾病用药查询；“腹泻是什么病？”识别为症状反查疾病，从而实现了自然语言问题到图数据库查询的自动化映射。
2.代码示例
from py2neo import Graph import re # 连接 Neo4j graph = Graph("bolt://localhost:7687", auth=("neo4j", "12345678")) # 意图关键词 INTENT_KEYWORDS = { "disease_symptoms": {"症状", "表现"}, "symptom_diseases": {"是什么病", "什么病", "为什么", "原因"}, "disease_drugs": {"吃什么药", "用什么药", "药", "用药"}, "disease_department": {"挂什么科", "看什么科", "什么科", "科室"}, "disease_treatments": {"如何治疗", "怎么治疗", "治疗方法", "治疗", "怎么治", "如何治"}, "disease_complications": {"并发症"}, } # 会话状态（记忆） class SessionState: def __init__(self): self.last_disease = None self.last_symptom = None self.last_intent = None STATE = SessionState() #文本处理 def normalize(text): return text.strip().replace("？", "").replace("。", "").replace("，", "").lower() def extract_candidate_phrase(text): toks = re.findall(r"[\u4e00-\u9fa5A-Za-z0-9]+", text) return max(toks, key=len) if toks else "" # 实体清理函数 def clean_entity_phrase(phrase): p = re.sub(r"\s+", "", phrase) # 去除疑问词/功能词 for w in [ "是什么病", "什么病", "可能是什么病", "为什么", "为啥", "原因", "怎么办", "怎么回事", "怎么引起", "怎么办呢", "如何治疗", "怎么治疗", "治疗方法", "治疗", "怎么治", "如何治", "用什么药", "吃什么药", "用药", "药", "挂什么科", "什么科", "科室", "科", "的症状", "症状", "的并发症", "并发症" ]: p = p.replace(w, "") p = re.sub(r"的+$", "", p) return p.strip() def detect_intent(text): for intent, keys in INTENT_KEYWORDS.items(): if any(k in text for k in keys): return intent if "症状" in text: return "disease_symptoms" if "药" in text: return "disease_drugs" if "科" in text: return "disease_department" if "并发症" in text: return "disease_complications" if "治疗" in text or "怎么治" in text: return "disease_treatments" return "disease_symptoms" # 模糊匹配 def fuzzy_pick_disease(phrase): if not phrase: return None q1 = graph.run(""" MATCH (d:Disease) WHERE toLower(d.name) CONTAINS toLower($p) RETURN d.name AS name ORDER BY size(d.name) ASC LIMIT 3 """, p=phrase).data() if q1: return q1[0]["name"] q2 = graph.run(""" MATCH (d:Disease)-[:HAS_ALIAS]->(a:alias) WHERE toLower(a.name) CONTAINS toLower($p) RETURN d.name AS name ORDER BY size(d.name) ASC LIMIT 3 """, p=phrase).data() if q2: return q2[0]["name"] q3 = graph.run(""" MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:symptom) WHERE toLower(s.name) CONTAINS toLower($p) RETURN d.name AS name ORDER BY size(d.name) ASC LIMIT 3 """, p=phrase).data() if q3: return q3[0]["name"] return None def fuzzy_pick_symptom(phrase): if not phrase: return None q = graph.run(""" MATCH (s:symptom) WHERE toLower(s.name) CONTAINS toLower($p) RETURN s.name AS name LIMIT 3 """, p=phrase).data() return q[0]["name"] if q else None # Cypher 生成 def to_cypher(intent, disease=None, symptom=None): if intent == "disease_symptoms" and disease: return f""" MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:symptom) WHERE toLower(d.name) CONTAINS toLower('{disease}') RETURN d.name AS 疾病, collect(DISTINCT s.name) AS 症状 LIMIT 100 """ if intent == "symptom_diseases" and symptom: return f""" MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:symptom) WHERE toLower(s.name) CONTAINS toLower('{symptom}') RETURN s.name AS 症状, collect(DISTINCT d.name) AS 相关疾病 LIMIT 100 """ if intent == "disease_drugs" and disease: return f""" MATCH (d:Disease)-[:HAS_Drug]->(m:drug) WHERE toLower(d.name) CONTAINS toLower('{disease}') RETURN d.name AS 疾病, collect(DISTINCT m.name) AS 用药 LIMIT 100 """ if intent == "disease_department" and disease: return f""" MATCH (d:Disease)-[:IS_OF_DEPARTMENT]->(dp:department) WHERE toLower(d.name) CONTAINS toLower('{disease}') RETURN d.name AS 疾病, collect(DISTINCT dp.name) AS 科室 LIMIT 50 """ if intent == "disease_treatments" and disease: return f""" MATCH (d:Disease)-[:HAS_Treatment]->(t:treatment) WHERE toLower(d.name) CONTAINS toLower('{disease}') RETURN d.name AS 疾病, collect(DISTINCT t.name) AS 治疗方法 LIMIT 100 """ if intent == "disease_complications" and disease: return f""" MATCH (d:Disease)-[:HAS_Complication]->(c:complication) WHERE toLower(d.name) CONTAINS toLower('{disease}') RETURN d.name AS 疾病, collect(DISTINCT c.name) AS 并发症 LIMIT 100 """ return "MATCH (d:Disease) RETURN d.name AS 疾病 LIMIT 10" # 主问答函数 def answer(user_text): text = normalize(user_text) intent = detect_intent(text) candidate = clean_entity_phrase(extract_candidate_phrase(text)) disease, symptom = None, None if intent == "symptom_diseases": symptom = fuzzy_pick_symptom(candidate) or STATE.last_symptom else: disease = fuzzy_pick_disease(candidate) or STATE.last_disease if not disease and intent.startswith("disease_"): disease = STATE.last_disease if not symptom and intent == "symptom_diseases": symptom = STATE.last_symptom cypher = to_cypher(intent, disease, symptom) data = graph.run(cypher).data() def first_list(key): return (data and key in data[0] and data[0][key]) or [] # 输出回答 if intent == "disease_symptoms": items = first_list("症状") answer = f"「{disease or candidate}」的症状包括：{ '、'.join(items[:30]) }" if items else f"未查到与「{disease or candidate}」相关的症状。" if disease: STATE.last_disease = disease elif intent == "symptom_diseases": items = first_list("相关疾病") answer = f"出现「{symptom or candidate}」可能与以下疾病相关：{ '、'.join(items[:30]) }" if items else f"未查到与「{symptom or candidate}」相关的疾病。" if symptom: STATE.last_symptom = symptom elif intent == "disease_drugs": items = first_list("用药") answer = f"「{disease or candidate}」常用药物：{ '、'.join(items[:30]) }" if items else f"未查到与「{disease or candidate}」相关的药物。" if disease: STATE.last_disease = disease elif intent == "disease_department": items = first_list("科室") answer = f"「{disease or candidate}」建议就诊科室：{ '、'.join(items[:10]) }" if items else f"未查到与「{disease or candidate}」相关的科室。" if disease: STATE.last_disease = disease elif intent == "disease_treatments": items = first_list("治疗方法") answer = f"「{disease or candidate}」常见治疗方法：{ '、'.join(items[:30]) }" if items else f"未查到与「{disease or candidate}」相关的治疗方法。" if disease: STATE.last_disease = disease elif intent == "disease_complications": items = first_list("并发症") answer = f"「{disease or candidate}」常见并发症：{ '、'.join(items[:30]) }" if items else f"未查到与「{disease or candidate}」相关的并发症。" if disease: STATE.last_disease = disease else: answer = "我还不能理解这个问题，可以换种问法试试～" STATE.last_intent = intent return { "intent": intent, "disease": disease, "symptom": symptom, "answer": answer, "cypher": cypher, } # 命令行对话入口 if __name__ == "__main__": print("🩺 医疗知识图谱问答系统") while True: q = input("患者：").strip() if q.lower() in {"q", "quit", "exit", "再见", "结束对话"}: print("医生：祝您早日康复！") break resp = answer(q) print("医生：", resp["answer"]) 注：在初版代码实现中，我发现问答系统不能很好的延续上文提问内容。
为此，我在代码中加入了对话记忆等相关内容，这样让问答更具有延续性。
五、对话系统实现与交互演示
运行结果如图所示·：
六、问题与改进
1.运行的时候提问“科室“”相关问题时会出现无法找到（已解决）
解决方案：线上排查发现，“挂什么科”查询返回为空并非数据缺失所致，而是关系类型命名不一致（如 IS_OF_Department vs IS_OF_DEPARTMENT）导致查询模板无法命中。
解决方案：
（1）在问答层临时兼容多个关系名；
（2）在图谱侧使用 APOC/纯 Cypher 将历史关系统一为 IS_OF_DEPARTMENT；
（3）同时统一终点标签为 department。
同时加入了“科室缺失时按照疾病名称推断”这一机制
修复后，“挂什么科”问答输出正常。
# -*- coding: utf-8 -*- """ build_medical_graph.py@composedbysharniT 功能：从 disease1.csv 构建医疗知识图谱（包含“科室缺失时按疾病名猜测”逻辑） 说明： - 关系名与问答系统保持一致：HAS_SYMPTOM / HAS_Drug / HAS_Treatment / IS_OF_DEPARTMENT 等 - 仅当 department 为空或 'nan' 时，调用 guess_department_by_disease() """ from py2neo import Graph, Node, Relationship import pandas as pd import math import re # Neo4j 连接 graph = Graph("bolt://localhost:7687", auth=("neo4j", "12345678")) CLEAR_BEFORE_IMPORT = False if CLEAR_BEFORE_IMPORT: graph.run("MATCH (n) DETACH DELETE n") print("已清空数据库") #读取 CSV df = pd.read_csv("disease1.csv", encoding="ANSI") # 去掉空列（Unnamed） df = df[[c for c in df.columns if not str(c).lower().startswith("unnamed")]] print(f"已加载 CSV，共 {len(df)} 行，列：{list(df.columns)}") #关系映射 node_rel_map = { "alias": "HAS_ALIAS", "part": "IS_OF_PART", "age": "IS_OF_AGE", "infection": "IS_INFECTIOUS", "insurance": "IN_INSURANCE", "department": "IS_OF_DEPARTMENT", "checklist": "HAS_CHECKLIST", "symptom": "HAS_SYMPTOM", "complication":"HAS_Complication", "treatment": "HAS_Treatment", "drug": "HAS_Drug", "period": "CURE_PERIOD", "rate": "CURE_RATE", "money": "NEED_MONEY", } # 分隔与清洗 SEP_PATTERN = re.compile(r"[，、,;；\s\|]+") def split_values(s: str): if not s or s.lower() == "nan": return [] # 统一替换分隔符，分裂后去空 parts = [p.strip() for p in SEP_PATTERN.split(s) if p and p.strip() and p.strip().lower() != "nan"] # 去掉可能的占位符，如“[详细]” parts = [p for p in parts if p not in {"[详细]", "详细", "无", "暂无", "none"}] return parts def is_empty(x) -> bool: if x is None: return True if isinstance(x, float) and math.isnan(x): return True s = str(x).strip() return s == "" or s.lower() == "nan" # 规则：科室缺失时按疾病名猜测 def guess_department_by_disease(disease_name: str) -> str: """ 返回单个科室名称字符串；若无法判断，返回空字符串 说明：只在 department 列为空/缺失时使用，不会覆盖已有真实数据 """ if not disease_name: return "" name = disease_name.lower() # 关键字 -> 科室（可按需扩充,没什么必要） rules = [ (["癫痫", "脑", "抽搐", "惊厥", "中风", "偏瘫", "头痛", "眩晕", "帕金森"], "神经内科"), (["腹泻", "腹痛", "胃痛", "肠炎", "胃炎", "痢疾", "消化", "胰腺", "肝硬化", "肝炎"], "消化内科"), (["咳嗽", "肺炎", "支气管炎", "气喘", "呼吸衰竭", "哮喘"], "呼吸内科"), (["冠心病", "心绞痛", "心衰", "心肌梗死", "心律失常", "高血压"], "心内科"), (["甲亢", "糖尿病", "甲状腺", "内分泌"], "内分泌科"), (["肾炎", "肾病", "尿毒症", "肾功能", "肾衰"], "肾内科"), (["过敏", "湿疹", "皮炎", "荨麻疹", "银屑病"], "皮肤科"), (["阳痿", "早泄", "前列腺", "泌尿", "肾结石"], "泌尿外科"), (["盆腔炎", "月经", "多囊", "子宫", "卵巢", "妊娠"], "妇科"), (["小儿", "儿童", "新生儿", "发育迟缓", "手足口", "疱疹性咽峡炎"], "儿科"), (["鼻炎", "咽炎", "扁桃体", "中耳炎", "耳鸣", "鼻窦炎"], "耳鼻喉科"), (["近视", "白内障", "青光眼", "结膜炎"], "眼科"), (["龋齿", "牙周", "口腔溃疡", "智齿"], "口腔科"), (["风湿", "关节炎", "红斑狼疮", "痛风"], "风湿免疫科"), (["贫血", "白血病", "淋巴瘤", "血小板"], "血液科"), (["发热不明原因", "败血症", "结核", "流行性", "感染"], "感染科"), (["抑郁", "焦虑", "精神分裂", "失眠"], "精神科"), (["肿瘤", "癌", "恶性"], "肿瘤科"), (["骨折", "关节置换", "腰椎间盘突出", "颈椎病"], "骨科"), ] for keys, dept in rules: if any(k in name for k in keys): return dept return "" # 不确定就返回空，让导入逻辑跳过 # 主导入：逐行建节点与关系（带批量事务） tx = graph.begin() batch_size = 50 for idx, row in df.iterrows(): disease_name = str(row.get("name", "")).strip() if not disease_name: continue # 疾病节点 disease_node = Node("Disease", name=disease_name) tx.merge(disease_node, "Disease", "name") # 遍历属性列 for col_name, rel_type in node_rel_map.items(): if col_name not in df.columns: continue raw_value = row.get(col_name, "") # 统一成字符串处理 if is_empty(raw_value) else str(raw_value).strip() if col_name == "department" and (raw_str == "" or raw_str.lower() == "nan"): guessed = guess_department_by_disease(disease_name) raw_str = guessed # 如果猜不到则为空串 if raw_str == "" or raw_str.lower() == "nan": continue # 拆分多值 values = split_values(raw_str) if not values: continue target_label = col_name for v in values: target_node = Node(target_label, name=v) tx.merge(target_node, target_label, "name") tx.merge(Relationship(disease_node, rel_type, target_node)) # 批量提交 if (idx + 1) % batch_size == 0: tx.commit() print(f"✅ 已处理 {idx + 1} / {len(df)} 行") tx = graph.begin() tx.commit() print("图谱构建完成！") 修改后关系图
