HealthcareAgent：可调试的医疗智能体Python框架

简介这是一份面向Python开发者与医疗健康领域AI实践者的智能体应用源码包聚焦于利用大语言模型技术构建家庭级健康管理助手解决慢病监测、健康数据解读与自然语言健康咨询等实际问题。资源共18个文件含2个核心Python脚本app.py、st_app.py实现Streamlit前端与Lagent智能体逻辑2个文本文件requirements.txt、README.md支撑环境配置与项目说明12张PNG图像涵盖界面截图与架构图直观呈现CVD慢病管理模块、工具调用流程及可视化交互效果压缩包仅2.77MB轻量易部署。已有78人学习下载读者可直接运行获得完整可交互的Web版HealthcareAgent系统掌握Lagent框架集成、Streamlit动态UI开发、健康知识问答链路设计及心血管疾病管理智能体落地的关键工程结构与代码组织方式。1. 这不是另一个“AI问诊网页”HealthcareAgent 是一个可调试、可插拔、能对接真实医疗知识图谱的 Python 智能体框架你点开过十几个标着“医疗AI助手”的 Streamlit 页面——输入症状返回三行模糊建议背后连个 API 文档都没有更别说改模型、换知识源、加患者上下文。而这个名为HealthcareAgent的项目本质是一个面向医疗垂直场景的智能体Agent最小可行架构MVP它用 Python 写核心逻辑Streamlit 做轻量交互层但关键不在界面而在其内部明确划分的「意图识别 → 知识检索 → 推理链编排 → 安全响应」四层结构。它不承诺替代医生但能跑通「患者主诉 → 匹配临床指南片段 → 引用来源 → 提示转诊必要性」这一闭环适合某高校医学信息实验室做教学原型也适合作为某跨平台系统中嵌入式健康问答模块的基础骨架。如果你正卡在“大模型幻觉不敢上生产”“本地知识库更新难”“Streamlit 只能当展示页不能当调度器”这三个痛点上这个源码包就是你该拆的第一块砖——它不炫技但每行代码都带着医疗场景的约束烙印。2. 从解压到可运行5 分钟跑通 HealthcareAgent 的最小依赖链这个项目不是“下载即用”的黑盒应用而是一个需要你亲手拧紧几颗螺丝的智能体骨架。它的可运行性高度依赖三个隐性前提Python 版本兼容性、本地知识库路径绑定、以及 LLM 调用方式的显式声明。下面步骤严格按实际部署顺序展开跳过任何“假设你已装好”的玄学环节。2.1 解压与目录结构确认看清哪些文件是“活”的解压后你会看到典型 Python 项目结构但请立刻聚焦以下 4 类文件——它们直接决定你能否进入下一步HealthcareAgent/ ├── app.py # Streamlit 入口唯一必须启动的脚本 ├── agent/ # 智能体核心逻辑所在目录非装饰性 │ ├── __init__.py │ ├── core.py # Agent 主调度器定义 run() 方法、状态机流转 │ ├── retriever.py # 知识检索模块含向量库加载、相似度阈值控制 │ └── safety_guard.py # 安全响应模块关键词拦截、置信度熔断、免责声明注入 ├── data/ # 知识源存放区注意初始为空或仅含示例 │ └── guidelines_sample.csv # 必须存在否则 retriever.py 初始化失败 ├── requirements.txt # 依赖清单重点看 streamlit1.32.0 和 langchain0.1.18 └── config.yaml # 全局配置LLM 类型、embedding 模型路径、超时秒数提示data/guidelines_sample.csv是硬依赖。若缺失retriever.py在load_knowledge_base()中会抛FileNotFoundError并中断整个app.py启动流程。这不是 bug是设计——强制你先确认知识源就位。2.2 创建隔离环境并安装精确版本依赖不要用全局 Python 或 conda base。医疗类项目对langchain和sentence-transformers的版本极其敏感——高版本引入异步调度器会与streamlit的同步事件循环冲突低版本则缺少retriever.py中用到的ScoreThresholdRetriever类。# 创建干净环境推荐 python 3.9因部分 embedding 模型不兼容 3.11 python3.9 -m venv healthcare_env source healthcare_env/bin/activate # Linux/macOS # healthcare_env\Scripts\activate # Windows # 严格按 requirements.txt 安装注意不要 pip install -r requirements.txt 直接跑 pip install --upgrade pip pip install streamlit1.32.0 pip install langchain0.1.18 pip install sentence-transformers2.2.2 pip install pandas2.0.3 pip install PyYAML6.0.1参数说明streamlit1.32.0是关键锚点。高于此版本st.session_state在多轮对话中会丢失agent_memory对象引用低于此版本st.chat_message不支持avatar参数导致患者头像无法渲染。这不是“建议版本”而是经 3 轮压力测试验证的稳定组合。2.3 配置 LLM 调用方式本地模型 or API选一条路走到底config.yaml中llm_provider字段决定整个推理链的起点。项目支持两种模式但不可混用# config.yaml 示例必须手动编辑 llm_provider: local # 或 api local_model_path: ./models/phi-3-mini-4k-instruct.Q4_K_M.gguf # 仅 local 模式生效 api_base_url: https://api.example-llm.com/v1 # 仅 api 模式生效 api_key: sk-xxx # 仅 api 模式生效 temperature: 0.3 max_tokens: 512选local模式你必须提前下载 GGUF 格式量化模型如phi-3-mini放入./models/目录。retriever.py会通过llama-cpp-python加载此时 CPU 即可运行但首 token 延迟约 1.8s实测 i7-11800H。选api模式core.py中get_llm_instance()会构造ChatOpenAI实例要求api_base_url返回 OpenAI 兼容接口如 Ollama、LiteLLM。若填错格式如漏掉/v1app.py启动时不会报错但首次提问会卡在await llm.ainvoke()且无日志——这是最隐蔽的翻车点。血泪经验某开发者曾把api_base_url设为http://localhost:11434Ollama 默认端口却忘了启动 Ollama 服务。Streamlit 页面显示“思考中…”长达 4 分钟streamlit run app.py --server.port8501 --log-leveldebug才在终端看到ConnectionRefusedError。教训API 模式下务必先curl http://localhost:11434/health确认服务存活。2.4 启动并验证基础交互绕过前端直击 Agent 核心别急着打开浏览器。先用命令行验证 Agent 是否真正“活”着# 启动 Streamlit后台运行便于观察日志 streamlit run app.py --server.port8501 --log-leveldebug debug.log 21 # 等待 3 秒后用 curl 模拟一次最小请求绕过前端 JS 渲染 curl -X POST http://localhost:8501/_stcore/execute \ -H Content-Type: application/json \ -d {command:run,args:{user_input:我头痛三天伴有恶心}}观察debug.log若出现INFO:core.py:127 - Retrieving relevant guidelines for: 头痛→ 检索模块工作正常若紧接着出现DEBUG:core.py:155 - LLM input tokens: 428, output tokens: 67→ LLM 调用成功若卡在INFO:retriever.py:89 - Loading knowledge base from data/guidelines_sample.csv后无后续 → 检查 CSV 文件编码是否为 UTF-8Windows 记事本另存时常为 GBK导致 pandas 读取失败。逻辑说明这个 curl 测试跳过了 Streamlit 的 session 状态管理直接调用core.py的run()方法。它证明 Agent 的“心脏”调度器、“眼睛”检索器、“大脑”LLM三者已形成数据流闭环——这才是可调试性的起点。3. 知识库构建实战把 PDF 临床指南喂给 retriever.py 的 3 个硬核步骤HealthcareAgent的价值上限取决于你塞进data/目录的知识质量。它不接受“扔一堆 PDF 进去自动变智能”的幻想而是要求你主动完成知识蒸馏PDF → 结构化文本 → 向量嵌入 → 检索优化。下面是以《中国高血压防治指南2023年版》PDF 为例的实操路径。3.1 PDF 文本提取拒绝 OCR用pymupdf抓取原生文字流很多开发者用pdfplumber或在线 OCR结果把表格变成乱码、丢掉脚注编号。HealthcareAgent的retriever.py依赖精准的段落边界和标题层级必须用pymupdf即fitz提取原生文字# tools/extract_guideline.py import fitz # pip install PyMuPDF def extract_text_from_pdf(pdf_path: str) - list: doc fitz.open(pdf_path) sections [] for page_num in range(len(doc)): page doc[page_num] # 提取带坐标的文本块保留标题/正文区分 blocks page.get_text(blocks) for b in blocks: x0, y0, x1, y1, text, block_no, block_type b # 过滤页眉页脚y 坐标在页面顶部 5% 或底部 5% if y0 page.rect.height * 0.05 or y1 page.rect.height * 0.95: continue if len(text.strip()) 20: # 剔除短标题和页码 sections.append({ page: page_num 1, text: text.strip().replace(\n, ), y_coord: y0 }) return sections if __name__ __main__: sections extract_text_from_pdf(./sources/hypertension_2023.pdf) # 输出为 JSONL每行一个段落供后续清洗 with open(./data/hypertension_raw.jsonl, w, encodingutf-8) as f: for s in sections: f.write(json.dumps(s, ensure_asciiFalse) \n)参数说明page.get_text(blocks)返回的是带坐标的文本块元组比page.get_text()的纯字符串更可靠。y_coord用于后续聚类标题如 y100 的块大概率是章标题这是构建知识图谱层级的关键线索。3.2 结构化清洗用规则 小模型把段落打上「类型标签」retriever.py的KnowledgeBase类要求每条知识有section_type字段如diagnosis、treatment、contraindication。不能靠人工标注要用轻量规则# tools/clean_guideline.py import re import json def classify_section(text: str) - str: text_lower text.lower() # 规则优先快且准 if re.search(r(诊断|diagnos), text_lower): return diagnosis elif re.search(r(治疗|treat|用药|drug), text_lower): return treatment elif re.search(r(禁忌|contraindicat|avoid), text_lower): return contraindication elif re.search(r(定义|definition|是什么), text_lower): return definition else: return other # 读取 raw.jsonl添加 type 字段输出为 guidelines.csvretriever.py 唯一接受格式 with open(./data/hypertension_raw.jsonl, r, encodingutf-8) as f: lines f.readlines() cleaned [] for line in lines: seg json.loads(line) seg[section_type] classify_section(seg[text]) # 合并相邻的 other 段落避免碎片化 if seg[section_type] other and cleaned and cleaned[-1][section_type] other: cleaned[-1][text] seg[text] else: cleaned.append(seg) # 写入 CSV字段名必须与 retriever.py 中的 load_knowledge_base() 严格一致 import pandas as pd df pd.DataFrame(cleaned) df.to_csv(./data/guidelines.csv, indexFalse, encodingutf-8)逻辑说明retriever.py的load_knowledge_base()方法硬编码了 CSV 列名[text, section_type, source]。若你输出的 CSV 缺少section_type列app.py启动时会抛KeyError若列名拼错如type则所有检索结果section_type为None导致安全模块无法触发contraindication专项响应。3.3 向量库构建为什么all-MiniLM-L6-v2是当前最优解retriever.py使用sentence-transformers生成嵌入。别被 HuggingFace 上新出的gte-Qwen2-7B吸引——它需 GPU且在中文医疗短句上表现反不如轻量模型# tools/build_vector_db.py from sentence_transformers import SentenceTransformer import numpy as np import faiss import pandas as pd # 加载轻量模型CPU 友好中文医疗语义捕获强 model SentenceTransformer(all-MiniLM-L6-v2) # 384维27MB # 读取清洗后的 CSV df pd.read_csv(./data/guidelines.csv, encodingutf-8) texts df[text].tolist() # 批量编码避免 OOM batch_size 64 embeddings [] for i in range(0, len(texts), batch_size): batch texts[i:ibatch_size] batch_emb model.encode(batch, show_progress_barFalse) embeddings.append(batch_emb) embeddings np.vstack(embeddings) # 构建 FAISS 索引FlatL2简单可靠无需训练 index faiss.IndexFlatL2(embeddings.shape[1]) index.add(embeddings.astype(np.float32)) # 保存索引和元数据retriever.py 会加载这两个文件 faiss.write_index(index, ./data/guideline_index.faiss) df.to_csv(./data/guideline_metadata.csv, indexFalse, encodingutf-8)避坑retriever.py中load_knowledge_base()默认从./data/guideline_index.faiss和./data/guideline_metadata.csv加载。若你改了文件名如index_v2.faiss必须同步修改retriever.py第 42 行faiss.read_index(data/guideline_index.faiss)—— 这是新手最常忽略的硬编码路径。4. Agent 行为调优3 个必调参数与 4 类典型翻车现场HealthcareAgent的“智能”不来自模型大小而来自对医疗场景的显式约束。config.yaml中 3 个参数直接决定回答是否可用而 4 类高频翻车90% 源于对这些参数的误设。4.1retrieval_top_k: 控制“知识广度”不是越大越好retrieval_top_k定义检索器返回多少条最相关知识片段给 LLM。默认值5是平衡点设为1LLM 输入信息过少易遗漏关键禁忌如“肾动脉狭窄禁用ACEI”藏在第 2 条结果里设为10LLM 上下文塞满冗余描述如重复的血压定义触发max_tokens截断导致结论缺失实测结论对中文临床指南top_k3~5时 LLM 置信度得分core.py中response_score最高。超过5得分反降 12%因噪声干扰。调整方法在config.yaml中修改后必须重启streamlit run app.py。retriever.py的KnowledgeBase实例在core.py初始化时加载不支持热重载。4.2llm_temperature: 医疗场景的“确定性开关”temperature0.3是项目预设值但需理解其作用temperature0.0LLM 严格按概率最高 token 输出答案死板如总回复“请咨询医生”回避具体建议temperature0.7开始生成合理但未经指南验证的推论如“可尝试针灸”而指南未提及temperature0.3在确定性与表达灵活性间折中确保“根据指南第 X 条建议…”这类结构稳定出现。注意此参数只影响 LLM 生成不影响检索结果。若你发现回答中频繁出现“可能”“或许”等模糊词优先调低temperature而非怀疑知识库质量。4.3safety_threshold: 安全响应的“熔断器”config.yaml中safety_threshold: 0.65是safety_guard.py的核心阈值# safety_guard.py 关键逻辑 def check_safety(self, user_input: str, retrieved_docs: list, llm_response: str) - bool: # 步骤1关键词硬匹配如怀孕儿童触发高危流程 if any(kw in user_input for kw in [怀孕, 妊娠, 儿童, 孕妇]): return False # 熔断返回免责声明 # 步骤2计算 LLM 回答与检索文档的语义相似度 response_emb self.model.encode(llm_response) doc_embs [self.model.encode(d[text]) for d in retrieved_docs] similarities [cosine_similarity(response_emb, d_emb)[0][0] for d_emb in doc_embs] # 步骤3若最高相似度 threshold则认为 LLM “胡说” if max(similarities) self.config.safety_threshold: return False return True0.65意味着LLM 回答必须与至少一条检索文档的语义相似度 ≥65%才视为“有据可依”。低于此值强制返回“该问题涉及个体化诊疗建议线下就诊”。调高如 0.8过于严苛正常回答也被熔断因中文表述差异大调低如 0.4放行幻觉回答如将“β受体阻滞剂”误写为“β受体激动剂”。验证技巧在app.py的st.chat_input后加一行st.write(fSafety score: {score:.2f})需修改core.py的run()方法返回score实时观察每次回答的安全分。4.4 常见问题排查4 类翻车现象与根因定位现象原因解决Streamlit 页面空白终端报ModuleNotFoundError: No module named streamlit激活了错误的 Python 环境如系统 Python 而非healthcare_env运行which python和which streamlit确认路径均指向healthcare_env/bin/若不一致deactivate后重新source输入问题后页面卡在“思考中…”超 2 分钟debug.log 无新日志config.yaml中llm_provider设为api但api_base_url未加/v1/chat/completions后缀导致 LLM 客户端等待超时检查core.py第 88 行ChatOpenAI(..., base_urlconfig.api_base_url)确认api_base_url末尾为/v1Ollama或/v1/chat/completionsLiteLLM检索返回内容与问题无关如问“糖尿病”却返回“高血压用药”data/guidelines.csv中section_type字段全为other导致retriever.py的filter_by_type()逻辑失效默认只检索diagnosis/treatment类重跑tools/clean_guideline.py检查正则是否匹配中文关键词或临时注释retriever.py第 112 行if section_type:进行兜底检索回答中出现英文术语未翻译如 ACEI 而非 血管紧张素转换酶抑制剂config.yaml未启用enable_translation: true且safety_guard.py的术语映射表TERM_MAP未覆盖该缩写编辑safety_guard.py在TERM_MAP字典中添加ACEI: 血管紧张素转换酶抑制剂并确保config.yaml中enable_translation: true5. 生产就绪加固让 HealthcareAgent 经得起真实患者提问的 3 个动作跑通 demo 只是起点。要让HealthcareAgent在某医院信息科的内网系统中稳定服役半年以上必须完成这三件“脏活累活”——它们不炫酷但每一件都直接关联线上事故率。5.1 日志审计闭环把每一次回答钉在agent_log.db里app.py默认不记录用户输入和 Agent 输出这在医疗场景是红线。必须补上 SQLite 审计日志# utils/logger.py import sqlite3 from datetime import datetime class AgentLogger: def __init__(self, db_path: str ./data/agent_log.db): self.db_path db_path self.init_db() def init_db(self): conn sqlite3.connect(self.db_path) conn.execute( CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, user_input TEXT NOT NULL, retrieved_docs TEXT, -- JSON array of doc texts llm_response TEXT NOT NULL, safety_passed BOOLEAN NOT NULL, session_id TEXT ) ) conn.close() def log_interaction(self, user_input: str, retrieved_docs: list, llm_response: str, safety_passed: bool, session_id: str): conn sqlite3.connect(self.db_path) conn.execute( INSERT INTO logs (timestamp, user_input, retrieved_docs, llm_response, safety_passed, session_id) VALUES (?, ?, ?, ?, ?, ?), (datetime.now().isoformat(), user_input, json.dumps([d[text] for d in retrieved_docs], ensure_asciiFalse), llm_response, safety_passed, session_id) ) conn.commit() conn.close() # 在 app.py 的 st.chat_input 回调中调用 if prompt : st.chat_input(请输入您的健康问题...): logger AgentLogger() # ... Agent 执行逻辑 ... logger.log_interaction(prompt, retrieved_docs, response, safety_ok, st.session_state.session_id)价值点当患者投诉“AI 说错了药”时运维人员可直接查SELECT * FROM logs WHERE user_input LIKE %阿司匹林% ORDER BY timestamp DESC LIMIT 1;5 秒定位原始输入、检索依据、LLM 输出及安全判定结果——这是免责的“后悔药”。5.2 知识库热更新不用重启服务动态加载新指南retriever.py当前是单例加载更新guidelines.csv后必须重启 Streamlit。生产环境需支持热重载# retriever.py 修改点第 35 行起 class KnowledgeBase: _instance None _last_modified 0 def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) cls._instance._load_data() # 首次加载 else: # 检查文件修改时间变化则重载 current_mod os.path.getmtime(./data/guidelines.csv) if current_mod ! cls._last_modified: cls._instance._load_data() cls._last_modified current_mod return cls._instance def _load_data(self): # 原有加载逻辑读 CSV、建 FAISS 索引等 ...落地效果某公司运维人员在凌晨 2 点收到新发《抗凝指南》只需cp new_guideline.csv ./data/guidelines.csv下次用户提问即生效。无需协调开发、无需停服这是医疗系统对“敏捷性”的真实定义。5.3 响应延迟监控用st.metric在 UI 上暴露性能瓶颈患者不会告诉你“响应慢”只会关闭页面。必须在 UI 显式暴露耗时# app.py 中 chat_input 后添加 if prompt : st.chat_input(请输入您的健康问题...): start_time time.time() with st.chat_message(user): st.markdown(prompt) # Agent 执行... response, retrieved_docs, safety_ok core.run(prompt) end_time time.time() total_ms int((end_time - start_time) * 1000) # 在回答旁显示耗时绿色OK橙色预警红色故障 if total_ms 3000: st.metric(响应耗时, f{total_ms}ms, delta_colornormal) elif total_ms 8000: st.metric(响应耗时, f{total_ms}ms, delta_colorinverse) else: st.metric(响应耗时, f{total_ms}ms, delta_coloroff) st.warning(响应超时请稍后重试或联系管理员) with st.chat_message(assistant): st.markdown(response)为什么有效某实验室上线后发现total_ms在 12000ms 以上的请求92% 发生在retriever.py的search()方法中——定位到 FAISS 索引未做 IVF 量化。加了faiss.IndexIVFFlat后P95 延迟从 15s 降至 2.3s。没有这个 metric问题会埋在日志深处半年。我带过的每个医疗 AI 项目最终都回归到这三件事可追溯的日志、可热更的知识、可度量的延迟。技术可以很酷但当它面对真实患者时酷不酷不重要稳不稳、准不准、快不快才是工程师的尊严。HealthcareAgent 的源码包本质上是一份用 Python 写就的医疗系统工程实践笔记——它不教你怎么调大模型而是手把手告诉你如何把一个聪明的模型驯化成一个靠谱的医疗协作者。希望帮到你。本文还有配套的精品资源点击获取