我要投稿

医疗AI大模型教程：把碎片化病程“一键熬成标准化首诊病史”：Agentic RAG + DPO/RLAIF 在医院的落地范式

发布日期：2025-09-06 07:50:54 浏览次数： 1541

作者：医工交叉AI大模型

微信搜一搜，关注“医工交叉AI大模型”

科研、项目、商务合作：nnhhce (注明来意,清北硕博团队专注于AI for Science大模型)

独家整理，盗用必究

❝
这不是“泛泛而谈”的大模型文章。我们只聚焦一个可立刻落地的细分场景：把住院/门诊的零散病程笔记，自动生成结构化的“首诊病史（H&P）文档，并直接映射为 FHIR 文档包，进入院内系统联审与归档。文末有可复用代码与30/60/90 天落地清单。（⚠️ 声明：本文仅用于学术与工程交流，不构成任何医疗建议；临床使用须经伦理与信息安全评审。）

01 为什么是“首诊病史自动结构化”？

医护写作负担重、模板多、风格杂，信息碎片化（查房记录、病程小结、化验回报、会诊意见散落在不同时间段的笔记里）。
只要把这些碎片熬成一碗“标准化首诊病史”（主诉、现病史、既往史、用药史、过敏史、体温脉搏血压等体征、问题列表），就能立刻：

减少重复录入，提升交接班效率；
将关键信息映射为 FHIR 结构化资源（Condition/MedicationStatement/AllergyIntolerance 等），纳入临床文档 Bundle和 H&P LOINC 34117-2 标识，便于互联互通与质控。

02 总体方案（6 步流水线）

输入： 某患者若干天内的病程/检验/会诊自由文本；输出： 一份结构化 H&P（FHIR Bundle + Composition + Condition/MedicationStatement/AllergyIntolerance 等），同时生成医生可校对的自然语言版本。

流水线：

去标识化（De-ID）：先做 PHI/PII 去标识，满足隐私与研究合规。
笔记分段/章节识别（Sectionizer）：识别“主诉/现病史/既往史”等章节。
Agentic RAG（第一次检索）：

混合检索（BM25 + 向量 bge-m3）→ 交叉重排（monoT5）→ 汇总上下文。

自我反思与纠偏（Self-RAG / CRAG）：

让模型先质疑再回答：评估检索质量、必要时二次检索或改写查询，并给出“证据-断言”对齐。

结构化映射（FHIR Mapping）：

把“问题列表→Condition”，“用药史→MedicationStatement”，“过敏史→AllergyIntolerance”，并封装为Clinical Document Bundle。

人机联审与对齐训练（DPO/ORPO/RLAIF）：

医师界面快速校对 → 形成偏好对（好/差）→ 用 DPO/ORPO 微调风格与格式；
安全合规与拒答策略可用 RLAIF/Constitutional AI 加强。

03 关键“前沿算法”如何落地到这个场景？

A. Agentic RAG：让检索变成“会思考的工具链”

混合检索：BM25 保证关键词召回；bge-m3 多语种/多粒度向量增强召回广度；再用 monoT5 交叉重排确保前列文档相关。
Self-RAG：模型先生成“反思标记”，判断是否需要更多证据或是否矛盾，再生成最终答案。
CRAG：当检索质量差时，触发“纠偏策略”（改写查询/扩大检索/分解再组装）。
工作流编排可用 LangGraph 等状态机式 Agent 框架实现可观测、可回滚的长链路流程。

B. 对齐训练：让生成更“懂医生”

DPO/ORPO：不用复杂的 RL 管线，仅凭偏好对就能稳定把输出对齐到院内书写风格与质控要点。
RLAIF/Constitutional AI：用模型生成的偏好或“宪法原则”进行安全对齐，降低人力成本并增强拒答与解释一致性。
Guardrails：在生产侧加可编程的安全护栏（话题控制、RAG 证据强制、越狱防护、PII 侦测）。

C. 评测：用“事实一致性”说话

公共语料可用 MIMIC-IV-Note（去标识临床笔记）做预研/验证。
FActScore/faithfulness：把输出拆成原子事实，核对是否被检索证据支持；同时校验 FHIR JSON Schema 合法性。
临床特定任务中，建议同时跟踪漏查关键要素率（如过敏史、现用药、现病史时间线）。

04 端到端最小可用实现（可改造成你院内 PoC）

❝
说明：以下代码仅演示架构与接口，省略实际大模型/外部服务调用；在院内部署时请接入贵院的私有大模型与知识库。

环境与数据准备（示意）

知识库（RAG 索引）：院内 H&P 模板、书写规范、质控要点、常见术语词典；
去标识化：建议离线管线（如 Philter/Presidio）→ 再进入向量化与模型阶段。

1) 章节识别 + 混合检索 + 重排

# pip install regex rank_bm25 faiss-cpu transformers # 根据环境调整
import re, math, json
from rank_bm25 import BM25Okapi

SECTION_HEADERS = {
    "chief_complaint": [r"主诉[:：]", r"Chief Complaint", r"CC[:：]"],
    "history_present_illness": [r"现病史[:：]", r"History of Present Illness", r"HPI[:：]"],
    "past_history": [r"既往史[:：]", r"Past Medical History", r"PMH[:：]"],
    "medications": [r"用药史[:：]", r"Medications?[:：]"],
    "allergies": [r"过敏史[:：]", r"Allerg(y|ies)[:：]"],
}

def sectionize(note: str):
    spans = []
    for key, pats in SECTION_HEADERS.items():
        for p in pats:
            for m in re.finditer(p, note, flags=re.I):
                spans.append((m.start(), key))
    spans.sort()
    sections = {}
    for i, (start, key) in enumerate(spans):
        end = spans[i+1][0] if i+1 < len(spans) else len(note)
        sections[key] = note[start:end].strip()
    return sections

# --- Hybrid retrieval (BM25 + dense placeholder) ---
def bm25_topk(query, corpus, k=8):
    tokenized = [doc.split() for doc in corpus]
    bm25 = BM25Okapi(tokenized)
    scores = bm25.get_scores(query.split())
    ranking = sorted(range(len(corpus)), key=lambda i: scores[i], reverse=True)[:k]
    return [(i, scores[i]) for i in ranking]

def dense_topk(query_emb, doc_embs, k=8):
    # 这里用占位的余弦相似度；实际请加载如 BAAI/bge-m3 向量模型。
    def cos(a, b):
        dot = sum(x*y for x,y in zip(a,b))
        na = math.sqrt(sum(x*x for x in a)); nb = math.sqrt(sum(x*x for x in b))
        return dot/(na*nb+1e-9)
    sims = [cos(query_emb, e) for e in doc_embs]
    ranking = sorted(range(len(doc_embs)), key=lambda i: sims[i], reverse=True)[:k]
    return [(i, sims[i]) for i in ranking]

def hybrid_rerank(query, passages, query_emb, doc_embs, k=8):
    bm = bm25_topk(query, passages, k*3)
    dn = dense_topk(query_emb, doc_embs, k*3)
    cand = {i:max(bs, ds) for (i,bs) in bm for (j,ds) in dn if i==j}
    # 交叉重排（monoT5）在此处调用实际重排模型：返回 top-k 索引
    top = sorted(cand.items(), key=lambda x: x[1], reverse=True)[:k]
    return [passages[i] for i,_ in top]

2) Self-RAG/CRAG 风格的“先批后答”

class AgentCritic:
    def critique(self, query, evidence_snippets, draft):
        """
        返回 {need_more_retrieval: bool, contradictions: list, missing_facts: list}
        真正实现中，可用带“反思标记”的 Self-RAG 或轻量评估器判断检索质量。
        """
        signals = {"need_more_retrieval": False, "contradictions": [], "missing_facts": []}
        for ev in evidence_snippets:
            if"过敏"in query and"无过敏"in draft and"青霉素过敏"in ev:
                signals["contradictions"].append("过敏史矛盾")
        signals["need_more_retrieval"] = len(signals["contradictions"])>0
        return signals

def generate_hpi(llm, sections, evidence):
    # 提示词省略：包含“引用证据写作”“逐条列出事实-证据映射”“无法确认要标记‘待核实’”
    draft = llm.generate(sections, evidence)
    critic = AgentCritic().critique(sections.get("history_present_illness",""), evidence, draft)
    if critic["need_more_retrieval"]:
        # 触发 CRAG 纠偏：改写查询/扩大检索，再生成
        more_evidence = evidence # 这里应调用新的检索
        draft = llm.generate(sections, more_evidence, critique=critic)
    return draft

3) FHIR 映射与文档打包

def to_fhir_bundle(patient_id, author_id, h_and_p_sections, problems, meds, allergies):
    composition = {
      "resourceType":"Composition",
      "status":"final",
      "type": {"coding":[{"system":"http://loinc.org","code":"34117-2","display":"History and physical note"}]},
      "subject":{"reference":f"Patient/{patient_id}"},
      "author":[{"reference":f"Practitioner/{author_id}"}],
      "title":"首诊病史（自动生成，已人工复核）",
      "section":[
        {"title":"主诉","text":{"status":"generated","div":h_and_p_sections.get("chief_complaint","")}},
        {"title":"现病史","text":{"status":"generated","div":h_and_p_sections.get("history_present_illness","")}},
        {"title":"既往史","text":{"status":"generated","div":h_and_p_sections.get("past_history","")}},
        {"title":"用药史","text":{"status":"generated","div":h_and_p_sections.get("medications","")}},
        {"title":"过敏史","text":{"status":"generated","div":h_and_p_sections.get("allergies","")}},
      ]
    }
    entries = [{"resource": composition}]
    for p in problems:
        entries.append({"resource":{
            "resourceType":"Condition",
            "clinicalStatus":{"coding":[{"system":"http://terminology.hl7.org/CodeSystem/condition-clinical","code":"active"}]},
            "code": {"text": p["name"]},
            "onsetDateTime": p.get("onset"),
        }})
    for m in meds:
        entries.append({"resource":{
            "resourceType":"MedicationStatement",
            "status":"active",
            "medicationCodeableConcept":{"text":m["name"]},
            "dosage":[{"text": m.get("text","")}]
        }})
    for a in allergies:
        entries.append({"resource":{
            "resourceType":"AllergyIntolerance",
            "clinicalStatus":{"coding":[{"system":"http://terminology.hl7.org/CodeSystem/allergyintolerance-clinical","code":"active"}]},
            "code":{"text":a["substance"]},
            "reaction":[{"description":a.get("reaction","")}]
        }})
    return {"resourceType":"Bundle","type":"document","entry":entries}