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)
这段代码的核心在于 chunk_document 函数，它负责将长文档拆解为语义块。注意这里使用了 MD5 哈希来生成 chunk_id，这样既能保证唯一性，又能在不同环境间保持一致。upsert_vector_store 则负责批量写入，实际运行时建议根据内存情况调整 batch 大小。
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(...)
在这个网络中，每个节点代表一个临床特征或诊断结果，边表示因果依赖关系。TabularCPD 定义了条件概率表，这是贝叶斯网络推理的基础。实际部署时，这些概率值通常来自历史数据统计或专家知识校准。通过 VariableElimination 等推理算法，我们可以计算给定症状下的疾病后验概率，从而辅助决策。