医疗 AI 可信系统全栈实现：向量索引与贝叶斯网络

9.3 向量索引构建示例

文档进入向量库前，通常需要先经过清洗、切分、打标签和嵌入处理。下面展示一个最简流程，真实环境中可以替换为 Milvus 或 Qdrant SDK。

from dataclasses import dataclass
from typing import Iterable
import hashlib

@dataclass
class Chunk:
    chunk_id: str
    text: str
    metadata: dict

def chunk_document(doc_id: str, title: str, text: str, source_type: str) -> list[Chunk]:
    # 按段落切分，过滤空行
    parts = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks = []
    for i, part in enumerate(parts):
        # 生成唯一 ID，避免冲突
        cid = hashlib.md5(f"{doc_id}-{i}-{part[:100]}".encode()).hexdigest()
        chunks.append(
            Chunk(
                chunk_id=cid,
                text=f"[{title}]\n{part}",
                metadata={"doc_id": doc_id, "source_type": source_type, "seq": i},
            )
        )
    return chunks

def upsert_vector_store(chunks: Iterable[Chunk], embed_fn, store):
    rows = []
    for c in chunks:
        vec = embed_fn(c.text)
        # 将文本本身也存入 payload，方便后续回溯
        rows.append({"id": c.chunk_id, "vector": vec, "payload": c.metadata | {"text": c.text}})
    store.upsert(rows)

9.4 检索后的证据过滤

可信系统不能把 Top-k 检索结果直接交给 LLM。至少应做三步过滤：版本过滤、来源过滤和任务过滤。比如分诊任务应优先召回急诊路径与院内制度，而不是科研论文；药物风险任务应优先结构化说明书和高等级指南，而非社区帖子。经过过滤后，再做重排与证据回指，才能进入后续推理层。

9.5 结构化表示的真正价值

很多团队把结构化表示视为'加一道前处理'，但事实上它改变的是系统认知方式。没有它，Agent 面对的是一堆文本；有了它，Agent 面对的是一个带证据索引的状态空间。前者天然容易走向语言幻觉，后者才有可能接入概率图模型与不确定性度量。

10 核心算法与代码实现（二）：贝叶斯网络、风险阈值与 Agent 路由

10.1 用贝叶斯网络表示诊断状态

下面给出一个极简的胸痛场景示例。真实医疗系统当然远比这个复杂，但它足以说明'结论不是一句话，而是一组后验概率'。

from pgmpy.models import BayesianNetwork
from pgmpy.factors.discrete import TabularCPD
from pgmpy.inference import VariableElimination

model = BayesianNetwork([
    ("age_risk", "mi"),
    ("st_elevation", "mi"),
    ("troponin_high", "mi"),
    ("d_dimer_high", "pe"),
    ("tachycardia", "pe"),
    ("mi", "chest_pain"),
    ("pe", "chest_pain"),
])

cpd_age = TabularCPD("age_risk", 2, [[0.6], [0.4]])
cpd_st = TabularCPD("st_elevation", 2, [[0.85], [0.15]])
cpd_trop = TabularCPD("troponin_high", 2, [[0.8], [0.2]])
cpd_dd = TabularCPD("d_dimer_high", 2, [[0.7], [0.3]])
cpd_tachy = TabularCPD("tachycardia", 2, [[0.75], [0.25]])

cpd_mi = TabularCPD(
    "mi", 2,
    values=[
        [0.99, 0.85, 0.80, 0.30, 0.40, 0.10, 0.08, 0.01],
        [0.01, 0.15, 0.20, 0.70, 0.60, 0.90, 0.92, 0.99],
    ],
    evidence=["age_risk", "st_elevation", "troponin_high"],
    evidence_card=[2, 2, 2]
)

cpd_pe = TabularCPD(
    "pe", 2,
    values=[
        [0.97, 0.70, 0.50, 0.10],
        [0.03, 0.30, 0.50, 0.90],
    ],
    evidence=["d_dimer_high", "tachycardia"],
    evidence_card=[2, 2]
)

# 将 CPD 加入模型
model.add_cpds(cpd_age, cpd_st, cpd_trop, cpd_dd, cpd_tachy, cpd_mi, cpd_pe)

# 验证模型有效性
assert model.check_model()

# 进行推断
infer = VariableElimination(model)
query = infer.query(variables=["mi", "pe"], evidence={"age_risk": 1, "st_elevation": 1})
print(query)

在这个示例中，我们定义了节点之间的依赖关系，并填充了条件概率表（CPD）。实际运行时，VariableElimination 会计算给定证据下的后验概率分布。这种结构化的推理方式比单纯依赖 LLM 生成文本更可控，特别是在高风险的医疗决策场景中。