简介这套基于Python的医学知识图谱问答系统源码面向医学研究者、临床医生及对智能问诊感兴趣的开发者旨在将分散的医学数据整理为结构化知识图谱并通过问答交互快速返回结果。资源包共31个文件大小约49.19MB其中包含8个Python源文件覆盖医学图谱构建、问题分类、意图解析、答案检索与对话整合等完整链路另有3个编译文件、9张流程示意图、9个文本文件、JSON配置及PPTX演示文稿从数据准备到成果展示一应俱全。代码模块划分清晰配合说明文档和数据文件夹可帮助读者理解医学实体抽取、图谱存储及问答匹配的实现思路也能直接作为医疗信息化课题或毕业设计的参考框架。目前已有176人学习下载适合希望掌握知识图谱落地方法的中高级Python开发者。1. 医学知识图谱问答系统真正难住的不是问答是图谱本身医学知识图谱问答系统说白了就是让用户用自然语言问“高血压不能吃什么药”系统在后台把这句话翻译成对知识图谱的查询再返回一个能读懂的答案。很多课程设计和毕业论文把这个项目当成“Python 问答接口”来做结果往往卡在最前面知识图谱本身没有建好后面问答做得再花哨也答不对。我的经验是这套系统的复杂度不在模型而在数据清洗、Schema 设计和问句到查询的翻译规则。本文要把一条走得通的落地路径拆开讲适合正在做医学问答系统源码、知识图谱构建或者想从零搭一个能演示的智能问答系统的工程师。下面按“建模 → 灌数据 → 写问答 → 避坑 → 验收”的顺序展开。2. 先把 Schema 当本体建模来做实体与关系的取舍医学问答系统的可回答范围在动手写代码之前就由 Schema 定死了。你定义了“药物 — 治疗 → 疾病”系统就能回答“阿司匹林治什么病”你没定义“食物与药物的相互作用”用户问“吃柚子还能吃降压药吗”就永远答不上来。这跟工业场景下的知识图谱设计是一个道理图谱后期难改成本最高的往往不是数据规模而是本体层反复调整。所以第一步别急着爬数据先花半天把实体、属性、关系定清楚。2.1 医学核心实体选六个还是十二个常见做法是从六个实体类型起步疾病、症状、药物、检查项目、科室、身体部位。这六个类型覆盖了医学问句里 80% 以上的主语和宾语比如“感冒挂什么科”“肺炎做什么检查”“头痛吃什么药”主语和查询目标都能落在上面。每类实体的核心属性建议这样规划实体类型核心属性唯一键建议Disease疾病name, alias, category, icd_codename规范化名称Symptom症状name, alias, descriptionnameDrug药物name, alias, spec, manufacturernameExam检查项目name, alias, exam_typenameDepartment科室name, alias, locationnameBodyPart身体部位name, alias, systemname我不建议一开始就把手术、医生、医保政策全部纳入。实体类型越多后面做实体链接和意图识别要维护的词典就越庞大。把“医生”这类角色先放掉因为它的属性职称、排班和问答复杂度远超六类起步能覆盖的范围。原则是能成为问句入口的词才做实体只是做展示说明的内容放属性。比如“头痛”如果只是某个疾病的描述文字不要单独建 Symptom 节点如果用户会直接问“头痛挂什么科”它就必须是实体。这是知识表示里最常见的设计分歧宁可先少建也不要建一堆用不上的空节点。2.2 关系拆细一点问答就好写一点实体定了之后关系类型决定问答系统的意图种类。我一般把关系拆成动词短语而不是用一个通用的“related_to”糊弄过去。医学问答系统里最常用的关系如下疾病 → 症状Disease -[:HAS_SYMPTOM]- Symptom对应问法“高血压有什么症状”药物 → 疾病Drug -[:TREATS]- Disease对应问法“阿司匹林治什么病”疾病 → 科室Disease -[:BELONGS_TO_DEPT]- Department对应问法“偏头痛挂什么科”疾病 → 检查Disease -[:NEEDS_EXAM]- Exam对应问法“糖尿病要做什么检查”药物 → 成分Drug -[:CONTAINS]- Ingredient对应问法“这个药含什么成分”检查 → 身体部位Exam -[:EXAMINES]- BodyPart对应问法“胃镜查什么部位”关系方向的约定要全项目统一。我习惯把查询主语放在箭头左边让动作方向从主语指向宾语。比如“阿司匹林治什么病”是Drug -[:TREATS]- Disease反过来写Disease -[:TREATS]- Drug虽然图谱上等价但后面写 Cypher 时极易搞乱方向。每个关系都要在注释里写清“左端类型 右端类型”这样问答模块拿到意图名就知道该往哪个方向查。关系拆细还有一个好处意图识别可以直接拿关系名做映射不用再维护一套“意图 → 查询模板”的中间表。2.3 用 Cypher 把 Schema 固化下来Schema 定完不要只留在文档里要落成 Neo4j 的约束和索引。很多人数据灌完了才发现重复节点一大堆就是少了这一步唯一约束。我一般在写导入脚本之前先执行一组约束CREATE CONSTRAINT disease_name_unique IF NOT EXISTS ON (d:Disease) ASSERT d.name IS UNIQUE; CREATE CONSTRAINT drug_name_unique IF NOT EXISTS ON (d:Drug) ASSERT d.name IS UNIQUE; CREATE CONSTRAINT symptom_name_unique IF NOT EXISTS ON (s:Symptom) ASSERT s.name IS UNIQUE; CREATE INDEX dept_name_idx IF NOT EXISTS FOR (d:Department) ON (d.name); CREATE INDEX exam_name_idx IF NOT EXISTS FOR (e:Exam) ON (e.name);说明IF NOT EXISTS是 Neo4j 4.x 之后的幂等写法脚本重复执行不会报错ASSERT ... IS UNIQUE为节点创建唯一约束后续用MERGE写入时能靠它去重。注意这里只对 Disease、Drug、Symptom 建了唯一约束因为这三类是问句里最容易出现重复别名的主体Department 和 Exam 条目少做了普通索引就够了约束过多会拖慢批量导入速度。执行完这段再去看 Neo4j 的约束列表确认每个标签都生效再进入数据层。3. 用 Python 把医学三元组灌进 Neo4j从 CSV 到 Cypher 的完整管道Schema 固化后接下来是数据管道。先说明数据来源我建议用公开的医学百科词条做语料比如药物说明书里的“适应症”“禁忌症”小节、疾病词条里的“临床表现”“就诊科室”小节。用 Python 爬虫抓公开页面转成三元组是合法且常见的做法前提是控制抓取频率、尊重源站 robots 声明、不要用抓下来的数据做商用。当然如果你只想跑通演示系统也可以直接用公开的医学数据包或者手工整理 200 条三元组也够支撑课堂答辩。3.1 先落 CSV再连库别一边爬一边写图我见过不少人写脚本时爬一个词条就往 Neo4j 插一条结果爬到一半字段没对齐库里留下一堆半成品节点。正确做法是先把结果落成 CSV人工抽查后再导入。这样做的好处是导入失败可以重来不用反复清库CSV 也能当数据集交付物写报告时直接用。下面是我常用的实体表生成逻辑import pandas as pd # entities_drug.csv 示例结构 drug_rows [] for item in drug_pages: # drug_pages 来自解析后的词条数据 drug_rows.append({ name: item[standard_name], # 规范化名称唯一键 alias: |.join(item[alias_list]), # 别名用竖线分隔存同一条记录 spec: item.get(spec, ), manufacturer: item.get(manufacturer, ), }) df_drug pd.DataFrame(drug_rows) df_drug df_drug.drop_duplicates(subset[name]) # 按唯一键去重 df_drug.to_csv(entities_drug.csv, indexFalse, encodingutf-8-sig)说明alias字段把多个别名用|拼到同一行是为了导入后用split做别名索引时方便如果你打算在 Neo4j 里为每个别名单独建节点就不要这么存。drop_duplicates(subset[name])是按唯一键去重的兜底防止同一个药品在多个词条里出现。这里故意没做复杂的清洗函数先把数据形状稳住后面有问题再回头处理是工程上的取舍。3.2 用 py2neo 批量写入比逐条插入快一个量级CSV 准备好之后用 py2neo 读取并分批写入。逐条CREATE3000 个节点可能要半分钟而批量提交能在几秒内完成。核心代码from py2neo import Graph, Node, Relationship import pandas as pd graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) df_drug pd.read_csv(entities_drug.csv, dtypestr).fillna() df_disease pd.read_csv(entities_disease.csv, dtypestr).fillna() df_rel pd.read_csv(relations.csv, dtypestr).fillna() BATCH_SIZE 500 # 单个事务的写入条数过大容易爆内存 def load_entities(df, label): for i in range(0, len(df), BATCH_SIZE): batch df.iloc[i:iBATCH_SIZE] tx graph.begin() for _, row in batch.iterrows(): node Node(label, **row.to_dict()) tx.merge(node, label, name) # 按唯一键合并 tx.commit() print(f{label} 已写入 {i len(batch)} 条) load_entities(df_drug, Drug) load_entities(df_disease, Disease)这里关键参数是tx.merge(node, label, name)它先查有没有同名节点有就只更新属性没有才创建。配合第 2 章建的唯一约束能从根源上避免重复疾病、重复药物。BATCH_SIZE我一般设在 300 到 500 之间太小事务开销大太大一次提交可能把 JVM 堆顶爆第一次跑先用 200 稳妥确认内存没问题再加大。注意dtypestr是为了防止 pandas 把药品规格里的“0.5g”读成数字导致属性类型全乱。3.3 写完先验证数量别急着写问答数据导入后不要急着写问答接口先用几条 Cypher 确认图结构。常见做法是查每个标签的节点数和每种关系的条数再抽一条完整路径看方向是否正确MATCH (d:Disease) RETURN count(d) AS disease_count; MATCH (r:Drug)-[:TREATS]-(d:Disease) RETURN count(r) AS treat_count; MATCH (d:Drug)-[:TREATS]-(dis:Disease) WHERE d.name 阿司匹林 RETURN dis.name AS disease_name LIMIT 10;说明第二条 count 查询如果返回 0说明关系文件里的实体名没有跟实体表对上常见原因是同义词不一致。比如药典里写“乙酰水杨酸”CSV 里却写“阿司匹林”MERGE 时会按 name 找不到已有节点实际插入了一堆孤立节点。这时应该回去检查 relations.csv 里的两端名称统一成实体表里的 name而不是在查询时临时做别名匹配。抽路径查询能直接看到 TREATS 方向的箭头指向方向反了在这里一眼就能发现。4. 问答系统的三层解析意图识别、实体提取、查询生成图谱装好之后真正面向用户的是问答模块。我从不建议这类课程设计和中小型项目一上来就套 BERT 做语义匹配原因很简单医学问法相对固定规则加词典的方案可解释性强跑得也足够快等规则覆盖不了的 case 累积到一定量再换模型也不迟。问答链路就三件事识别意图、提取实体、拼接 Cypher。4.1 同义词表是实体提取的第一道防线先看实体提取。用户不会总按标准名提问“高血压”可能被说成“血压高”“原发性高血压”“阿司匹林”可能是“乙酰水杨酸”“APC”。我一般维护一个同义词映射表加载到 jieba 用户词典里再做一层别名归一化import jieba import jieba.posseg as pseg ALIAS_MAP { 乙酰水杨酸: 阿司匹林, 血压高: 高血压, 原发性高血压: 高血压, 肺部ct: CT检查, } def build_user_dict(): for entity_name in ALIAS_MAP.values(): jieba.add_word(entity_name, freq2000, tagn) for alias in ALIAS_MAP.keys(): jieba.add_word(alias, freq2000, tagn) def extract_entities(question): words pseg.cut(question) candidates [] for word, flag in words: if flag.startswith(n) and len(word) 2: # 名词候选 candidates.append(word) # 归一化别名转标准名 normalized [ALIAS_MAP.get(w, w) for w in candidates] return normalized # 后面再做实体类型匹配说明freq2000是给自定义词一个较高词频让 jieba 优先把它们切成完整词而不是把“乙酰水杨酸”切成“乙酰 水杨酸”。flag.startswith(n)只保留名词候选过滤掉“什么”“哪些”这类疑问词。注意这里只是候选集合最终能不能当成实体还要拿去和 Neo4j 里的节点做比对不能在常识上只凭词性下结论。4.2 意图识别用模板映射表比多个 if-else 好维护意图识别我的做法是维护一张“意图 → 关系 → 问句模板”的配置表代码里只做顺序匹配。直接看实现INTENT_PATTERNS [ { intent: drug_treats_disease, relation: TREATS, patterns: [{drug}治什么病, {drug}能治什么, {drug}对什么有效], order: 1, # 具体意图排前面 }, { intent: disease_has_symptom, relation: HAS_SYMPTOM, patterns: [{disease}什么症状, {disease}有哪些表现, {disease}会怎样], order: 2, }, { intent: disease_belongs_dept, relation: BELONGS_TO_DEPT, patterns: [{disease}挂什么科, {disease}看哪个科], order: 3, }, ] def classify_intent(question, entity): for item in sorted(INTENT_PATTERNS, keylambda x: x[order]): for pat in item[patterns]: test pat.replace({disease}, entity).replace({drug}, entity) if test in question.replace( , ): return item[intent], item[relation] return None, None说明这里没有用复杂的句法分析核心思路是“把问句中已识别出的实体替换回模板再做子串匹配”。order字段保证意图之间的优先级比如“阿司匹林能治什么病”同时命中的可能性很小但“高血压有什么症状”和“高血压看什么科”如果字面有重叠order 小的先匹配。模板覆盖不了的问法后续要做的就是加新模板而不是加 if 分支。维护成本和理解成本都低答辩时也讲得清楚。4.3 拼 Cypher 用参数化查询别用字符串拼接意图和实体都有了查询生成就是一层翻译。这里最需要小心的坑不要把实体名称直接拼进 Cypher 字符串。用户问句里可能存在引号、括号直接拼接轻则查询报错重则把整条查询语义搞坏。用参数化查询from py2neo import Graph graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) RELATION_CYPHER { TREATS: MATCH (d:{entity_label})-[r:TREATS]-(target) RETURN target.name AS answer, HAS_SYMPTOM: MATCH (d:{entity_label})-[r:HAS_SYMPTOM]-(target) RETURN target.name AS answer, BELONGS_TO_DEPT: MATCH (d:{entity_label})-[r:BELONGS_TO_DEPT]-(target) RETURN target.name AS answer, } def answer(question): entity extract_entities(question) # 简写实际取第一个匹配到图谱的实体 if not entity: return {status: no_entity, answer: 没识别到疾病或药物换个说法试试} intent, relation classify_intent(question, entity) if not intent: return {status: no_intent, answer: 这个问题我还不会换个问法问问看} entity_label detect_label(entity) # 判断 Drug / Disease / Symptom cypher RELATION_CYPHER[relation].replace({entity_label}, entity_label) results graph.run(cypher, entity_nameentity).data() if not results: return {status: no_result, answer: f图谱里有这个实体但找不到{relation}关系} answers [r[answer] for r in results[:5]] return {status: ok, answer: 、.join(answers)}说明graph.run(cypher, entity_nameentity)里的entity_name会作为参数传给 Cypher 中的$entity_name避免特殊字符干扰查询。detect_label用两个 MATCH 分别判断实体是 Drug 还是 Disease这里不贴全了核心是按节点 label 的 count 来决定把实体放进哪类节点。status字段不是给用户看的是给后面离线评测用的。把“没识别到实体”“没匹配到关系”“查询成功”拆成三种状态后面统计准确率才能知道失败到底失败在哪一环。5. 医学问答系统最常翻车的四个现场与排查路径这套系统我前前后后写过三轮最典型的坑基本集中在下面四个。每一条都按“现象 → 原因 → 解决”来写都是能直接照做的排查路径。5.1 实体链接总是把“感冒症状”当成疾病名现象用户问“感冒症状有哪些”系统把“感冒症状”整个识别成疾病实体查不到节点返回“没找到疾病”。原因jieba 词典里“感冒”和“症状”各自是词但组合起来“感冒症状”在自定义词典里被整体切出或者实体提取逻辑把所有两字以上的名词都当候选。解决实体候选提取后不要直接拿去查库先把候选词与图谱节点名做精确匹配匹配不上的再尝试把“症状”“检查”“科室”这类词尾去掉重试TAIL_STOP_WORDS [症状, 检查, 科室, 吃什么药, 怎么办] def normalize_entity(candidate): for tail in TAIL_STOP_WORDS: if candidate.endswith(tail) and len(candidate) len(tail): return candidate[:-len(tail)] return candidate这个函数加在extract_entities之后、查库之前能解决一半实体过宽的问题。剩余的问题靠图谱侧唯一约束兜底查不到就是查不到不要硬猜。5.2 方向定义没问题查询却返回空现象数据管道验证时(Drug)-[:TREATS]-(Disease)能查出结果问答接口里同样意图却总是空结果。原因问答模块的实体类型判断错了把“阿司匹林”识别成了 Disease于是执行了(Disease)-[:TREATS]-(target)方向反了自然为空。解决排查时不看上层代码先在 Neo4j Browser 里用反向查询确认数据确实存在MATCH (d)-[r:TREATS]-(target) WHERE d.name $entity_name RETURN labels(d) AS from_label, type(r) AS rel, labels(target) AS to_label再看返回的from_label是什么如果发现是 Disease 而不是 Drug问题不在查询生成而在实体分类。心跳法则是问答系统的空结果先区分“图谱里没这条关系”和“图谱里一查就有但查错了方向”这两者的排查路径完全不同。5.3 意图模板互相覆盖答案张冠李戴现象问“高血压做什么检查”返回的是“高血压有什么症状”的答案。原因模板匹配时没有去掉标点和空格或者两个模板有公共子串“什么症状”和“做什么检查”里的“什么”被错误命中。解决一是把模板替换成实体后做全串匹配而不是包含匹配二是在配置表里加 order 字段把包含“检查”“科室”这类特征词的意图排前面。我实际维护时还加了一个小约束模板禁止出现“什么”这种无信息量词开头必须包含一个能区分意图的行为动词比如“治”“挂”“做”“有”。5.4 大批量导入时 Neo4j 内存爆掉或卡死现象一次性导入 5 万条关系Neo4j 进程直接卡死日志里报堆内存溢出。原因一个事务提交的数据量太大或者事务里混合了大量 MERGE 操作导致每一条都要查索引。解决把数据分批提交每批 200 到 500 条提交后释放事务BATCH_SIZE 300 tx graph.begin() for idx, (start, rel_type, end) in enumerate(relation_triples): tx.run( MATCH (a {name:$s}), (b {name:$e}) MERGE (a)-[r:%s]-(b) RETURN r % rel_type, sstart, eend, ) if (idx 1) % BATCH_SIZE 0: tx.commit() tx graph.begin() # 旧事务已关闭必须新开注意这里%s拼接的只是关系类型名关系类型来自程序内部的固定枚举不是用户输入所以不会引入安全风险两端的节点名仍然用参数传递。如果还卡就去 Neo4j 的neo4j.conf里把dbms.memory.heap.max_size从默认值调大同时确认唯一约束已经建好否则 MERGE 每一条都要全表扫描一次。6. 把问答准确率测出来离线评测与验收技巧写一套自动化评测脚本是这套源码从“能跑”到“真能用”的分水岭。我一般准备一个 30 到 50 条的测试集合覆盖每个意图至少 5 条同时故意混入别名和无结果问法。测试集用 Python 直接构造便于统计test_cases [ {question: 阿司匹林治什么病, expect_intent: drug_treats_disease, expect_answer_contains: 发热}, {question: 乙酰水杨酸能治什么, expect_intent: drug_treats_disease, expect_answer_contains: 发热}, {question: 高血压有什么症状, expect_intent: disease_has_symptom, expect_answer_contains: 头晕}, {question: 偏头痛挂什么科, expect_intent: disease_belongs_dept, expect_answer_contains: 神经内科}, {question: 这个病会不会传染, expect_intent: None, expect_answer_contains: None}, ] def evaluate(test_cases): intent_hit, entity_hit, answer_hit 0, 0, 0 for case in test_cases: result answer(case[question]) if result[intent] case[expect_intent]: intent_hit 1 if result[status] ! no_entity: entity_hit 1 if case[expect_answer_contains] and case[expect_answer_contains] in result[answer]: answer_hit 1 return { intent_acc: intent_hit / len(test_cases), entity_acc: entity_hit / len(test_cases), answer_acc: answer_hit / len(test_cases), }跑完先看三个数意图准确率低说明模板覆盖不够实体准确率低说明词典和归一化有问题两者都正常但答案准确率低问题在 Cypher 生成或返回结果的截断逻辑。我习惯把这套评测脚本留着每次改完同义词表或模板就重跑一遍防止修了一个坑又带出另一个坑。这也是我拿到任何同类源码时做的第一件事不看花哨的界面先建测试集跑准确率再决定是否值得往里面加语义模型。希望帮到你。本文还有配套的精品资源点击获取 医学知识图谱问答系统构建：从Schema设计到Cypher查询实践
简介这套基于Python的医学知识图谱问答系统源码面向医学研究者、临床医生及对智能问诊感兴趣的开发者旨在将分散的医学数据整理为结构化知识图谱并通过问答交互快速返回结果。资源包共31个文件大小约49.19MB其中包含8个Python源文件覆盖医学图谱构建、问题分类、意图解析、答案检索与对话整合等完整链路另有3个编译文件、9张流程示意图、9个文本文件、JSON配置及PPTX演示文稿从数据准备到成果展示一应俱全。代码模块划分清晰配合说明文档和数据文件夹可帮助读者理解医学实体抽取、图谱存储及问答匹配的实现思路也能直接作为医疗信息化课题或毕业设计的参考框架。目前已有176人学习下载适合希望掌握知识图谱落地方法的中高级Python开发者。1. 医学知识图谱问答系统真正难住的不是问答是图谱本身医学知识图谱问答系统说白了就是让用户用自然语言问“高血压不能吃什么药”系统在后台把这句话翻译成对知识图谱的查询再返回一个能读懂的答案。很多课程设计和毕业论文把这个项目当成“Python 问答接口”来做结果往往卡在最前面知识图谱本身没有建好后面问答做得再花哨也答不对。我的经验是这套系统的复杂度不在模型而在数据清洗、Schema 设计和问句到查询的翻译规则。本文要把一条走得通的落地路径拆开讲适合正在做医学问答系统源码、知识图谱构建或者想从零搭一个能演示的智能问答系统的工程师。下面按“建模 → 灌数据 → 写问答 → 避坑 → 验收”的顺序展开。2. 先把 Schema 当本体建模来做实体与关系的取舍医学问答系统的可回答范围在动手写代码之前就由 Schema 定死了。你定义了“药物 — 治疗 → 疾病”系统就能回答“阿司匹林治什么病”你没定义“食物与药物的相互作用”用户问“吃柚子还能吃降压药吗”就永远答不上来。这跟工业场景下的知识图谱设计是一个道理图谱后期难改成本最高的往往不是数据规模而是本体层反复调整。所以第一步别急着爬数据先花半天把实体、属性、关系定清楚。2.1 医学核心实体选六个还是十二个常见做法是从六个实体类型起步疾病、症状、药物、检查项目、科室、身体部位。这六个类型覆盖了医学问句里 80% 以上的主语和宾语比如“感冒挂什么科”“肺炎做什么检查”“头痛吃什么药”主语和查询目标都能落在上面。每类实体的核心属性建议这样规划实体类型核心属性唯一键建议Disease疾病name, alias, category, icd_codename规范化名称Symptom症状name, alias, descriptionnameDrug药物name, alias, spec, manufacturernameExam检查项目name, alias, exam_typenameDepartment科室name, alias, locationnameBodyPart身体部位name, alias, systemname我不建议一开始就把手术、医生、医保政策全部纳入。实体类型越多后面做实体链接和意图识别要维护的词典就越庞大。把“医生”这类角色先放掉因为它的属性职称、排班和问答复杂度远超六类起步能覆盖的范围。原则是能成为问句入口的词才做实体只是做展示说明的内容放属性。比如“头痛”如果只是某个疾病的描述文字不要单独建 Symptom 节点如果用户会直接问“头痛挂什么科”它就必须是实体。这是知识表示里最常见的设计分歧宁可先少建也不要建一堆用不上的空节点。2.2 关系拆细一点问答就好写一点实体定了之后关系类型决定问答系统的意图种类。我一般把关系拆成动词短语而不是用一个通用的“related_to”糊弄过去。医学问答系统里最常用的关系如下疾病 → 症状Disease -[:HAS_SYMPTOM]- Symptom对应问法“高血压有什么症状”药物 → 疾病Drug -[:TREATS]- Disease对应问法“阿司匹林治什么病”疾病 → 科室Disease -[:BELONGS_TO_DEPT]- Department对应问法“偏头痛挂什么科”疾病 → 检查Disease -[:NEEDS_EXAM]- Exam对应问法“糖尿病要做什么检查”药物 → 成分Drug -[:CONTAINS]- Ingredient对应问法“这个药含什么成分”检查 → 身体部位Exam -[:EXAMINES]- BodyPart对应问法“胃镜查什么部位”关系方向的约定要全项目统一。我习惯把查询主语放在箭头左边让动作方向从主语指向宾语。比如“阿司匹林治什么病”是Drug -[:TREATS]- Disease反过来写Disease -[:TREATS]- Drug虽然图谱上等价但后面写 Cypher 时极易搞乱方向。每个关系都要在注释里写清“左端类型 右端类型”这样问答模块拿到意图名就知道该往哪个方向查。关系拆细还有一个好处意图识别可以直接拿关系名做映射不用再维护一套“意图 → 查询模板”的中间表。2.3 用 Cypher 把 Schema 固化下来Schema 定完不要只留在文档里要落成 Neo4j 的约束和索引。很多人数据灌完了才发现重复节点一大堆就是少了这一步唯一约束。我一般在写导入脚本之前先执行一组约束CREATE CONSTRAINT disease_name_unique IF NOT EXISTS ON (d:Disease) ASSERT d.name IS UNIQUE; CREATE CONSTRAINT drug_name_unique IF NOT EXISTS ON (d:Drug) ASSERT d.name IS UNIQUE; CREATE CONSTRAINT symptom_name_unique IF NOT EXISTS ON (s:Symptom) ASSERT s.name IS UNIQUE; CREATE INDEX dept_name_idx IF NOT EXISTS FOR (d:Department) ON (d.name); CREATE INDEX exam_name_idx IF NOT EXISTS FOR (e:Exam) ON (e.name);说明IF NOT EXISTS是 Neo4j 4.x 之后的幂等写法脚本重复执行不会报错ASSERT ... IS UNIQUE为节点创建唯一约束后续用MERGE写入时能靠它去重。注意这里只对 Disease、Drug、Symptom 建了唯一约束因为这三类是问句里最容易出现重复别名的主体Department 和 Exam 条目少做了普通索引就够了约束过多会拖慢批量导入速度。执行完这段再去看 Neo4j 的约束列表确认每个标签都生效再进入数据层。3. 用 Python 把医学三元组灌进 Neo4j从 CSV 到 Cypher 的完整管道Schema 固化后接下来是数据管道。先说明数据来源我建议用公开的医学百科词条做语料比如药物说明书里的“适应症”“禁忌症”小节、疾病词条里的“临床表现”“就诊科室”小节。用 Python 爬虫抓公开页面转成三元组是合法且常见的做法前提是控制抓取频率、尊重源站 robots 声明、不要用抓下来的数据做商用。当然如果你只想跑通演示系统也可以直接用公开的医学数据包或者手工整理 200 条三元组也够支撑课堂答辩。3.1 先落 CSV再连库别一边爬一边写图我见过不少人写脚本时爬一个词条就往 Neo4j 插一条结果爬到一半字段没对齐库里留下一堆半成品节点。正确做法是先把结果落成 CSV人工抽查后再导入。这样做的好处是导入失败可以重来不用反复清库CSV 也能当数据集交付物写报告时直接用。下面是我常用的实体表生成逻辑import pandas as pd # entities_drug.csv 示例结构 drug_rows [] for item in drug_pages: # drug_pages 来自解析后的词条数据 drug_rows.append({ name: item[standard_name], # 规范化名称唯一键 alias: |.join(item[alias_list]), # 别名用竖线分隔存同一条记录 spec: item.get(spec, ), manufacturer: item.get(manufacturer, ), }) df_drug pd.DataFrame(drug_rows) df_drug df_drug.drop_duplicates(subset[name]) # 按唯一键去重 df_drug.to_csv(entities_drug.csv, indexFalse, encodingutf-8-sig)说明alias字段把多个别名用|拼到同一行是为了导入后用split做别名索引时方便如果你打算在 Neo4j 里为每个别名单独建节点就不要这么存。drop_duplicates(subset[name])是按唯一键去重的兜底防止同一个药品在多个词条里出现。这里故意没做复杂的清洗函数先把数据形状稳住后面有问题再回头处理是工程上的取舍。3.2 用 py2neo 批量写入比逐条插入快一个量级CSV 准备好之后用 py2neo 读取并分批写入。逐条CREATE3000 个节点可能要半分钟而批量提交能在几秒内完成。核心代码from py2neo import Graph, Node, Relationship import pandas as pd graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) df_drug pd.read_csv(entities_drug.csv, dtypestr).fillna() df_disease pd.read_csv(entities_disease.csv, dtypestr).fillna() df_rel pd.read_csv(relations.csv, dtypestr).fillna() BATCH_SIZE 500 # 单个事务的写入条数过大容易爆内存 def load_entities(df, label): for i in range(0, len(df), BATCH_SIZE): batch df.iloc[i:iBATCH_SIZE] tx graph.begin() for _, row in batch.iterrows(): node Node(label, **row.to_dict()) tx.merge(node, label, name) # 按唯一键合并 tx.commit() print(f{label} 已写入 {i len(batch)} 条) load_entities(df_drug, Drug) load_entities(df_disease, Disease)这里关键参数是tx.merge(node, label, name)它先查有没有同名节点有就只更新属性没有才创建。配合第 2 章建的唯一约束能从根源上避免重复疾病、重复药物。BATCH_SIZE我一般设在 300 到 500 之间太小事务开销大太大一次提交可能把 JVM 堆顶爆第一次跑先用 200 稳妥确认内存没问题再加大。注意dtypestr是为了防止 pandas 把药品规格里的“0.5g”读成数字导致属性类型全乱。3.3 写完先验证数量别急着写问答数据导入后不要急着写问答接口先用几条 Cypher 确认图结构。常见做法是查每个标签的节点数和每种关系的条数再抽一条完整路径看方向是否正确MATCH (d:Disease) RETURN count(d) AS disease_count; MATCH (r:Drug)-[:TREATS]-(d:Disease) RETURN count(r) AS treat_count; MATCH (d:Drug)-[:TREATS]-(dis:Disease) WHERE d.name 阿司匹林 RETURN dis.name AS disease_name LIMIT 10;说明第二条 count 查询如果返回 0说明关系文件里的实体名没有跟实体表对上常见原因是同义词不一致。比如药典里写“乙酰水杨酸”CSV 里却写“阿司匹林”MERGE 时会按 name 找不到已有节点实际插入了一堆孤立节点。这时应该回去检查 relations.csv 里的两端名称统一成实体表里的 name而不是在查询时临时做别名匹配。抽路径查询能直接看到 TREATS 方向的箭头指向方向反了在这里一眼就能发现。4. 问答系统的三层解析意图识别、实体提取、查询生成图谱装好之后真正面向用户的是问答模块。我从不建议这类课程设计和中小型项目一上来就套 BERT 做语义匹配原因很简单医学问法相对固定规则加词典的方案可解释性强跑得也足够快等规则覆盖不了的 case 累积到一定量再换模型也不迟。问答链路就三件事识别意图、提取实体、拼接 Cypher。4.1 同义词表是实体提取的第一道防线先看实体提取。用户不会总按标准名提问“高血压”可能被说成“血压高”“原发性高血压”“阿司匹林”可能是“乙酰水杨酸”“APC”。我一般维护一个同义词映射表加载到 jieba 用户词典里再做一层别名归一化import jieba import jieba.posseg as pseg ALIAS_MAP { 乙酰水杨酸: 阿司匹林, 血压高: 高血压, 原发性高血压: 高血压, 肺部ct: CT检查, } def build_user_dict(): for entity_name in ALIAS_MAP.values(): jieba.add_word(entity_name, freq2000, tagn) for alias in ALIAS_MAP.keys(): jieba.add_word(alias, freq2000, tagn) def extract_entities(question): words pseg.cut(question) candidates [] for word, flag in words: if flag.startswith(n) and len(word) 2: # 名词候选 candidates.append(word) # 归一化别名转标准名 normalized [ALIAS_MAP.get(w, w) for w in candidates] return normalized # 后面再做实体类型匹配说明freq2000是给自定义词一个较高词频让 jieba 优先把它们切成完整词而不是把“乙酰水杨酸”切成“乙酰 水杨酸”。flag.startswith(n)只保留名词候选过滤掉“什么”“哪些”这类疑问词。注意这里只是候选集合最终能不能当成实体还要拿去和 Neo4j 里的节点做比对不能在常识上只凭词性下结论。4.2 意图识别用模板映射表比多个 if-else 好维护意图识别我的做法是维护一张“意图 → 关系 → 问句模板”的配置表代码里只做顺序匹配。直接看实现INTENT_PATTERNS [ { intent: drug_treats_disease, relation: TREATS, patterns: [{drug}治什么病, {drug}能治什么, {drug}对什么有效], order: 1, # 具体意图排前面 }, { intent: disease_has_symptom, relation: HAS_SYMPTOM, patterns: [{disease}什么症状, {disease}有哪些表现, {disease}会怎样], order: 2, }, { intent: disease_belongs_dept, relation: BELONGS_TO_DEPT, patterns: [{disease}挂什么科, {disease}看哪个科], order: 3, }, ] def classify_intent(question, entity): for item in sorted(INTENT_PATTERNS, keylambda x: x[order]): for pat in item[patterns]: test pat.replace({disease}, entity).replace({drug}, entity) if test in question.replace( , ): return item[intent], item[relation] return None, None说明这里没有用复杂的句法分析核心思路是“把问句中已识别出的实体替换回模板再做子串匹配”。order字段保证意图之间的优先级比如“阿司匹林能治什么病”同时命中的可能性很小但“高血压有什么症状”和“高血压看什么科”如果字面有重叠order 小的先匹配。模板覆盖不了的问法后续要做的就是加新模板而不是加 if 分支。维护成本和理解成本都低答辩时也讲得清楚。4.3 拼 Cypher 用参数化查询别用字符串拼接意图和实体都有了查询生成就是一层翻译。这里最需要小心的坑不要把实体名称直接拼进 Cypher 字符串。用户问句里可能存在引号、括号直接拼接轻则查询报错重则把整条查询语义搞坏。用参数化查询from py2neo import Graph graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) RELATION_CYPHER { TREATS: MATCH (d:{entity_label})-[r:TREATS]-(target) RETURN target.name AS answer, HAS_SYMPTOM: MATCH (d:{entity_label})-[r:HAS_SYMPTOM]-(target) RETURN target.name AS answer, BELONGS_TO_DEPT: MATCH (d:{entity_label})-[r:BELONGS_TO_DEPT]-(target) RETURN target.name AS answer, } def answer(question): entity extract_entities(question) # 简写实际取第一个匹配到图谱的实体 if not entity: return {status: no_entity, answer: 没识别到疾病或药物换个说法试试} intent, relation classify_intent(question, entity) if not intent: return {status: no_intent, answer: 这个问题我还不会换个问法问问看} entity_label detect_label(entity) # 判断 Drug / Disease / Symptom cypher RELATION_CYPHER[relation].replace({entity_label}, entity_label) results graph.run(cypher, entity_nameentity).data() if not results: return {status: no_result, answer: f图谱里有这个实体但找不到{relation}关系} answers [r[answer] for r in results[:5]] return {status: ok, answer: 、.join(answers)}说明graph.run(cypher, entity_nameentity)里的entity_name会作为参数传给 Cypher 中的$entity_name避免特殊字符干扰查询。detect_label用两个 MATCH 分别判断实体是 Drug 还是 Disease这里不贴全了核心是按节点 label 的 count 来决定把实体放进哪类节点。status字段不是给用户看的是给后面离线评测用的。把“没识别到实体”“没匹配到关系”“查询成功”拆成三种状态后面统计准确率才能知道失败到底失败在哪一环。5. 医学问答系统最常翻车的四个现场与排查路径这套系统我前前后后写过三轮最典型的坑基本集中在下面四个。每一条都按“现象 → 原因 → 解决”来写都是能直接照做的排查路径。5.1 实体链接总是把“感冒症状”当成疾病名现象用户问“感冒症状有哪些”系统把“感冒症状”整个识别成疾病实体查不到节点返回“没找到疾病”。原因jieba 词典里“感冒”和“症状”各自是词但组合起来“感冒症状”在自定义词典里被整体切出或者实体提取逻辑把所有两字以上的名词都当候选。解决实体候选提取后不要直接拿去查库先把候选词与图谱节点名做精确匹配匹配不上的再尝试把“症状”“检查”“科室”这类词尾去掉重试TAIL_STOP_WORDS [症状, 检查, 科室, 吃什么药, 怎么办] def normalize_entity(candidate): for tail in TAIL_STOP_WORDS: if candidate.endswith(tail) and len(candidate) len(tail): return candidate[:-len(tail)] return candidate这个函数加在extract_entities之后、查库之前能解决一半实体过宽的问题。剩余的问题靠图谱侧唯一约束兜底查不到就是查不到不要硬猜。5.2 方向定义没问题查询却返回空现象数据管道验证时(Drug)-[:TREATS]-(Disease)能查出结果问答接口里同样意图却总是空结果。原因问答模块的实体类型判断错了把“阿司匹林”识别成了 Disease于是执行了(Disease)-[:TREATS]-(target)方向反了自然为空。解决排查时不看上层代码先在 Neo4j Browser 里用反向查询确认数据确实存在MATCH (d)-[r:TREATS]-(target) WHERE d.name $entity_name RETURN labels(d) AS from_label, type(r) AS rel, labels(target) AS to_label再看返回的from_label是什么如果发现是 Disease 而不是 Drug问题不在查询生成而在实体分类。心跳法则是问答系统的空结果先区分“图谱里没这条关系”和“图谱里一查就有但查错了方向”这两者的排查路径完全不同。5.3 意图模板互相覆盖答案张冠李戴现象问“高血压做什么检查”返回的是“高血压有什么症状”的答案。原因模板匹配时没有去掉标点和空格或者两个模板有公共子串“什么症状”和“做什么检查”里的“什么”被错误命中。解决一是把模板替换成实体后做全串匹配而不是包含匹配二是在配置表里加 order 字段把包含“检查”“科室”这类特征词的意图排前面。我实际维护时还加了一个小约束模板禁止出现“什么”这种无信息量词开头必须包含一个能区分意图的行为动词比如“治”“挂”“做”“有”。5.4 大批量导入时 Neo4j 内存爆掉或卡死现象一次性导入 5 万条关系Neo4j 进程直接卡死日志里报堆内存溢出。原因一个事务提交的数据量太大或者事务里混合了大量 MERGE 操作导致每一条都要查索引。解决把数据分批提交每批 200 到 500 条提交后释放事务BATCH_SIZE 300 tx graph.begin() for idx, (start, rel_type, end) in enumerate(relation_triples): tx.run( MATCH (a {name:$s}), (b {name:$e}) MERGE (a)-[r:%s]-(b) RETURN r % rel_type, sstart, eend, ) if (idx 1) % BATCH_SIZE 0: tx.commit() tx graph.begin() # 旧事务已关闭必须新开注意这里%s拼接的只是关系类型名关系类型来自程序内部的固定枚举不是用户输入所以不会引入安全风险两端的节点名仍然用参数传递。如果还卡就去 Neo4j 的neo4j.conf里把dbms.memory.heap.max_size从默认值调大同时确认唯一约束已经建好否则 MERGE 每一条都要全表扫描一次。6. 把问答准确率测出来离线评测与验收技巧写一套自动化评测脚本是这套源码从“能跑”到“真能用”的分水岭。我一般准备一个 30 到 50 条的测试集合覆盖每个意图至少 5 条同时故意混入别名和无结果问法。测试集用 Python 直接构造便于统计test_cases [ {question: 阿司匹林治什么病, expect_intent: drug_treats_disease, expect_answer_contains: 发热}, {question: 乙酰水杨酸能治什么, expect_intent: drug_treats_disease, expect_answer_contains: 发热}, {question: 高血压有什么症状, expect_intent: disease_has_symptom, expect_answer_contains: 头晕}, {question: 偏头痛挂什么科, expect_intent: disease_belongs_dept, expect_answer_contains: 神经内科}, {question: 这个病会不会传染, expect_intent: None, expect_answer_contains: None}, ] def evaluate(test_cases): intent_hit, entity_hit, answer_hit 0, 0, 0 for case in test_cases: result answer(case[question]) if result[intent] case[expect_intent]: intent_hit 1 if result[status] ! no_entity: entity_hit 1 if case[expect_answer_contains] and case[expect_answer_contains] in result[answer]: answer_hit 1 return { intent_acc: intent_hit / len(test_cases), entity_acc: entity_hit / len(test_cases), answer_acc: answer_hit / len(test_cases), }跑完先看三个数意图准确率低说明模板覆盖不够实体准确率低说明词典和归一化有问题两者都正常但答案准确率低问题在 Cypher 生成或返回结果的截断逻辑。我习惯把这套评测脚本留着每次改完同义词表或模板就重跑一遍防止修了一个坑又带出另一个坑。这也是我拿到任何同类源码时做的第一件事不看花哨的界面先建测试集跑准确率再决定是否值得往里面加语义模型。希望帮到你。本文还有配套的精品资源点击获取