心内科RAG与多智能体协同：构建可追溯的智能诊断系统

简介：本资源面向医疗人工智能方向的研究者、算法工程师与心内科临床信息化开发者，提供一套基于检索增强生成（RAG）与多智能体协同架构的心内科疾病智能诊断系统开发项目。项目围绕心电图、超声心动图、生化指标等临床数据，结合文献与案例库检索，实现个性化诊断建议生成，并通过多智能体分工完成数据整理、风险评估与治疗方案生成，适合作为医疗AI课题复现、毕业设计或工程落地的参考方案。资源包共23个文件，包含11个json病例与配置数据、10个pdf心内科指南及文献、1个txt说明与1个md文档，压缩包约12.58MB，目录按语料、病例与输出示例分层组织。已有68人学习下载，读者可获取完整项目结构、病例样例、医学知识语料与输出示例，快速理解RAG与多智能体在医疗诊断中的协同实现路径。

1. 心内科诊断系统为什么需要 RAG 加多智能体协同

心内科门诊有个很现实的问题：一个主诉“胸闷三天”的患者，可能对应稳定型心绞痛、急性冠脉综合征、主动脉夹层、肺栓塞、胃食管反流甚至焦虑障碍。指南文本动辄上百页，检验指标几十项，影像报告又各说各话。单靠一个大模型直接问答，最常见的翻车是“编指南”——它会把 2018 年的旧推荐当成 2023 年更新，把禁忌症说成适应证。检索增强生成（RAG）解决的是“知识从哪来、引用可不可查”，多智能体协同解决的是“一个复杂诊断任务该拆给谁、谁对谁负责”。把这两件事拼起来，才是一个能落地的心内科疾病智能诊断系统：RAG 负责把指南、共识、药品说明书、院内路径变成可检索的知识底座，多智能体负责分诊、证据检索、鉴别诊断、用药核查、报告汇总各司其职。这套架构适合谁？适合手里有结构化病历和指南文档、想做一个可解释诊断辅助的工程团队，也适合想从“套壳问答”升级到“可追溯推理”的医疗 AI 开发者。下面按“知识底座怎么搭 → 智能体怎么分工 → 怎么跑通 → 坑在哪 → 怎么验证”的顺序讲透。

2. 心内科 RAG 知识底座：从指南 PDF 到可检索证据链

2.1 为什么心内科不能只靠向量库，要区分三类知识库

热搜里常有人问“rag知识库和结构知识库区分以及应用场景”，放到心内科特别典型。心内科知识大致分三类，混在一起存会直接拖垮召回质量。

第一类是非结构化指南文本，比如《稳定性冠心病诊断与治疗指南》《心力衰竭诊断和治疗指南》。这类适合切块后进向量库，用语义相似度召回。第二类是结构化知识，比如“肌钙蛋白 I 超过参考上限第 99 百分位 + 缺血症状 = 心肌损伤”，这是规则和阈值，适合放进图数据库或关系表，用精确查询而不是向量近似。第三类是半结构化院内路径，比如胸痛中心的时间节点表、用药剂量表，适合用表格解析后存成键值对。

我一般会做三层：向量库存指南原文块，图库（KG）存疾病-症状-检查-药物-禁忌的关系，关系库存阈值和剂量。检索时先走图库做实体对齐，再用向量库补原文证据，最后用关系库校验数值。这就是热词里说的“kg知识库”和“ontology rag”的落地形态——不是二选一，而是分工。

提示：心内科药物剂量和禁忌必须走结构化查询，不要让向量库“猜”，否则阿司匹林在消化道出血患者身上的禁忌会被语义相似度抹平。

2.2 用 Python 把指南 PDF 切成带元数据的证据块

下面这段是最小可跑的分块脚本，核心是保留来源、页码、章节，方便后面做引用回溯。

import re from pathlib import Path from langchain.text_splitter import RecursiveCharacterTextSplitter def clean_text(raw: str) -> str: # 去掉页眉页脚、多余空白，保留中文标点和数字 raw = re.sub(r'\n{2,}', '\n', raw) raw = re.sub(r'第\s*\d+\s*页', '', raw) return raw.strip() def split_guideline(pdf_text: str, source: str, page: int): splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 中文指南一段约 400-600 字，500 较稳 chunk_overlap=80, # 重叠 80 字，避免阈值被切断 separators=["\n\n", "\n", "。", "；", "，"] ) chunks = splitter.split_text(clean_text(pdf_text)) return [{ "text": c, "source": source, # 指南文件名 "page": page, # 页码，用于引用 "dept": "cardiology" # 科室标签，便于过滤 } for c in chunks]

逻辑说明：chunk_size=500是中文医学文本的经验值，太小会丢上下文，太大召回噪声多。chunk_overlap=80保证“肌钙蛋白超过第 99 百分位”这类跨句阈值不被切断。source和page是后面做证据链引用的关键，没有它们，RAG 输出就无法被医生核查。参数怎么改：如果指南表格多，先把表格单独抽出来走结构化解析，不要硬塞进这个分块器。

2.3 检索层要加科室过滤和阈值重排

纯向量召回在心内科容易把“心衰”和“肾衰”混在一起，因为两者都涉及水肿、呼吸困难。常见做法是检索时加元数据过滤dept=cardiology，再用一个交叉编码器做重排。

from sentence_transformers import CrossEncoder reranker = CrossEncoder('BAAI/bge-reranker-base') def retrieve(query, vector_store, top_k=20, final_k=5): # 先粗召回 20 条，再重排取前 5 coarse = vector_store.similarity_search(query, k=top_k, filter={"dept": "cardiology"}) pairs = [[query, doc.page_content] for doc in coarse] scores = reranker.predict(pairs) ranked = sorted(zip(coarse, scores), key=lambda x: x[1], reverse=True) return [doc for doc, _ in ranked[:final_k]]

top_k=20是粗召回数量，final_k=5是喂给智能体的证据条数。重排模型选中文医学语料微调过的效果更好，没条件就用通用重排，但要把阈值调高，宁可少召回也不要引入错误证据。这一步是 RAG 质量的分水岭，热词里说的“rag瓶颈”很多时候就卡在召回噪声上。

3. 多智能体协同：分诊、检索、鉴别、用药核查怎么分工

3.1 四个智能体的职责边界与消息协议

多智能体最容易犯的错是“大家都干一样的活”，结果重复检索、互相矛盾。我一般按诊断流程切四个角色：

• 分诊智能体：输入主诉和生命体征，输出紧急程度和初步方向（胸痛/心衰/心律失常）。
• 证据检索智能体：拿分诊结果去 RAG 底座检索指南和结构化知识，返回带来源的证据列表。
• 鉴别诊断智能体：基于证据做鉴别，输出候选诊断及支持/反对证据。
• 用药核查智能体：对候选诊断对应的药物做禁忌和相互作用检查。

它们之间用结构化消息传递，而不是自由文本，否则下游没法解析。消息格式建议固定字段：task、payload、evidence、confidence。

from dataclasses import dataclass, field from typing import List @dataclass class AgentMessage: sender: str task: str payload: dict evidence: List[dict] = field(default_factory=list) confidence: float = 0.0 # 分诊智能体输出示例 triage_msg = AgentMessage( sender="triage", task="chest_pain", payload={"urgency": "high", "direction": ["ACS", "aortic_dissection"]}, confidence=0.7 )

task字段决定下游走哪条链路，evidence只在检索后填充，confidence用于最终汇总时加权。这样设计的好处是每个智能体可以独立替换和测试，不会牵一发动全身。

3.2 用状态机编排智能体，避免无限循环

多智能体协同如果没有终止条件，会出现“鉴别智能体要求补证据 → 检索智能体再检索 → 又要求补证据”的死循环。常见做法是用一个轻量状态机控制流转，设置最大轮次。

MAX_ROUNDS = 3 def run_diagnosis(patient, agents): state = {"round": 0, "messages": [], "diagnosis": None} msg = agents["triage"].run(patient) state["messages"].append(msg) while state["round"] < MAX_ROUNDS: state["round"] += 1 evidence = agents["retrieval"].run(msg) diff = agents["differential"].run(evidence) if diff.confidence > 0.8 or state["round"] == MAX_ROUNDS: state["diagnosis"] = diff break msg = diff # 置信度不足，带着新问题再检索 return agents["medication"].run(state["diagnosis"])

MAX_ROUNDS=3是经验值，心内科急诊场景不建议超过 3 轮，否则响应时间不可接受。confidence > 0.8是提前终止条件，具体阈值要按科室调。这个状态机是“多智能体代码”里最该先写对的部分，比智能体本身聪明更重要。

3.3 智能体之间共享证据，而不是各查各的

很多实现让每个智能体自己调 RAG，结果同一份指南被检索五次，既慢又容易不一致。正确做法是检索智能体统一查一次，把证据放进共享上下文，其他智能体只读不查。

class SharedContext: def __init__(self): self.evidence = [] self.structured = {} def add_evidence(self, docs): self.evidence.extend(docs) def get_by_source(self, source): return [e for e in self.evidence if e["source"] == source]

SharedContext是整条链路的单一事实来源，鉴别诊断和用药核查都从这里取证据。这样最终报告里的每条结论都能追溯到具体指南页码，医生才愿意用。这也是“多智能体协同”和“单模型多轮对话”的本质区别：前者有明确的责任分工和共享证据，后者只是换个说法继续编。

4. 把系统跑起来：从环境到一次完整诊断的最小命令

4.1 本地环境与依赖安装

先说明，这里只讲本地可复现的最小环境，不涉及任何网络访问工具。Python 3.10 以上，主要依赖向量库、嵌入模型和智能体编排。

python -m venv venv source venv/bin/activate pip install langchain chromadb sentence-transformers \ pypdf fastapi uvicorn pydantic

chromadb用作本地向量库，sentence-transformers提供嵌入和重排，fastapi用来暴露诊断接口。如果机器没有 GPU，嵌入模型选小尺寸的，速度优先。安装完先跑一个嵌入测试，确认模型能正常加载再往下走。

4.2 灌入指南并建索引

import chromadb from sentence_transformers import SentenceTransformer client = chromadb.PersistentClient(path="./cardio_db") collection = client.get_or_create_collection("guidelines") embedder = SentenceTransformer('BAAI/bge-small-zh-v1.5') def index_chunks(chunks): for c in chunks: vec = embedder.encode(c["text"]).tolist() collection.add( ids=[f"{c['source']}_{c['page']}_{hash(c['text'])}"], embeddings=[vec], documents=[c["text"]], metadatas=[{"source": c["source"], "page": c["page"], "dept": c["dept"]}] )

PersistentClient把索引落盘，重启不丢。ids用来源加页码加哈希，保证不重复。metadatas里的dept就是前面检索过滤的依据。灌完后用一条测试查询验证召回，比如“肌钙蛋白升高 胸痛 诊断”，看返回的是不是相关指南段落。

4.3 一次完整诊断的调用链

def diagnose(patient_input: dict): ctx = SharedContext() triage = triage_agent(patient_input) docs = retrieve(triage.payload["direction"][0], collection) ctx.add_evidence([{"text": d.page_content, **d.metadata} for d in docs]) diff = differential_agent(ctx) med = medication_agent(diff, ctx) return { "triage": triage.payload, "diagnosis": diff.payload, "medication_check": med.payload, "evidence": ctx.evidence }

调用链清晰：分诊 → 检索 → 鉴别 → 用药核查，证据全程共享。返回结构里带evidence，前端可以展开每条结论对应的指南来源。参数上，patient_input至少要有主诉、生命体征、关键检验；缺项时要在分诊阶段标记为信息不足，而不是让下游硬猜。

5. 避坑与排查：心内科 RAG 多智能体最常见的五个翻车点

5.1 现象：诊断结果引用了不存在的指南页码

原因：分块时没保留页码，或者重排后把元数据丢了。解决：分块阶段强制写入page，检索返回时校验source和page是否存在，缺失的证据块直接丢弃，不允许进入鉴别环节。

5.2 现象：同一患者两次运行给出不同诊断

原因：向量检索有随机性，或者智能体温度参数没固定。解决：嵌入和重排设为确定性模式，智能体调用时temperature=0，并把检索结果缓存，同一输入走同一证据集。

5.3 现象：用药核查漏掉禁忌

原因：禁忌知识存在指南正文里，向量召回没命中。解决：把禁忌和相互作用单独抽成结构化表，用药核查智能体优先查表，查不到再走 RAG，并在报告里标注“未在结构化库中找到，仅供参考”。

5.4 现象：多智能体循环超过预期轮次

原因：鉴别智能体置信度一直上不去，反复要求补证据。解决：设置MAX_ROUNDS，并在每轮记录已检索过的查询，避免重复检索同一问题；超过轮次直接输出当前最优并标注不确定性。

5.5 现象：中文指南里的表格被切碎，阈值丢失

原因：PDF 表格被当成普通文本分块。解决：解析阶段先识别表格，用专门工具抽成结构化行，再决定是进关系库还是转成自然语言块，不要和正文混切。

6. 怎么验证这套系统值不值得继续投入

验证不要只看“回答像不像”，要看三个可量化指标。第一是证据命中率：构造 50 条心内科典型病例，人工标注应引用的指南段落，看系统召回是否覆盖。第二是禁忌漏检率：专门测有用药禁忌的病例，看用药核查智能体是否拦截。第三是一致性：同一病例重复跑 10 次，诊断结论是否稳定。

我一般会做一个小的回归集，用表格管理：

证据命中为“否”的病例要回查检索层，禁忌拦截为“否”的要回查结构化库覆盖度。这张表比任何主观评价都有用，因为它直接告诉你瓶颈在 RAG 还是智能体。

最后一个技巧：把每次诊断的完整消息链和证据快照落盘，出问题时能回放。心内科场景没有后悔药，只有可追溯的日志。我自己踩过最深的坑是早期没存证据快照，结果一个错误诊断查了两天才发现是分块把“禁忌”和“慎用”切到了两个块里。从那以后，任何进入鉴别环节的证据都必须带来源和页码，这条习惯帮我省了无数返工。希望帮到你。

本文还有配套的精品资源，点击获取