医学知识图谱构建实战：从文献表格到Neo4j数据库

┌─────────────────┐
│  医学文献表格    │
│ (Excel/CSV)     │
└────────┬────────┘
         │ 1. 数据清洗
         ▼
┌─────────────────┐
│  结构化数据      │
│ (三元组格式)     │
└────────┬────────┘
         │ 2. 实体识别(NER)
         ▼
┌─────────────────┐
│  实体+关系       │
│ (疾病,药物...)   │
└────────┬────────┘
         │ 3. 图谱建模
         ▼
┌─────────────────┐
│  Neo4j图数据库   │
│ (节点+关系)      │
└────────┬────────┘
         │ 4. Cypher查询
         ▼
┌─────────────────┐
│  智能问答/推理   │
│ (Python接口)     │
└─────────────────┘

# 拉取Neo4j镜像
docker pull neo4j:5.15.0

# 启动容器
docker run -d \
    --name neo4j-medical \
    -p 7474:7474 \
    -p 7687:7687 \
    -e NEO4J_AUTH=neo4j/medical123 \
    -v $HOME/neo4j/data:/data \
    -v $HOME/neo4j/logs:/logs \
    neo4j:5.15.0

# 访问 http://localhost:7474
# 用户名: neo4j
# 密码: medical123
```txt

# macOS
brew install neo4j

# Ubuntu
wget -O - https://debian.neo4j.com/neotechnology.gpg.key | sudo apt-key add -
echo 'deb https://debian.neo4j.com stable latest' | sudo tee /etc/apt/sources.list.d/neo4j.list
sudo apt-get update
sudo apt-get install neo4j

# 启动
neo4j start
```txt

# 创建虚拟环境
python3 -m venv medical_kg_env
source medical_kg_env/bin/activate

# 安装核心依赖
pip install py2neo==2021.2.4 \
            pandas==2.2.0 \
            spacy==3.7.2 \
            pyvis==0.3.2 \
            tqdm==4.66.1

# 下载中文医学NER模型
python -m spacy download zh_core_web_sm

from py2neo import Graph

# 连接Neo4j
graph = Graph("bolt://localhost:7687", auth=("neo4j", "medical123"))

# 测试连接
print("Neo4j版本:", graph.database.kernel_version)
print("连接成功！")

import pandas as pd

# 示例：疾病-药物治疗数据
disease_drug_data = pd.DataFrame([
    {'疾病': '2型糖尿病', '药物': '二甲双胍', '有效率': '85%', '不良反应': '胃肠道反应'},
    {'疾病': '2型糖尿病', '药物': '格列美脲', '有效率': '78%', '不良反应': '低血糖'},
    {'疾病': '高血压', '药物': '氨氯地平', '有效率': '82%', '不良反应': '下肢水肿'},
])

# 示例：疾病-症状数据
disease_symptom_data = pd.DataFrame([
    {'疾病': '2型糖尿病', '症状': '多饮', '出现率': '90%'},
    {'疾病': '2型糖尿病', '症状': '多尿', '出现率': '88%'},
    {'疾病': '2型糖尿病', '症状': '体重下降', '出现率': '60%'},
    {'疾病': '高血压', '症状': '头痛', '出现率': '70%'},
])

# 示例：疾病-检查数据
disease_check_data = pd.DataFrame([
    {'疾病': '2型糖尿病', '检查项目': '空腹血糖', '参考值': '≥7.0 mmol/L'},
    {'疾病': '2型糖尿病', '检查项目': 'HbA1c', '参考值': '≥6.5%'},
    {'疾病': '高血压', '检查项目': '血压测量', '参考值': '≥140/90 mmHg'},
])

# 数据清洗
def clean_medical_data(df):
    """
    清洗医学数据（统一格式、去重、标准化）
    """
    # 去除空格
    df = df.applymap(lambda x: x.strip() if isinstance(x, str) else x)
    
    # 去除重复行
    df = df.drop_duplicates()
    
    # 标准化百分比格式
    for col in df.columns:
        if df[col].dtype == 'object':
            df[col] = df[col].str.replace('％', '%')
    
    return df

# 执行清洗
disease_drug_data = clean_medical_data(disease_drug_data)
disease_symptom_data = clean_medical_data(disease_symptom_data)
disease_check_data = clean_medical_data(disease_check_data)

# 医学知识图谱Schema设计
MEDICAL_KG_SCHEMA = {
    'nodes': {
        'Disease': {  # 疾病节点
            'properties': ['name', 'alias', 'desc', 'category'],
            'label': '疾病'
        },
        'Drug': {  # 药物节点
            'properties': ['name', 'alias', 'desc', 'type'],
            'label': '药物'
        },
        'Symptom': {  # 症状节点
            'properties': ['name', 'desc'],
            'label': '症状'
        },
        'Check': {  # 检查项目节点
            'properties': ['name', 'desc', 'reference'],
            'label': '检查'
        },
        'SideEffect': {  # 不良反应节点
            'properties': ['name', 'severity'],
            'label': '不良反应'
        }
    },
    'relationships': {
        'HAS_SYMPTOM': {  # 疾病-症状
            'properties': ['probability'],
            'label': '症状'
        },
        'TREAT_BY': {  # 疾病-药物
            'properties': ['efficacy', 'evidence_level'],
            'label': '治疗'
        },
        'NEED_CHECK': {  # 疾病-检查
            'properties': ['necessity'],
            'label': '检查'
        },
        'CAUSE': {  # 药物-不良反应
            'properties': ['incidence'],
            'label': '引起'
        },
        'COMPLICATION': {  # 疾病-并发症
            'properties': ['risk'],
            'label': '并发'
        }
    }
}

    [症状] ←─HAS_SYMPTOM─── [疾病] ───TREAT_BY→ [药物]
                             │                    │
                    NEED_CHECK│                    │CAUSE
                             ↓                    ↓
                          [检查]              [不良反应]

from py2neo import Graph, Node, Relationship
from tqdm import tqdm

class MedicalKGBuilder:
    """
    医学知识图谱构建器
    """
    
    def __init__(self, uri="bolt://localhost:7687", auth=("neo4j", "medical123")):
        """
        初始化连接
        """
        self.graph = Graph(uri, auth=auth)
        print(f"✅ 已连接到Neo4j: {self.graph.database.kernel_version}")
    
    def clear_database(self):
        """
        清空数据库（谨慎使用）
        """
        self.graph.delete_all()
        print("⚠️  数据库已清空")
    
    def create_disease_nodes(self, disease_names):
        """
        创建疾病节点
        """
        print("\n📌 创建疾病节点...")
        for disease_name in tqdm(disease_names):
            # 检查节点是否存在
            existing = self.graph.nodes.match("Disease", name=disease_name).first()
            
            if not existing:
                # 创建新节点
                disease_node = Node(
                    "Disease",
                    name=disease_name,
                    alias="",
                    desc=f"{disease_name}相关信息",
                    category="未分类"
                )
                self.graph.create(disease_node)
    
    def create_drug_nodes(self, drug_names):
        """
        创建药物节点
        """
        print("\n💊 创建药物节点...")
        for drug_name in tqdm(drug_names):
            existing = self.graph.nodes.match("Drug", name=drug_name).first()
            
            if not existing:
                drug_node = Node(
                    "Drug",
                    name=drug_name,
                    alias="",
                    desc=f"{drug_name}相关信息",
                    type="未分类"
                )
                self.graph.create(drug_node)
    
    def create_symptom_nodes(self, symptom_names):
        """
        创建症状节点
        """
        print("\n🔴 创建症状节点...")
        for symptom_name in tqdm(symptom_names):
            existing = self.graph.nodes.match("Symptom", name=symptom_name).first()
            
            if not existing:
                symptom_node = Node(
                    "Symptom",
                    name=symptom_name,
                    desc=f"{symptom_name}症状"
                )
                self.graph.create(symptom_node)
    
    def create_check_nodes(self, check_names):
        """
        创建检查项目节点
        """
        print("\n🔬 创建检查项目节点...")
        for check_name in tqdm(check_names):
            existing = self.graph.nodes.match("Check", name=check_name).first()
            
            if not existing:
                check_node = Node(
                    "Check",
                    name=check_name,
                    desc=f"{check_name}检查",
                    reference=""
                )
                self.graph.create(check_node)
    
    def create_disease_drug_relationships(self, df):
        """
        创建疾病-药物治疗关系
        """
        print("\n🔗 创建疾病-药物关系...")
        for _, row in tqdm(df.iterrows(), total=len(df)):
            # 查找节点
            disease = self.graph.nodes.match("Disease", name=row['疾病']).first()
            drug = self.graph.nodes.match("Drug", name=row['药物']).first()
            
            if disease and drug:
                # 创建关系
                relationship = Relationship(
                    disease,
                    "TREAT_BY",
                    drug,
                    efficacy=row.get('有效率', ''),
                    evidence_level='高'
                )
                self.graph.merge(relationship, "Disease", "name")
                
                # 如果有不良反应，创建药物-不良反应关系
                if '不良反应' in row and row['不良反应']:
                    side_effect_name = row['不良反应']
                    
                    # 创建不良反应节点
                    side_effect = self.graph.nodes.match("SideEffect", name=side_effect_name).first()
                    if not side_effect:
                        side_effect = Node("SideEffect", name=side_effect_name, severity='轻度')
                        self.graph.create(side_effect)
                    
                    # 创建药物-不良反应关系
                    cause_rel = Relationship(drug, "CAUSE", side_effect, incidence='常见')
                    self.graph.merge(cause_rel, "Drug", "name")
    
    def create_disease_symptom_relationships(self, df):
        """
        创建疾病-症状关系
        """
        print("\n🔗 创建疾病-症状关系...")
        for _, row in tqdm(df.iterrows(), total=len(df)):
            disease = self.graph.nodes.match("Disease", name=row['疾病']).first()
            symptom = self.graph.nodes.match("Symptom", name=row['症状']).first()
            
            if disease and symptom:
                relationship = Relationship(
                    disease,
                    "HAS_SYMPTOM",
                    symptom,
                    probability=row.get('出现率', '')
                )
                self.graph.merge(relationship, "Disease", "name")
    
    def create_disease_check_relationships(self, df):
        """
        创建疾病-检查关系
        """
        print("\n🔗 创建疾病-检查关系...")
        for _, row in tqdm(df.iterrows(), total=len(df)):
            disease = self.graph.nodes.match("Disease", name=row['疾病']).first()
            check = self.graph.nodes.match("Check", name=row['检查项目']).first()
            
            if disease and check:
                # 更新检查节点的参考值
                check['reference'] = row.get('参考值', '')
                self.graph.push(check)
                
                # 创建关系
                relationship = Relationship(
                    disease,
                    "NEED_CHECK",
                    check,
                    necessity='必需'
                )
                self.graph.merge(relationship, "Disease", "name")
    
    def build_from_dataframes(self, disease_drug_df, disease_symptom_df, disease_check_df):
        """
        从DataFrame批量构建知识图谱
        """
        print("\n" + "="*60)
        print("🚀 开始构建医学知识图谱")
        print("="*60)
        
        # 提取所有实体
        diseases = set(disease_drug_df['疾病'].unique()) | \
                   set(disease_symptom_df['疾病'].unique()) | \
                   set(disease_check_df['疾病'].unique())
        
        drugs = set(disease_drug_df['药物'].unique())
        symptoms = set(disease_symptom_df['症状'].unique())
        checks = set(disease_check_df['检查项目'].unique())
        
        # 创建节点
        self.create_disease_nodes(diseases)
        self.create_drug_nodes(drugs)
        self.create_symptom_nodes(symptoms)
        self.create_check_nodes(checks)
        
        # 创建关系
        self.create_disease_drug_relationships(disease_drug_df)
        self.create_disease_symptom_relationships(disease_symptom_df)
        self.create_disease_check_relationships(disease_check_df)
        
        # 统计信息
        stats = self.get_statistics()
        print("\n" + "="*60)
        print("✅ 知识图谱构建完成！")
        print("="*60)
        print(f"📊 统计信息:")
        for key, value in stats.items():
            print(f"   {key}: {value}")
        
        return stats
    
    def get_statistics(self):
        """
        获取图谱统计信息
        """
        stats = {}
        
        # 节点数量
        stats['疾病节点'] = len(list(self.graph.nodes.match("Disease")))
        stats['药物节点'] = len(list(self.graph.nodes.match("Drug")))
        stats['症状节点'] = len(list(self.graph.nodes.match("Symptom")))
        stats['检查节点'] = len(list(self.graph.nodes.match("Check")))
        stats['不良反应节点'] = len(list(self.graph.nodes.match("SideEffect")))
        
        # 关系数量
        result = self.graph.run("MATCH ()-[r]->() RETURN count(r) as count").data()
        stats['总关系数'] = result[0]['count'] if result else 0
        
        return stats

# 使用示例
if __name__ == "__main__":
    # 初始化构建器
    builder = MedicalKGBuilder()
    
    # 可选：清空现有数据
    # builder.clear_database()
    
    # 构建知识图谱
    stats = builder.build_from_dataframes(
        disease_drug_data,
        disease_symptom_data,
        disease_check_data
    )

============================================================
🚀 开始构建医学知识图谱
============================================================

📌 创建疾病节点...
100%|████████████████████████████████████| 2/2 [00:00<00:00, 15.2it/s]

💊 创建药物节点...
100%|████████████████████████████████████| 3/3 [00:00<00:00, 18.5it/s]

🔴 创建症状节点...
100%|████████████████████████████████████| 4/4 [00:00<00:00, 20.1it/s]

🔬 创建检查项目节点...
100%|████████████████████████████████████| 3/3 [00:00<00:00, 17.8it/s]

🔗 创建疾病-药物关系...
100%|████████████████████████████████████| 3/3 [00:00<00:00, 12.3it/s]

🔗 创建疾病-症状关系...
100%|████████████████████████████████████| 4/4 [00:00<00:00, 14.7it/s]

🔗 创建疾病-检查关系...
100%|████████████████████████████████████| 3/3 [00:00<00:00, 13.2it/s]

============================================================
✅ 知识图谱构建完成！
============================================================
📊 统计信息:
   疾病节点: 2
   药物节点: 3
   症状节点: 4
   检查节点: 3
   不良反应节点: 2
   总关系数: 10

# 基本模式：(节点)-[关系]->(节点)

# 1. 查询所有疾病
MATCH (d:Disease) RETURN d

# 2. 查询疾病的症状
MATCH (d:Disease {name: "2型糖尿病"})-[:HAS_SYMPTOM]->(s:Symptom)
RETURN s.name as 症状

# 3. 查询疾病的治疗药物
MATCH (d:Disease {name: "2型糖尿病"})-[r:TREAT_BY]->(drug:Drug)
RETURN drug.name as 药物, r.efficacy as 有效率

# 4. 多跳查询：药物的不良反应
MATCH (d:Disease {name: "2型糖尿病"})-[:TREAT_BY]->(drug:Drug)-[:CAUSE]->(se:SideEffect)
RETURN drug.name as 药物, se.name as 不良反应

class MedicalKGQuery:
    """
    医学知识图谱查询接口
    """
    
    def __init__(self, graph):
        self.graph = graph
    
    def query_disease_symptoms(self, disease_name):
        """
        查询疾病的所有症状
        """
        cypher = """
        MATCH (d:Disease {name: $disease_name})-[r:HAS_SYMPTOM]->(s:Symptom)
        RETURN s.name as symptom, r.probability as probability
        ORDER BY r.probability DESC
        """
        
        results = self.graph.run(cypher, disease_name=disease_name).data()
        
        if not results:
            return f"未找到疾病'{disease_name}'的症状信息"
        
        return pd.DataFrame(results)
    
    def query_disease_treatments(self, disease_name):
        """
        查询疾病的治疗方案
        """
        cypher = """
        MATCH (d:Disease {name: $disease_name})-[r:TREAT_BY]->(drug:Drug)
        OPTIONAL MATCH (drug)-[c:CAUSE]->(se:SideEffect)
        RETURN drug.name as 药物, 
               r.efficacy as 有效率,
               collect(DISTINCT se.name) as 不良反应
        ORDER BY r.efficacy DESC
        """
        
        results = self.graph.run(cypher, disease_name=disease_name).data()
        
        if not results:
            return f"未找到疾病'{disease_name}'的治疗方案"
        
        return pd.DataFrame(results)
    
    def query_disease_checks(self, disease_name):
        """
        查询疾病的检查项目
        """
        cypher = """
        MATCH (d:Disease {name: $disease_name})-[:NEED_CHECK]->(c:Check)
        RETURN c.name as 检查项目, c.reference as 参考值
        """
        
        results = self.graph.run(cypher, disease_name=disease_name).data()
        
        if not results:
            return f"未找到疾病'{disease_name}'的检查项目"
        
        return pd.DataFrame(results)
    
    def query_drug_safety(self, drug_name):
        """
        查询药物的不良反应
        """
        cypher = """
        MATCH (drug:Drug {name: $drug_name})-[c:CAUSE]->(se:SideEffect)
        RETURN se.name as 不良反应, 
               se.severity as 严重程度,
               c.incidence as 发生率
        """
        
        results = self.graph.run(cypher, drug_name=drug_name).data()
        
        if not results:
            return f"未找到药物'{drug_name}'的不良反应信息"
        
        return pd.DataFrame(results)
    
    def query_similar_diseases(self, disease_name, similarity_threshold=0.5):
        """
        查询相似疾病（基于共同症状）
        """
        cypher = """
        MATCH (d1:Disease {name: $disease_name})-[:HAS_SYMPTOM]->(s:Symptom)<-[:HAS_SYMPTOM]-(d2:Disease)
        WHERE d1 <> d2
        WITH d2, count(s) as common_symptoms
        MATCH (d2)-[:HAS_SYMPTOM]->(all_symptoms:Symptom)
        WITH d2, common_symptoms, count(all_symptoms) as total_symptoms
        WITH d2, common_symptoms, total_symptoms, 
             toFloat(common_symptoms) / total_symptoms as similarity
        WHERE similarity >= $threshold
        RETURN d2.name as 相似疾病, 
               common_symptoms as 共同症状数,
               round(similarity * 100, 2) as 相似度百分比
        ORDER BY similarity DESC
        """
        
        results = self.graph.run(
            cypher, 
            disease_name=disease_name, 
            threshold=similarity_threshold
        ).data()
        
        if not results:
            return f"未找到与'{disease_name}'相似的疾病"
        
        return pd.DataFrame(results)
    
    def query_comprehensive_info(self, disease_name):
        """
        查询疾病的综合信息（症状+治疗+检查）
        """
        print(f"\n{'='*60}")
        print(f"疾病：{disease_name}")
        print(f"{'='*60}\n")
        
        # 症状
        print("📍 主要症状：")
        symptoms = self.query_disease_symptoms(disease_name)
        if isinstance(symptoms, pd.DataFrame):
            print(symptoms.to_string(index=False))
        else:
            print(symptoms)
        
        print("\n💊 治疗方案：")
        treatments = self.query_disease_treatments(disease_name)
        if isinstance(treatments, pd.DataFrame):
            print(treatments.to_string(index=False))
        else:
            print(treatments)
        
        print("\n🔬 检查项目：")
        checks = self.query_disease_checks(disease_name)
        if isinstance(checks, pd.DataFrame):
            print(checks.to_string(index=False))
        else:
            print(checks)
        
        print(f"\n{'='*60}\n")

# 使用示例
query_engine = MedicalKGQuery(builder.graph)

# 查询2型糖尿病的综合信息
query_engine.query_comprehensive_info("2型糖尿病")

============================================================
疾病：2型糖尿病
============================================================

📍 主要症状：
 symptom probability
    多饮         90%
    多尿         88%
 体重下降         60%

💊 治疗方案：
    药物 有效率        不良反应
二甲双胍   85%    [胃肠道反应]
格列美脲   78%      [低血糖]

🔬 检查项目：
 检查项目      参考值
 空腹血糖 ≥7.0 mmol/L
  HbA1c     ≥6.5%

============================================================

def symptom_to_treatment_path(symptoms_list):
    """
    根据症状推荐可能的疾病和治疗方案
    
    输入：['多饮', '多尿', '体重下降']
    输出：可能的疾病及治疗建议
    """
    cypher = """
    MATCH (s:Symptom)<-[:HAS_SYMPTOM]-(d:Disease)-[:TREAT_BY]->(drug:Drug)
    WHERE s.name IN $symptoms
    WITH d, drug, count(s) as matched_symptoms
    RETURN d.name as 疾病,
           matched_symptoms as 匹配症状数,
           collect(DISTINCT drug.name) as 可选药物
    ORDER BY matched_symptoms DESC
    LIMIT 5
    """
    
    results = builder.graph.run(cypher, symptoms=symptoms_list).data()
    return pd.DataFrame(results)

# 测试
symptoms = ['多饮', '多尿']
result = symptom_to_treatment_path(symptoms)
print("\n🔍 根据症状推断：")
print(result.to_string(index=False))

🔍 根据症状推断：
    疾病 匹配症状数                可选药物
2型糖尿病        2  [二甲双胍, 格列美脲]

def check_drug_interaction(drug_list):
    """
    检测多种药物是否有相互作用
    （需要预先在图谱中添加INTERACT_WITH关系）
    """
    cypher = """
    UNWIND $drugs as drug1_name
    UNWIND $drugs as drug2_name
    WITH drug1_name, drug2_name
    WHERE drug1_name < drug2_name
    MATCH (d1:Drug {name: drug1_name})-[r:INTERACT_WITH]-(d2:Drug {name: drug2_name})
    RETURN d1.name + ' + ' + d2.name as 药物组合,
           r.level as 相互作用等级,
           r.effect as 作用描述
    """
    
    results = builder.graph.run(cypher, drugs=drug_list).data()
    
    if not results:
        return "✅ 未检测到药物相互作用"
    
    return pd.DataFrame(results)

def find_shortest_path(start_entity, end_entity):
    """
    找出两个实体之间的最短关系路径
    
    示例：找出"多饮"症状到"二甲双胍"药物的路径
    """
    cypher = """
    MATCH path = shortestPath(
        (start {name: $start})-[*]-(end {name: $end})
    )
    RETURN [node in nodes(path) | labels(node)[0] + ':' + node.name] as 路径,
           [rel in relationships(path) | type(rel)] as 关系
    """
    
    results = builder.graph.run(cypher, start=start_entity, end=end_entity).data()
    
    if not results:
        return f"未找到从'{start_entity}'到'{end_entity}'的路径"
    
    return results[0]

# 测试
path = find_shortest_path("多饮", "二甲双胍")
print("\n🛤️  路径查询：")
print("节点路径:", " → ".join(path['路径']))
print("关系类型:", " → ".join(path['关系']))

🛤️  路径查询：
节点路径: Symptom:多饮 → Disease:2型糖尿病 → Drug:二甲双胍
关系类型: HAS_SYMPTOM → TREAT_BY

def find_central_drugs():
    """
    找出最核心的药物（治疗疾病种类最多）
    """
    cypher = """
    MATCH (drug:Drug)<-[:TREAT_BY]-(disease:Disease)
    WITH drug, count(DISTINCT disease) as disease_count
    ORDER BY disease_count DESC
    RETURN drug.name as 药物,
           disease_count as 治疗疾病数
    LIMIT 10
    """
    
    results = builder.graph.run(cypher).data()
    return pd.DataFrame(results)

# 可视化2型糖尿病及其所有关系
MATCH (d:Disease {name: "2型糖尿病"})-[r]-(n)
RETURN d, r, n

from pyvis.network import Network
import networkx as nx

def visualize_disease_network(disease_name, output_file="medical_kg.html"):
    """
    可视化疾病的知识子图
    """
    # 查询数据
    cypher = """
    MATCH (d:Disease {name: $disease_name})-[r]-(n)
    RETURN d, type(r) as rel_type, n
    """
    
    results = builder.graph.run(cypher, disease_name=disease_name).data()
    
    # 创建网络图
    net = Network(height="750px", width="100%", bgcolor="#222222", font_color="white")
    net.barnes_hut()
    
    # 添加节点和边
    added_nodes = set()
    
    for record in results:
        # 中心疾病节点
        if disease_name not in added_nodes:
            net.add_node(disease_name, label=disease_name, color='#FF6B6B', size=30, title="疾病")
            added_nodes.add(disease_name)
        
        # 关联节点
        node = record['n']
        node_label = list(node.labels)[0]
        node_name = node['name']
        
        if node_name not in added_nodes:
            # 根据节点类型设置颜色
            color_map = {
                'Drug': '#4ECDC4',
                'Symptom': '#FFE66D',
                'Check': '#95E1D3',
                'SideEffect': '#FF6B9D'
            }
            color = color_map.get(node_label, '#FFFFFF')
            
            net.add_node(
                node_name, 
                label=node_name,
                color=color,
                size=20,
                title=node_label
            )
            added_nodes.add(node_name)
        
        # 添加边
        rel_type = record['rel_type']
        net.add_edge(disease_name, node_name, title=rel_type, label=rel_type)
    
    # 保存
    net.show(output_file)
    print(f"✅ 可视化图谱已保存至: {output_file}")

# 生成可视化
visualize_disease_network("2型糖尿病")

def batch_create_nodes(node_type, data_list, batch_size=1000):
    """
    批量创建节点（性能优化）
    """
    cypher = f"""
    UNWIND $batch as row
    MERGE (n:{node_type} {{name: row.name}})
    SET n += row.properties
    """
    
    for i in range(0, len(data_list), batch_size):
        batch = data_list[i:i+batch_size]
        builder.graph.run(cypher, batch=batch)

-- 为常用查询字段创建索引
CREATE INDEX disease_name_index FOR (d:Disease) ON (d.name);
CREATE INDEX drug_name_index FOR (d:Drug) ON (d.name);
CREATE INDEX symptom_name_index FOR (s:Symptom) ON (s.name);

-- 查看现有索引
SHOW INDEXES;

-- 创建全文索引
CREATE FULLTEXT INDEX disease_fulltext_index
FOR (d:Disease)
ON EACH [d.name, d.desc, d.alias];

-- 使用全文搜索
CALL db.index.fulltext.queryNodes("disease_fulltext_index", "糖尿病")
YIELD node, score
RETURN node.name, score
ORDER BY score DESC;

import spacy
from spacy.matcher import Matcher

# 加载中文医学NER模型（需要训练）
nlp = spacy.load("zh_medical_ner_model")

# 或使用基于词典的匹配
def extract_medical_entities(text, medical_dict):
    """
    基于医学词典的实体识别
    """
    entities = []
    
    for entity_type, entity_list in medical_dict.items():
        for entity in entity_list:
            if entity in text:
                entities.append({
                    'text': entity,
                    'type': entity_type
                })
    
    return entities

# 医学词典示例
medical_dict = {
    'Disease': ['糖尿病', '高血压', '冠心病'],
    'Drug': ['二甲双胍', '阿司匹林', '氨氯地平'],
    'Symptom': ['头痛', '发热', '咳嗽']
}

medical-knowledge-graph/
├── data/
│   ├── disease_drug.csv
│   ├── disease_symptom.csv
│   └── disease_check.csv
├── src/
│   ├── kg_builder.py      # 图谱构建
│   ├── kg_query.py        # 查询接口
│   ├── ner_extractor.py   # 实体识别
│   └── visualizer.py      # 可视化
├── tests/
│   └── test_kg.py
├── requirements.txt
├── config.yaml
└── README.md

# 克隆仓库
git clone https://github.com/your-repo/medical-knowledge-graph
cd medical-knowledge-graph

# 启动Neo4j（Docker）
docker-compose up -d

# 安装依赖
pip install -r requirements.txt

# 构建知识图谱
python src/kg_builder.py --data ./data --config config.yaml

# 启动查询服务
python src/api_server.py

class DataValidator:
    """
    数据质量检查
    """
    
    @staticmethod
    def validate_entity(entity_name, entity_type):
        """
        验证实体名称有效性
        """
        if not entity_name or entity_name.strip() == '':
            raise ValueError(f"实体名称不能为空")
        
        if entity_type == 'Drug':
            # 药物名称规则检查
            if len(entity_name) > 20:
                raise ValueError(f"药物名称过长: {entity_name}")
        
        return True
    
    @staticmethod
    def validate_relationship(relationship_data):
        """
        验证关系数据完整性
        """
        required_fields = ['source', 'target', 'type']
        
        for field in required_fields:
            if field not in relationship_data:
                raise ValueError(f"缺少必需字段: {field}")
        
        return True

def incremental_update(new_data_df, last_update_time):
    """
    增量更新知识图谱（避免全量重建）
    """
    # 筛选新增数据
    new_records = new_data_df[new_data_df['update_time'] > last_update_time]
    
    print(f"检测到 {len(new_records)} 条新数据")
    
    # 只更新变化的部分
    for _, row in new_records.iterrows():
        # MERGE操作：存在则更新，不存在则创建
        cypher = """
        MERGE (d:Disease {name: $name})
        ON CREATE SET d.created_at = timestamp()
        ON MATCH SET d.updated_at = timestamp()
        SET d += $properties
        """
        
        builder.graph.run(cypher, name=row['name'], properties=row.to_dict())

# 备份Neo4j数据库
docker exec neo4j-medical neo4j-admin dump \
    --database=neo4j \
    --to=/backups/neo4j-$(date +%Y%m%d).dump

# 恢复
docker exec neo4j-medical neo4j-admin load \
    --from=/backups/neo4j-20241110.dump \
    --database=neo4j \
    --force

from flask import Flask, jsonify, request

app = Flask(__name__)
query_engine = MedicalKGQuery(builder.graph)

@app.route('/api/disease/<disease_name>/symptoms', methods=['GET'])
def get_disease_symptoms(disease_name):
    """
    API: 查询疾病症状
    """
    try:
        result = query_engine.query_disease_symptoms(disease_name)
        return jsonify({
            'success': True,
            'data': result.to_dict('records') if isinstance(result, pd.DataFrame) else []
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/query', methods=['POST'])
def custom_query():
    """
    API: 自定义Cypher查询
    """
    data = request.json
    cypher = data.get('cypher')
    params = data.get('params', {})
    
    try:
        results = builder.graph.run(cypher, **params).data()
        return jsonify({
            'success': True,
            'data': results
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)