简介：这是一套面向Python开发者与医疗AI初学者的智能诊断问答系统实战项目，聚焦知识图谱构建与向量检索技术在健康医疗场景的落地应用，帮助用户掌握从医学知识建模、语义向量化到端到端问答服务部署的完整链路。资源包共188个文件，含71个核心Python脚本（覆盖NER识别、意图分类、图谱查询与Flask接口）、24张可视化图表（如疾病关系图、相似度热力图）、6个预训练模型文件（.pkl/.h5）、以及训练/测试/验证用CSV数据集和配套说明文档（MD/README），整体压缩后仅23.64MB，轻量易部署。已有497人学习下载，项目结构清晰，含可直接运行的bat启动脚本（如run_ner_service.bat）、分阶段训练数据（train/dev/test.csv）及词汇表文件（slot_vocab、intent_vocab），便于复现、调试与二次开发，是理解医疗领域知识增强型问答系统工程实践的优质参考样本。
1. 这不是另一个“AI问诊”Demo：它用Neo4j+BERT双引擎把症状-疾病-药物关系真正跑通了
你试过在医疗问答系统里输入“持续低热伴夜间盗汗三周，右上腹隐痛，体重下降5kg”，结果返回三条泛泛而谈的“可能为结核或肿瘤，请及时就医”？这不是模型能力不足，而是传统关键词匹配和单一路由检索根本无法建模医学知识间的多跳推理路径。本项目跳出了“问答=文本相似度”的惯性思维——它把《默克诊疗手册》《ICD-10编码规范》《药品说明书》等结构化与半结构化数据，先注入Neo4j图谱形成「症状→体征→检查指标→疾病→靶点→药物→禁忌」的6层关联网络；再用BERT微调后的双塔模型，将用户问题与图谱中每个节点（而非整篇文档）独立编码为768维向量；最后通过FAISS实现毫秒级向量召回+图遍历验证。实测在test.csv的327个真实临床问句中，Top-3准确率达89.3%，关键在于它不只返回“答案”，而是返回带置信度的推理链路：比如“盗汗→结核感染可能性↑→需查PPD/γ-干扰素释放试验→慎用糖皮质激素”。适合已有临床数据沉淀的三甲医院信息科、医疗AI初创团队，以及需要落地可解释性诊断辅助工具的科研组——它不替代医生，但能把医生从“翻指南→查文献→比对症状”的线性劳动中解放出来。
2. 知识图谱构建：从train.csv到Neo4j的实体-关系清洗与Schema设计
2.1 医学实体识别（NER）服务启动与字段映射逻辑
项目中的run_ner_service.bat并非简单调用预训练模型，而是针对中文医疗文本定制的三层NER流水线：
- 第一层：基于
spacy-transformers加载bert-base-chinese，识别原始文本中的粗粒度实体（如“肺结核”“阿司匹林”“ALT升高”）； - 第二层：用规则引擎校验实体边界——例如“左下腹痛”必须拆分为“左下腹”（解剖部位）+“痛”（症状），而“高血压性心脏病”需整体识别为疾病实体；
- 第三层：对接UMLS（Unified Medical Language System）概念ID，将“心肌梗死”映射到CUI:C0020373，确保跨数据源一致性。
执行run_ner_service.bat后，脚本会读取train.csv（含12,843条标注样本），输出ner_output.jsonl，每行格式为：
{ "text": "患者女，62岁，主诉反复上腹痛3月，伴反酸嗳气", "entities": [ {"start": 12, "end": 16, "label": "SYMPTOM", "text": "上腹痛"}, {"start": 22, "end": 26, "label": "SYMPTOM", "text": "反酸"}, {"start": 27, "end": 29, "label": "SYMPTOM", "text": "嗳气"} ], "cui_mapping": {"上腹痛": "C0023805", "反酸": "C0035001", "嗳气": "C0023804"} }提示：
train.csv字段必须包含text（原始问句）、label（人工标注的实体类型）、entity_text（实体原文）。若字段名不符，需修改ner_service/config.py中的CSV_COLUMN_MAP字典。
2.2 Neo4j Schema定义与关系抽取规则
图谱Schema严格遵循SNOMED CT临床术语标准，核心节点类型与关系如下：
| 节点类型 | 属性示例 | 关系类型 | 目标节点 | 业务约束 |
|---|---|---|---|---|
:Disease | cui: "C0020373",icd10: "I25.6" | HAS_SYMPTOM | :Symptom | 权重字段confidence: 0.92（来自文献支持度） |
:Symptom | cui: "C0023805",severity: "moderate" | TRIGGERS_BY | :Drug | 关系属性mechanism: "cholinergic_stimulation" |
:Drug | atc_code: "C01CA07",half_life: 3.5 | CONTRAINDICATED_FOR | :Disease | 添加evidence_level: "A"（GRADE证据等级） |
关系抽取不依赖纯规则，而是结合dev.csv中的专家标注样本训练BiLSTM-CRF模型。关键步骤：
- 执行
python ner_relation_extractor.py --input dev.csv --output neo4j_import.cypher - 生成的Cypher脚本包含批量创建语句：
// 创建疾病节点（去重） CREATE (d:Disease {cui: 'C0020373', name: '心肌梗死', icd10: 'I25.6'}) // 创建症状节点并建立关系（带置信度） CREATE (s:Symptom {cui: 'C0023805', name: '胸痛'}) CREATE (d)-[r:HAS_SYMPTOM {confidence: 0.97, source: 'UpToDate_2023'}]->(s)- 导入Neo4j前需执行
neo4j-admin import --nodes=nodes.csv --relationships=rels.csv，其中nodes.csv含id:cui:STRING,name:STRING,labels:STRING三列。
2.3 图谱质量验证：用Cypher查询检测常见医学逻辑错误
部署后必须运行以下验证查询，否则向量检索将因脏数据失效：
// 检查是否存在无出度的疾病节点（即未关联任何症状/检查/药物） MATCH (d:Disease) WHERE NOT (d)-[]->() RETURN d.name AS disease_name, count(*) AS orphan_count // 检查药物-疾病关系是否双向矛盾（如某药既治疗又禁忌同一疾病） MATCH (d:Disease)<-[:TREATS]-(drug:Drug)-[:CONTRAINDICATED_FOR]->(d) RETURN drug.name, d.name // 统计症状节点的平均关联疾病数（正常范围：3~12，低于2说明覆盖不足） MATCH (s:Symptom)-[r:HAS_SYMPTOM]->(d:Disease) RETURN s.name, count(d) AS disease_count ORDER BY disease_count ASC LIMIT 5若发现orphan_count > 0，需回溯train.csv中该疾病对应的问句是否缺失症状标注；若存在双向矛盾关系，需核查dev.csv中该药物的禁忌证标注是否错误。
3. 向量检索引擎：BERT双塔模型微调与FAISS索引构建
3.1 双塔模型结构设计与微调策略
项目未直接使用BERT原生模型，而是采用领域适配双塔架构：
- Query Tower：
bert-base-chinese+ 2层Transformer Block（参数量减少37%），输入用户问句，输出768维向量； - Knowledge Tower：独立编码图谱中每个节点（非整篇文档），对
:Disease节点拼接name + icd10 + description，对:Symptom节点拼接name + severity + duration，经相同BERT编码器后池化； - 损失函数：Triplet Loss + Hard Negative Mining，负样本从同疾病簇中采样（如“心肌梗死”的负样本选“心绞痛”而非“糖尿病”）。
微调脚本train_vector_model.py关键参数：
# config.py MODEL_NAME = "bert-base-chinese" MAX_LENGTH = 64 # 中文医疗问句平均长度，避免截断关键症状词 BATCH_SIZE = 16 LEARNING_RATE = 2e-5 # 比通用任务低10倍，防止灾难性遗忘 HARD_NEGATIVE_RATIO = 0.3 # 30% batch为难负样本训练数据来自train.csv与dev.csv，按8:2划分。微调后模型保存为./models/bert_dual_tower/，含pytorch_model.bin和config.json。
3.2 FAISS索引构建与向量化存储
向量检索性能取决于索引类型选择。本项目针对医疗图谱规模（约2.1万节点）采用IVF-PQ混合索引：
- 先用K-means聚类将向量空间划分为100个倒排文件（IVF）；
- 每个聚类内用乘积量化（PQ）压缩至64字节/向量（原768维→64字节，内存降低12倍）；
- 构建脚本
build_faiss_index.py核心代码：
import faiss import numpy as np from transformers import AutoModel, AutoTokenizer # 加载微调后的双塔模型 model = AutoModel.from_pretrained("./models/bert_dual_tower/") tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese") # 批量编码所有图谱节点（示例：编码1000个疾病节点） node_texts = ["心肌梗死 I25.6 急性冠脉综合征的一种...", "..."] # 来自Neo4j导出 inputs = tokenizer(node_texts, padding=True, truncation=True, max_length=64, return_tensors="pt") with torch.no_grad(): outputs = model(**inputs) vectors = outputs.last_hidden_state[:, 0, :].numpy() # [CLS]向量 # 构建IVF-PQ索引 dimension = vectors.shape[1] # 768 quantizer = faiss.IndexFlatIP(dimension) # 内积相似度 index = faiss.IndexIVFPQ(quantizer, dimension, 100, 32, 8) # 100聚类，32子向量，8bit精度 index.train(vectors) # 训练聚类中心 index.add(vectors) # 添加向量 faiss.write_index(index, "faiss_medical.index") # 保存索引文件注意：
faiss_medical.index需与node_id_mapping.pkl（节点ID到索引序号的映射字典）一同部署，否则召回结果无法对应Neo4j节点。
3.3 查询时向量检索与图谱路径融合
run_intent_recog_service.bat启动的服务接收HTTP请求，执行三阶段融合：
- 向量初筛：用户问句经Query Tower编码，用FAISS搜索Top-50相似节点；
- 图谱路径扩展：对每个初筛节点，在Neo4j中执行2跳遍历（如
MATCH (n)-[*1..2]-(m) WHERE id(n)=$node_id RETURN m），获取关联实体； - 重排序：按
confidence属性加权，公式为：final_score = 0.6 * vector_similarity + 0.3 * path_confidence + 0.1 * node_degree
其中node_degree为节点在图谱中的连接数，体现临床重要性。
实际请求示例（curl）：
curl -X POST "http://localhost:5000/query" \ -H "Content-Type: application/json" \ -d '{"question": "65岁男性，突发胸痛伴大汗，心电图ST段抬高"}' \ -o response.json响应中reasoning_path字段即为可解释的推理链：
{ "answer": "急性ST段抬高型心肌梗死（STEMI）", "reasoning_path": [ {"node": "胸痛", "type": "Symptom", "confidence": 0.94}, {"node": "大汗", "type": "Symptom", "confidence": 0.87}, {"node": "ST段抬高", "type": "ExamFinding", "confidence": 0.98}, {"node": "急性ST段抬高型心肌梗死", "type": "Disease", "confidence": 0.96} ] }4. 系统集成与生产级部署：Flask API封装与性能压测
4.1 Flask服务模块化设计与健康检查端点
app.py采用分层架构，避免单文件臃肿：
vector_search.py：封装FAISS查询与Neo4j路径扩展逻辑；graph_service.py：提供get_related_nodes(node_id, hops=2)等原子操作；cache_manager.py：用Redis缓存高频问句向量（TTL=3600秒），降低BERT推理负载。
关键健康检查端点/health返回结构化状态：
@app.route('/health') def health_check(): # 检查Neo4j连接 try: with driver.session() as session: session.run("RETURN 1").single() neo4j_status = "UP" except Exception as e: neo4j_status = f"DOWN: {str(e)}" # 检查FAISS索引加载 try: index = faiss.read_index("faiss_medical.index") faiss_status = "UP" except Exception as e: faiss_status = f"DOWN: {str(e)}" return jsonify({ "status": "OK" if neo4j_status == "UP" and faiss_status == "UP" else "DEGRADED", "components": { "neo4j": neo4j_status, "faiss": faiss_status, "redis_cache": "UP" if redis_client.ping() else "DOWN" } })部署时需配置.env文件：
NEO4J_URI=neo4j://localhost:7687 NEO4J_USER=neo4j NEO4J_PASSWORD=your_strong_password FAISS_INDEX_PATH=./faiss_medical.index REDIS_URL=redis://localhost:6379/04.2 压力测试：Locust脚本验证并发能力
使用Locust模拟真实问诊流量，脚本locustfile.py重点验证：
- 单节点QPS极限（目标≥120 QPS）；
- 长尾延迟（P95 < 800ms）；
- 内存泄漏（运行2小时后RSS增长<5%）。
核心测试逻辑：
from locust import HttpUser, task, between import random class MedicalQAUser(HttpUser): wait_time = between(1, 3) # 模拟用户思考时间 @task def query_symptom(self): # 从test.csv随机采样问句 questions = [ "儿童发热39度伴咳嗽流涕3天", "绝经后阴道出血，B超提示子宫内膜增厚", "服用华法林期间INR值升至8.2" ] question = random.choice(questions) self.client.post("/query", json={"question": question}) # 运行命令：locust -f locustfile.py --host=http://localhost:5000 --users 200 --spawn-rate 20压测结果需满足：
| 指标 | 合格阈值 | 实测值（200并发） |
|---|---|---|
| 平均响应时间 | < 450ms | 382ms |
| P95延迟 | < 800ms | 715ms |
| 错误率 | < 0.1% | 0.02% |
| CPU占用 | < 75% | 68% |
若P95超限，优先优化Neo4j查询——将MATCH (n)-[*1..2]-(m)改为指定关系类型：MATCH (n)-[:HAS_SYMPTOM|:TREATS|:CONTRAINDICATED_FOR*1..2]-(m)，避免全图扫描。
4.3 Docker Compose一键部署与资源隔离
docker-compose.yml实现服务解耦：
version: '3.8' services: web: build: . ports: ["5000:5000"] environment: - PYTHONUNBUFFERED=1 depends_on: [neo4j, redis] deploy: resources: limits: memory: 2G cpus: '1.0' neo4j: image: neo4j:5.16-enterprise volumes: - ./neo4j/data:/data - ./neo4j/plugins:/plugins environment: - NEO4J_AUTH=neo4j/your_password - NEO4J_dbms_memory_heap_max__size=2g - NEO4J_dbms_memory_pagecache_size=1g redis: image: redis:7-alpine command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru ports: ["6379:6379"]构建镜像前需在Dockerfile中预装关键依赖：
FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt && \ apt-get update && apt-get install -y libsm6 libxext6 && \ rm -rf /var/lib/apt/lists/* # 预编译FAISS（避免容器内编译耗时） RUN pip install faiss-cpu==1.7.4 COPY . . CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]提示：
requirements.txt必须锁定版本，尤其transformers==4.35.2、faiss-cpu==1.7.4、neo4j==5.16.0，避免PyPI版本更新导致兼容性问题。
5. 效果调优：从test.csv分析bad case并针对性修复
5.1 Bad Case归因分析表与修复优先级
对test.csv中23个Top-1错误样本进行人工归因，分类统计如下：
| 错误类型 | 样本数 | 典型案例 | 修复方案 | 优先级 |
|---|---|---|---|---|
| 实体歧义 | 9 | “小便黄”被识别为:Drug（因数据库中有“黄连”），实际应为:Symptom | 在NER规则中添加否定词过滤：“小便黄”→排除含“黄连”“黄芩”等药物名的上下文 | P0 |
| 关系缺失 | 7 | “糖尿病肾病”未关联:Drug节点，因train.csv中无该组合标注 | 用neo4j执行MATCH (d:Disease {name:"糖尿病肾病"}) CREATE (d)-[:TREATS]->(:Drug {name:"厄贝沙坦"})补全 | P1 |
| 向量漂移 | 5 | “心前区压榨感”与“胸骨后紧缩感”余弦相似度仅0.62（应>0.85） | 对train.csv中同义症状对做数据增强，生成100条变体问句（如“胸口像石头压着”） | P1 |
| 路径断裂 | 2 | “肝硬化门脉高压”未链接到:Varices（食管胃底静脉曲张） | 在图谱中添加[:CAUSES]->(:Complication)关系，并设置confidence: 0.99 | P0 |
P0级问题必须在上线前修复，否则直接影响核心场景准确率。
5.2 NER模型增量训练：用新样本快速迭代
当发现实体歧义问题时，无需重训整个模型，采用LoRA（Low-Rank Adaptation）微调：
from peft import get_peft_model, LoraConfig from transformers import AutoModelForTokenClassification base_model = AutoModelForTokenClassification.from_pretrained( "bert-base-chinese", num_labels=12 # 12个医疗实体类型 ) # 配置LoRA：仅训练注意力层的低秩矩阵 peft_config = LoraConfig( r=8, lora_alpha=16, target_modules=["query", "value"], # 只适配Q/V矩阵 lora_dropout=0.1, bias="none" ) model = get_peft_model(base_model, peft_config) # 使用新增的50条标注样本训练（epochs=3，batch_size=8） trainer.train() model.save_pretrained("./models/ner_lora_finetuned/")LoRA训练仅需原模型15%显存，2小时即可完成，且能保持原有知识不被覆盖。
5.3 向量检索的动态阈值调整技巧
FAISS默认返回固定Top-K结果，但医疗场景需根据问题确定性动态调整。在vector_search.py中加入置信度校准：
def search_with_dynamic_k(query_vector, base_k=10): # 获取Top-20相似度分数 scores, indices = index.search(query_vector.reshape(1, -1), 20) scores = scores[0] # 计算分数衰减率：若Top-3分数差距>0.2，说明问题模糊，扩大召回 if scores[0] - scores[2] > 0.2: k = min(25, len(scores)) # 最多召回25个 elif scores[0] - scores[4] < 0.05: # Top-5分数密集，说明高度确定 k = 5 else: k = base_k # 重新搜索指定k值 scores, indices = index.search(query_vector.reshape(1, -1), k) return scores[0], indices[0] # 调用示例 scores, indices = search_with_dynamic_k(user_query_vector)该技巧使模糊问句（如“最近不舒服”）召回更多候选，而明确问句（如“阿司匹林禁忌证”）精准返回3个核心节点，平衡召回率与推理效率。
本文还有配套的精品资源，点击获取
