医学知识图谱的存储与查询：图数据库与向量检索的混合架构

医学知识图谱的存储与查询图数据库与向量检索的混合架构一、当医生搜索糖尿病肾病需要的不只是关键词匹配某三甲医院的临床决策支持系统上线时医生提出了一个让搜索引擎团队为难的需求输入二甲双胍 肾功能不全 禁忌系统应该知道肾功能不全和慢性肾脏病3期、eGFR30、肌酐清除率下降是同一个临床概念的不同表述并返回《二甲双胍临床应用专家共识》中关于eGFR30时禁用的相关段落。这不是关键词匹配能解决的问题。医学概念之间存在复杂的层级关系糖尿病 → 2型糖尿病 → 伴肾病的2型糖尿病、等价关系心肌梗死 MI 心梗 Heart Attack、关联关系药物-适应症、药物-禁忌症、疾病-并发症。这就是知识图谱的用武之地。但纯图数据库如Neo4j在以下场景中表现不佳当医生用自然语言描述症状最近一周走路时胸闷、气短需要找到知识图谱中最接近的症状和疾病时这是典型的语义匹配问题——需要向量检索。二、混合架构图结构承载医学逻辑向量检索承载语义匹配医学知识图谱的Schema设计以疾病为核心节点// Neo4j中创建医学知识图谱的核心节点和关系 // 疾病节点 CREATE (d:DiabetesType2:Disease { icd10_code: E11, name: 2型糖尿病, aliases: [T2DM, 非胰岛素依赖型糖尿病], embedding_id: emb_disease_001 // 关联到Milvus中的向量 }) // 药物节点 CREATE (m:Metformin:Drug { atc_code: A10BA02, name: 二甲双胍, generic_name: Metformin, embedding_id: emb_drug_042 }) // 症状节点 CREATE (s:Nausea:Symptom { name: 恶心, embedding_id: emb_symptom_118 }) // 检查指标节点 CREATE (l:eGFR:LabTest { loinc_code: 62238-1, name: 估算肾小球滤过率 }) // 关系 CREATE (m)-[:TREATS {evidence_level: 1A}]-(d) CREATE (m)-[:CONTRANDICATED_WHEN { condition: eGFR 30, level: ABSOLUTE }]-(l) CREATE (m)-[:HAS_SIDE_EFFECT {frequency: 5-10%}]-(s) CREATE (d)-[:HAS_COMPLICATION]-(:Disease {name: 糖尿病肾病})三、双引擎的查询融合实现from sentence_transformers import SentenceTransformer from pymilvus import Collection, connections from neo4j import GraphDatabase class MedicalKnowledgeHybridSearch: def __init__(self, neo4j_uri, milvus_host): self.neo4j GraphDatabase.driver(neo4j_uri) self.encoder SentenceTransformer( microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext ) connections.connect(hostmilvus_host, port19530) self.vector_collection Collection(medical_entities) self.vector_collection.load() def search(self, query: str, search_type: str auto) - dict: 统一的混合搜索入口 results {} if search_type in (auto, graph): results[graph] self._graph_search(query) if search_type in (auto, semantic): results[semantic] self._vector_search(query) if search_type auto: results[merged] self._merge_results(results) return results def _graph_search(self, query: str) - list: 基于知识图谱的结构化查询 try: with self.neo4j.session() as session: # 查询某药物的所有禁忌症和替代药物 result session.run( MATCH (drug:Drug {name: $name})-[:CONTRANDICATED_WHEN]-(cond) OPTIONAL MATCH (drug)-[:TREATS]-(disease:Disease) OPTIONAL MATCH (alternative:Drug)-[:TREATS]-(disease) WHERE alternative drug RETURN drug.name AS drug, collect(DISTINCT cond.name) AS contraindications, collect(DISTINCT disease.name) AS indications, collect(DISTINCT alternative.name) AS alternatives , namequery) return [record.data() for record in result] except Exception as e: raise GraphSearchException(f图搜索失败: {e}) def _vector_search(self, query: str, top_k: int 10) - list: 基于语义向量的相似度检索 try: query_vector self.encoder.encode(query).tolist() search_params { metric_type: IP, # Inner Product (余弦相似度) params: {nprobe: 16} } results self.vector_collection.search( data[query_vector], anns_fieldembedding, paramsearch_params, limittop_k, output_fields[entity_type, entity_name, description] ) return [ { entity_name: hit.entity.get(entity_name), entity_type: hit.entity.get(entity_type), score: hit.score, description: hit.entity.get(description) } for hit in results[0] ] except Exception as e: raise VectorSearchException(f向量检索失败: {e}) def _merge_results(self, results: dict) - list: 融合图谱和向量检索的结果 merged [] graph_entities set() # 图谱结果优先精确匹配 for item in results.get(graph, []): merged.append({**item, source: graph, priority: 1}) if drug in item: graph_entities.add(item[drug]) # 向量结果补充语义匹配 for item in results.get(semantic, []): if item[entity_name] not in graph_entities: merged.append({**item, source: semantic, priority: 2}) merged.sort(keylambda x: x[priority]) return mergedEmbedding的预生成与管理class MedicalEntityEmbedding: def __init__(self, encoder, milvus_collection): self.encoder encoder self.collection milvus_collection def embed_entity(self, entity: dict) - list: 为医学实体生成Embedding # 拼接实体的多维度文本表示 text f{entity[type]}: {entity[name]} if entity.get(synonyms): text | 别名: , .join(entity[synonyms]) if entity.get(definition): text | 定义: entity[definition] if entity.get(clinical_features): text | 临床表现: entity[clinical_features] embedding self.encoder.encode(text).tolist() return embedding def batch_embed_and_insert(self, entities: list): 批量生成Embedding并插入Milvus batch_size 100 for i in range(0, len(entities), batch_size): batch entities[i:i batch_size] embeddings [] insert_data [] for entity in batch: emb self.embed_entity(entity) embeddings.append(emb) insert_data.append({ entity_type: entity[type], entity_name: entity[name], neo4j_node_id: entity.get(neo4j_id, ), description: entity.get(definition, ), embedding: emb }) try: mr self.collection.insert(insert_data) self.collection.flush() except Exception as e: raise EmbeddingInsertException(f批量插入失败: offset{i}, e)四、医学知识图谱混合架构的四条红线红线一Embedding的领域适配。通用BERT如bert-base-chinese在医学文本上的语义理解有限。必须使用领域预训练模型PubMedBERT/BioBERT/MedicalBERT或在自己的医学语料上做微调否则心梗和心肌梗死的余弦相似度可能只有0.6而非期望的0.95。红线二知识更新的时效性。2024年ADA糖尿病指南调整了SGLT2抑制剂的推荐级别知识图谱必须在新指南发布后72小时内更新。这要求知识抽取管道的全自动化人工审校只能作为事后质量检查。红线三证据等级的存储与展示。不能只说二甲双胍治疗2型糖尿病——必须标注证据等级1A/1B/2A/2B/3/4。查询结果也需要按证据等级排序避免将低质量的个案报道与RCT结论并列展示。红线四知识冲突的处理。不同指南对同一问题的推荐可能矛盾如AHA和ESC对血压目标值的建议不同。知识图谱需要支持多版本并存——同一关系携带source和guideline_year属性前端展示时标注出处和时间。五、总结医学知识图谱的混合架构解决了精确匹配和语义理解的固有矛盾图数据库保证医学逻辑的严谨性药物A→禁忌症B→替代药物C是精确的图遍历向量检索保证自然语言查询的灵活性走路胸闷→劳力性呼吸困难→心功能不全是语义推导。两者通过实体ID作为桥梁关联——Neo4j中的每个实体节点都对应Milvus中的一个Embedding向量。查询时图结果和语义结果通过实体ID做去重融合既保证了准确性又提升了召回率。本文属于「行业场景与项目复盘」系列探索医学知识图谱的图数据库向量检索混合存储方案。