100天跟着CP学PostgreSQL+AI,第30天 :临床决策支持：知识图谱如何辅助医生诊断？

sql
-- 创建测试数据库
CREATE DATABASE medical_kg;
\c medical_kg
-- 安装pg_rdf扩展
CREATE EXTENSION pg_rdf;
-- 创建RDF存储表
CREATE TABLE medical_triples (
    subject TEXT,
    predicate TEXT,
    object TEXT,
    context TEXT DEFAULT 'medical_context'
);
-- 创建GIN索引加速查询
CREATE INDEX idx_medical_triples_gin ON medical_triples 
    USING gin (subject, predicate, object);
数据导入与预处理：
python
运行
import psycopg2
from rdflib import Graph, URIRef, Literal, Namespace
from tqdm import tqdm
# 连接数据库
conn = psycopg2.connect(
    dbname="medical_kg",
    user="postgres",
    password="your_password",
    host="localhost",
    port="5432"
)
cur = conn.cursor()
# 定义命名空间
MED = Namespace("http://example.org/medical#")
REL = Namespace("http://example.org/relations#")
# 构建示例知识图谱
g = Graph()
# 添加500种疾病
diseases = []
for i in range(1, 501):
    disease = URIRef(MED[f"disease_{i}"])
    diseases.append(disease)
    g.add((disease, URIRef(REL["name"]), Literal(f"疾病{i}")))
    # 为每种疾病添加2-5个症状
    symptom_count = random.randint(2, 5)
    for j in range(symptom_count):
        symptom = URIRef(MED[f"symptom_{i*100+j}"])
        g.add((symptom, URIRef(REL["name"]), Literal(f"症状{i*100+j}")))
        g.add((disease, URIRef(REL["hasSymptom"]), symptom))
    # 为每种疾病添加1-3个检查项目
    exam_count = random.randint(1, 3)
    for k in range(exam_count):
        exam = URIRef(MED[f"exam_{i*50+k}"])
        g.add((exam, URIRef(REL["name"]), Literal(f"检查项目{i*50+k}")))
        g.add((disease, URIRef(REL["requiresExam"]), exam))
# 批量导入到PostgreSQL
batch_size = 1000
triples = list(g)
for i in tqdm(range(0, len(triples), batch_size)):
    batch = triples[i:i+batch_size]
    insert_query = """
    INSERT INTO medical_triples (subject, predicate, object) 
    VALUES (%s, %s, %s)
    """
    cur.executemany(insert_query, [(str(s), str(p), str(o)) for s, p, o in batch])
conn.commit()
SPARQL 查询示例：
sql
-- 查询1：根据症状推断可能的疾病
WITH symptoms AS (
    SELECT object AS disease
    FROM medical_triples
    WHERE predicate = 'http://example.org/relations#hasSymptom'
      AND object IN (
          'http://example.org/medical#symptom_101', -- 发热
          'http://example.org/medical#symptom_202'  -- 咳嗽
      )
    GROUP BY object
    HAVING COUNT(*) >= 2 -- 同时具有这两个症状
)
SELECT 
    d.object AS disease_uri,
    dn.object AS disease_name,
    COUNT(e.object) AS required_exams
FROM symptoms s
JOIN medical_triples d ON s.disease = d.subject
JOIN medical_triples dn ON d.subject = dn.subject AND dn.predicate = 'http://example.org/relations#name'
LEFT JOIN medical_triples e ON d.subject = e.subject AND e.predicate = 'http://example.org/relations#requiresExam'
GROUP BY d.object, dn.object
ORDER BY required_exams DESC
LIMIT 10;
-- 查询2：计算疾病-症状关联强度
WITH symptom_counts AS (
    SELECT 
        s.object AS disease,
        COUNT(p.object) AS symptom_count
    FROM medical_triples s
    JOIN medical_triples p ON s.subject = p.subject AND p.predicate = 'http://example.org/relations#hasSymptom'
    WHERE s.predicate = 'http://example.org/relations#name'
    GROUP BY s.object
)
SELECT 
    d.object AS disease_name,
    sym.object AS symptom_name,
    COUNT(*)::float / sc.symptom_count AS association_strength
FROM medical_triples d
JOIN medical_triples ds ON d.subject = ds.subject AND ds.predicate = 'http://example.org/relations#hasSymptom'
JOIN medical_triples sym ON ds.object = sym.subject AND sym.predicate = 'http://example.org/relations#name'
JOIN symptom_counts sc ON d.subject = sc.disease
GROUP BY d.object, sym.object, sc.symptom_count
ORDER BY association_strength DESC
LIMIT 20;
查询性能优化：
sql
-- 创建复合GIN索引
CREATE INDEX idx_medical_triples_gin ON medical_triples 
    USING gin (subject, predicate, object);
-- 物化视图加速常用查询
CREATE MATERIALIZED VIEW disease_symptom_view AS
SELECT 
    d.object AS disease,
    s.object AS symptom
FROM medical_triples d
JOIN medical_triples s ON d.subject = s.subject 
WHERE d.predicate = 'http://example.org/relations#name'
  AND s.predicate = 'http://example.org/relations#hasSymptom';
-- 对物化视图创建索引
CREATE INDEX idx_disease_symptom ON disease_symptom_view (disease, symptom);