"""ECG卡使用者问答智能体 (方案 A2) — Gradio + HF Inference Router. 部署环境: Hugging Face Space (Gradio SDK). 所需 Secrets: HF_TOKEN (Read 权限). 可选 Variables: MODEL_ID (默认 Qwen/Qwen2.5-7B-Instruct). """ from __future__ import annotations import json import os import re import traceback from difflib import SequenceMatcher from pathlib import Path from typing import Any, Dict, List, Tuple # spaces 必须在 gradio 之前 import;在非 ZeroGPU 环境下 spaces 是 no-op,不影响本地测试。 import spaces # noqa: F401 import gradio as gr try: # 允许无 huggingface_hub 时也能启动 UI(离线只跑本地降级回复) from huggingface_hub import InferenceClient except Exception: # pragma: no cover InferenceClient = None # type: ignore # ZeroGPU 启动检查要求至少一个 @spaces.GPU 装饰器出现在 Gradio 绑定的 handler 上。 # 本 Space 的 LLM 走 HF Inference Router(远端算),本地不需要 GPU; # 加装饰器只是通过启动校验,duration=1 表示最多占用 1 秒 GPU 配额, # 而且函数体内根本不触发 GPU 操作,实际配额消耗为 0。 # ------------------------------------------------------------------ # # 常量 / 配置 # # ------------------------------------------------------------------ # APP_DIR = Path(__file__).parent FAQ_PATH = APP_DIR / "faq_ecg.json" SAMPLE_PATIENTS_PATH = APP_DIR / "sample_patients.json" MODEL_ID = os.environ.get("MODEL_ID", "Qwen/Qwen2.5-7B-Instruct") HF_TOKEN = os.environ.get("HF_TOKEN", "") # 三级风险 forced-override 关键词(方案A2 附录) LEVEL_3_KEYWORDS = [ "胸痛", "剧烈胸痛", "压榨性", "晕厥", "昏倒", "昏迷", "意识不清", "失去意识", "呼吸困难", "喘不上气", "持续室速", "持续性室速", "抽搐", "口角歪斜", "半身不遂", ] LEVEL_3_TAIL = "\n\n🚨 当前情况可能存在紧急风险,请立即前往急诊或联系急救服务(120)。" # 输出模板固定四段 OUTPUT_SECTIONS = ["问题分析", "原因说明", "建议", "是否需要医生"] DOCTOR_MARKS = { "ok": "✅ 当前无需就诊", "warn": "⚠ 建议尽快咨询医生", "urgent": "🚨 请立即前往医院急诊", } # System prompt(方案A2 Reasoner 说明) SYSTEM_PROMPT_ZH = """你是"ECG卡使用者问答智能体"。角色:面向ECG卡使用者(以中老年慢性心血管病患者及家属为主)的中文问答助手。 【语气与人设】 - 专业、耐心、温和;避免使用"你必须""不要害怕"这类命令或恐吓; - 对老年用户友好:短句、简明、避免生僻缩写(必要时括号说明); - 不做医学诊断,不下结论,不替代医生;仅做知识解释、风险提示、正确使用引导。 【能力范围】 1) 设备使用指导(充电/贴电极/蓝牙/上传等); 2) 采集信号质量指导(接触、噪声、漂移等); 3) 心电基础知识与常见术语解释(HR/PR/QRS/QT/PVC/PAC/AF 等); 4) 报告结果的通俗解释; 5) 依据用户描述做三级风险初判(见下)。 【三级风险分类】 - Level 1(常规问题):日常使用、基础术语 → 结尾用 "✅ 当前无需就诊"。 - Level 2(需就诊):症状明显、指标异常、需要专业评估 → 结尾用 "⚠ 建议尽快咨询医生"。 - Level 3(急诊):持续胸痛、晕厥、呼吸困难、意识障碍、持续室速等 → 结尾用 "🚨 请立即前往医院急诊"。 【严格输出模板(硬性要求,不遵守视为错误输出)】 你的每一次回答必须**严格**按以下 4 段结构,顺序**不能颠倒**,标题**不能省略**、**不能改写**、**不能翻译**。 每段标题必须**逐字出现**且**独占一行**: - 第 1 段标题必须是: 问题分析 - 第 2 段标题必须是: 原因说明 - 第 3 段标题必须是: 建议 - 第 4 段标题必须是: 是否需要医生 结构示例(严格照抄标题格式;省略号是每段内容占位,你自己填): 问题分析 ...(30-80字,针对用户问题的直接分析或档案相关个人化解读) 原因说明 ...(简明的机制/原因,面向普通人) 建议 1) ... 2) ... 3) ... (2-5条可执行步骤,用 1) 2) 3) 分条) 是否需要医生 根据风险等级恰好填写下列之一(不要增加多余说明): ✅ 当前无需就诊 ⚠ 建议尽快咨询医生 🚨 请立即前往医院急诊 【错误示例(禁止)】 - ❌ 直接以内容开头,漏掉"问题分析"标题; - ❌ 用"分析:"、"分析如下"、"Analysis"等替代"问题分析"; - ❌ 4 段顺序颠倒(如把"是否需要医生"放在中间); - ❌ 在"是否需要医生"这段之后追加任何额外总结或补充段落。 【硬约束】 - 不下诊断结论,不推荐具体处方药剂量;涉及用药调整请建议咨询医生; - 未上传档案时不要编造历史数据; - 出现胸痛/晕厥/意识不清/呼吸困难等关键词一律按 Level 3 处理。 【个人化回答规则(极其重要)】 - 如果 system prompt 中出现 "参考此患者档案" 部分,该部分是本用户真实档案。 - 当用户提问包含"我的""我最近""我目前""我的记录""帮我看看"等第一人称表达,或询问自身指标/心率/用药/病史/趋势时,你必须优先引用"参考此患者档案"中的具体数字与事实,而不是给出通用科普。 - 引用示例:"根据您的档案,近8次ECG心率范围 X-Y bpm"、"档案显示您正在服用美托洛尔缓释片 47.5mg 每日一次"、"您近期有 1 次疑似房颤记录(2026-07-11)"。 - 未在档案中出现的数据严禁编造。 - 若档案与用户描述冲突,以档案为客观基线,同时提示用户可能的差异。""" # ------------------------------------------------------------------ # # 数据加载 # # ------------------------------------------------------------------ # def _load_json(path: Path) -> Any: if not path.exists(): return None with path.open("r", encoding="utf-8") as f: return json.load(f) FAQ_DB = _load_json(FAQ_PATH) or {"faqs": []} FAQS: List[Dict[str, Any]] = FAQ_DB.get("faqs", []) SAMPLE_PATIENTS_DB = _load_json(SAMPLE_PATIENTS_PATH) or {"patients": []} SAMPLE_PATIENTS: List[Dict[str, Any]] = SAMPLE_PATIENTS_DB.get("patients", []) SAMPLE_MAP: Dict[str, Dict[str, Any]] = { p["profile_id"]: p for p in SAMPLE_PATIENTS } # ------------------------------------------------------------------ # # FAQ 检索:关键词 + 字符相似度 # # ------------------------------------------------------------------ # def _normalize(text: str) -> str: return re.sub(r"\s+", "", text or "").lower() def _score_faq(query: str, faq: Dict[str, Any]) -> float: q = _normalize(query) if not q: return 0.0 # 关键词命中权重 kw_hits = 0 for kw in faq.get("keywords", []): if _normalize(kw) and _normalize(kw) in q: kw_hits += 1 kw_score = min(1.0, kw_hits * 0.4) # 问题相似度 sim = SequenceMatcher(None, q, _normalize(faq.get("question", ""))).ratio() # 关键词与查询串的最大相似度(处理短问题) kw_sim = 0.0 for kw in faq.get("keywords", []): kw_sim = max(kw_sim, SequenceMatcher(None, q, _normalize(kw)).ratio()) return 0.55 * kw_score + 0.30 * sim + 0.15 * kw_sim def retrieve_faqs(query: str, top_k: int = 3, min_score: float = 0.15) -> List[Dict[str, Any]]: if not query or not FAQS: return [] scored = [(f, _score_faq(query, f)) for f in FAQS] scored.sort(key=lambda x: x[1], reverse=True) picked = [f for f, s in scored if s >= min_score][:top_k] return picked def build_faq_context(faqs: List[Dict[str, Any]]) -> str: if not faqs: return "参考知识\n(未匹配到高相关度FAQ,请基于系统角色与通用心电常识回答。)" lines = ["参考知识(检索自内置FAQ,可参考其中的原因、建议、风险分级思路)"] for i, f in enumerate(faqs, 1): lines.append(f"\n[{i}] 问题:{f['question']}") lines.append(f" 类别:{f.get('category','?')} | 风险等级:L{f.get('risk_level','?')}") lines.append(f" 参考答复片段:\n{f.get('answer','')}") lines.append(f" 来源:{f.get('source','')}") return "\n".join(lines) # ------------------------------------------------------------------ # # 患者档案 → 压缩上下文 # # ------------------------------------------------------------------ # def _tail(seq, n): if not seq: return [] return list(seq)[-n:] def compress_profile(profile: Dict[str, Any]) -> Tuple[str, Dict[str, int]]: """按方案A2压缩规则生成 patient_context 字符串。返回 (text, stats)。""" if not profile: return "", {} lines: List[str] = [] stats: Dict[str, int] = {} pt = profile.get("patient") or {} if pt: keys = ["name", "age", "gender", "diagnosis"] vals = [f"{k}={pt.get(k)}" for k in keys if pt.get(k) is not None] lines.append("【患者基本信息】" + ", ".join(vals) if vals else "【患者基本信息】(未提供)") mh = profile.get("medical_history") or {} if mh: cond = ", ".join(mh.get("conditions") or []) or "(无)" events = ", ".join(mh.get("past_events") or []) or "(无)" lines.append(f"【既往病史】{cond}") lines.append(f"【重要事件】{events}") meds = profile.get("medications") or [] if meds: med_str = "; ".join( f"{m.get('name','?')} {m.get('dose','')} {m.get('freq','')}".strip() for m in meds ) lines.append(f"【当前用药】{med_str}") stats["medications"] = len(meds) ecg = profile.get("ecg_history") or [] if ecg: recent = _tail(ecg, 8) stats["ecg_total"] = len(ecg) stats["ecg_used"] = len(recent) lines.append(f"【近{len(recent)}次ECG(共{len(ecg)}次)】") for e in recent: lines.append( " - {t} HR={hr} PR={pr} QRS={qrs} QTc={qtc} " "节律={rhy} AI={ai} 信号={sq}".format( t=e.get("time", "?"), hr=e.get("HR", "?"), pr=e.get("PR", "?"), qrs=e.get("QRS", "?"), qtc=e.get("QTc", "?"), rhy=e.get("rhythm", ""), ai=e.get("ai_finding", ""), sq=e.get("signal_quality", ""), ) ) reports = profile.get("reports") or [] if reports: latest = reports[-1] stats["reports_total"] = len(reports) km = latest.get("key_metrics") or {} km_str = ", ".join(f"{k}={v}" for k, v in km.items()) or "(无)" content = (latest.get("content") or "")[:800] lines.append( "【最近报告】{t} 类型={typ} 关键指标: {km}\n内容摘要:{c}".format( t=latest.get("time", "?"), typ=latest.get("type", "?"), km=km_str, c=content, ) ) vitals = profile.get("vitals_log") or [] if vitals: recent_v = _tail(vitals, 5) v_str = "; ".join( f"{v.get('time','?')} 体重={v.get('weight_kg','?')}kg 血压={v.get('bp','?')}" for v in recent_v ) lines.append(f"【近{len(recent_v)}天生命体征】{v_str}") alerts = profile.get("alerts") or [] if alerts: recent_a = _tail(alerts, 3) stats["alerts_total"] = len(alerts) for a in recent_a: lines.append( "【预警】{t} L={lv} 内容:{c} 处置:{h}".format( t=a.get("time", "?"), lv=a.get("level", "?"), c=a.get("content", ""), h=a.get("handled", ""), ) ) else: lines.append("【预警】暂无历史预警记录") header = "参考此患者档案(仅供AI理解上下文;所有个人化建议仍需医生确认)" return header + "\n" + "\n".join(lines), stats def parse_profile_input(text: str) -> Tuple[Dict[str, Any] | None, str]: """解析用户粘贴的JSON。返回 (profile_or_None, feedback)。""" if not text or not text.strip(): return None, "已清空档案(未使用历史数据)。" try: obj = json.loads(text) except Exception as e: return None, f"❌ JSON解析失败:{e}" # 允许直接 profile 或 {"patients":[...]} 挑第一个 if isinstance(obj, dict) and "patients" in obj and isinstance(obj["patients"], list): if not obj["patients"]: return None, "❌ patients 数组为空。" return obj["patients"][0], "✅ 已加载 patients[0]。" if isinstance(obj, dict): return obj, "✅ 已加载档案。" return None, "❌ 无法识别的档案结构。" # ------------------------------------------------------------------ # # 输出后处理:补齐模板 / 强制 Level 3 # # ------------------------------------------------------------------ # def _has_urgent_keyword(text: str) -> bool: if not text: return False return any(k in text for k in LEVEL_3_KEYWORDS) def ensure_template(answer: str, forced_level: int | None = None) -> str: text = (answer or "").strip() # 补齐缺失段落 for section in OUTPUT_SECTIONS: if section not in text: text += f"\n\n{section}\n(本次回复未提供该段落,请以医生意见为准。)" # 强制 Level 3:必须包含 🚨 急诊结论,且末尾必须出现 "120" 字样(生命安全兜底) if forced_level == 3: # 移除已有的其它 doctor 结论 text = text.replace(DOCTOR_MARKS["ok"], "") text = text.replace(DOCTOR_MARKS["warn"], "") # 如果 LLM 没输出急诊标记,追加(带 120) if DOCTOR_MARKS["urgent"] not in text: text += LEVEL_3_TAIL # 若 LLM 已给出急诊标记但缺 120,单独补一条 120 提示(方案A2 硬要求) elif "120" not in text: text += LEVEL_3_TAIL return text.strip() # ------------------------------------------------------------------ # # HF Router 调用 # # ------------------------------------------------------------------ # def _call_llm(system_prompt: str, history: List[Tuple[str, str]], user_msg: str, max_tokens: int = 1024, temperature: float = 0.5, top_p: float = 0.9) -> str: if InferenceClient is None: return _fallback_answer(user_msg, reason="huggingface_hub 未安装。") if not HF_TOKEN: return _fallback_answer(user_msg, reason="未配置 HF_TOKEN(Space Settings → Secrets → HF_TOKEN)。") client = InferenceClient(token=HF_TOKEN, timeout=60) messages = [{"role": "system", "content": system_prompt}] for u, a in history or []: if u: messages.append({"role": "user", "content": u}) if a: messages.append({"role": "assistant", "content": a}) messages.append({"role": "user", "content": user_msg}) try: resp = client.chat_completion( model=MODEL_ID, messages=messages, max_tokens=max_tokens, temperature=temperature, top_p=top_p, ) # openai-style try: return resp.choices[0].message.content or "" except Exception: return str(resp) except Exception as e: return _fallback_answer(user_msg, reason=f"模型调用失败:{e}") def _fallback_answer(user_msg: str, reason: str = "") -> str: """离线/无 token 时的降级回复:如果匹配到 FAQ 就直接返回,否则给结构化占位。""" hits = retrieve_faqs(user_msg, top_k=1, min_score=0.3) if hits: base = hits[0].get("answer", "") note = f"\n\n(⚠ 本次因 {reason} 未能调用大模型,已直接使用最相关FAQ返回。)" if reason else "" return base + note return ( "问题分析\n暂时无法调用大模型进行个性化解释。\n\n" "原因说明\n" + (reason or "未知") + "\n\n" "建议\n1) 稍后再试;2) 如问题紧急请直接联系管理师或医生;" "3) 部署侧请检查 Space Secrets 中 HF_TOKEN 是否配置。\n\n" "是否需要医生\n" + DOCTOR_MARKS["warn"] ) # ------------------------------------------------------------------ # # Gradio 回调 # # ------------------------------------------------------------------ # def _chatbot_to_history(chatbot_list) -> List[Tuple[str, str]]: """gr.Chatbot list → [(user, assistant), ...]""" hist: List[Tuple[str, str]] = [] for item in chatbot_list or []: if isinstance(item, (list, tuple)) and len(item) == 2: hist.append((item[0] or "", item[1] or "")) return hist def on_load_sample(profile_id: str) -> Tuple[str, str]: """加载样本档案 → (json_text, preview)""" if not profile_id or profile_id not in SAMPLE_MAP: return "", "(未选择样本档案)" obj = SAMPLE_MAP[profile_id] return json.dumps(obj, ensure_ascii=False, indent=2), _profile_preview(obj) def on_clear_profile() -> Tuple[str, str]: return "", "已清空档案。" def on_apply_profile(text: str) -> Tuple[str, str]: profile, msg = parse_profile_input(text) if profile is None: return "", msg return _profile_preview(profile), msg def _profile_preview(profile: Dict[str, Any]) -> str: ctx, stats = compress_profile(profile) stat_line = " ".join(f"{k}={v}" for k, v in stats.items()) if stats else "(无压缩统计)" return f"档案已就绪。压缩统计:{stat_line}\n\n{ctx}" @spaces.GPU(duration=1) def respond(user_msg: str, chat_history: List[Any], profile_json_text: str) -> Tuple[List[Any], str]: """核心回复函数。返回 (new_chat_history, cleared_input)。 注:此函数被 @spaces.GPU 装饰仅为满足 ZeroGPU 启动检查, 实际计算不使用本地 GPU(全部远端 API)。 """ if not user_msg or not user_msg.strip(): return chat_history or [], "" try: history = _chatbot_to_history(chat_history) # 1) FAQ 检索 top_faqs = retrieve_faqs(user_msg, top_k=3) faq_ctx = build_faq_context(top_faqs) # 2) 患者档案压缩 profile, _ = parse_profile_input(profile_json_text or "") patient_ctx, _ = compress_profile(profile) if profile else ("", {}) # 3) 三级风险预判 forced_level = 3 if _has_urgent_keyword(user_msg) else None # 4) 组装 system prompt parts = [SYSTEM_PROMPT_ZH, faq_ctx] if patient_ctx: parts.append(patient_ctx) else: parts.append("参考此患者档案\n(用户当前未上传历史档案。请勿虚构历史数据,若被问及具体历史请友善提醒用户先在左侧上传档案。)") if forced_level == 3: parts.append("紧急信号提示\n(检测到用户描述中包含高危关键词,必须以 Level 3 处理,最后必须给出 🚨 前往急诊 结论。)") system_prompt = "\n\n".join(parts) # 5) 调用 LLM raw = _call_llm(system_prompt, history, user_msg) # 6) 后处理 final = ensure_template(raw, forced_level=forced_level) chat_history = (chat_history or []) + [[user_msg, final]] return chat_history, "" except Exception: err = "内部错误:\n```\n" + traceback.format_exc() + "\n```" chat_history = (chat_history or []) + [[user_msg, err]] return chat_history, "" def on_example_click(example: str) -> str: return example # ------------------------------------------------------------------ # # UI # # ------------------------------------------------------------------ # EXAMPLE_QUESTIONS = [ "ECG卡提示接触不好怎么办?", "报告显示PVC多是什么意思?", "心率120需要马上就医吗?", "电极贴多久换一次?", "房颤危险吗?", ] SAMPLE_CHOICES = [(p["display_name"], p["profile_id"]) for p in SAMPLE_PATIENTS] CSS = """ .gradio-container {max-width: 1200px !important;} #patient_json textarea {font-family: ui-monospace, monospace; font-size: 12px;} #preview {background: #F5F7FA; border-radius: 12px; padding: 12px;} """ def build_ui() -> gr.Blocks: with gr.Blocks(title="ECG卡使用者问答智能体", css=CSS, theme=gr.themes.Soft()) as demo: gr.Markdown( "# 🫀 ECG卡使用者问答智能体\n" "面向ECG卡使用者的中文问答助手(方案 A2)。回复严格遵循 **问题分析 / 原因说明 / 建议 / 是否需要医生** 四段模板;涉及胸痛、晕厥、呼吸困难等关键词会自动升级为急诊提示。\n\n" "> 本 Demo 内置约 160 条 FAQ 为示例条目,标注 `需临床团队审核`;所有回复不构成医学诊断,以医生意见为准。" ) with gr.Row(): # 左侧:患者档案 with gr.Column(scale=1): gr.Markdown("### 🗂 我的历史数据(可选)") gr.Markdown( "上传或粘贴 JSON 档案,机器人将结合您的 ECG 历史、报告、用药、预警给出更贴合的回复。\n" "留空时机器人不会引用任何患者数据。" ) sample_dropdown = gr.Dropdown( choices=SAMPLE_CHOICES, value=None, label="载入示例档案", interactive=True, ) with gr.Row(): load_btn = gr.Button("载入示例", variant="secondary", size="sm") clear_btn = gr.Button("清空档案", size="sm") profile_json = gr.Textbox( label="档案 JSON", placeholder='{"patient": {"name": "张**", "age": 68, ...}, "ecg_history": [...], ...}', lines=12, elem_id="patient_json", ) apply_btn = gr.Button("解析并预览", variant="primary") preview = gr.Markdown("(暂无档案预览)", elem_id="preview") # 右侧:对话 with gr.Column(scale=2): gr.Markdown("### 💬 咨询对话") chatbot = gr.Chatbot( label="ECG助手", height=520, show_copy_button=True, type="tuples", # 保持与现有 [user, assistant] 元组格式兼容 ) with gr.Row(): user_box = gr.Textbox( placeholder="请输入问题,例如:『我最近心率怎么样?』", lines=2, scale=8, show_label=False, ) send_btn = gr.Button("发送", variant="primary", scale=1) with gr.Row(): clear_chat_btn = gr.Button("🗑 清空对话", size="sm") gr.Markdown("**示例问题(点击自动填入输入框)**") with gr.Row(): example_btns = [gr.Button(q, size="sm") for q in EXAMPLE_QUESTIONS] # ---- 绑定回调 ---- load_btn.click(on_load_sample, inputs=[sample_dropdown], outputs=[profile_json, preview]) clear_btn.click(on_clear_profile, outputs=[profile_json, preview]) apply_btn.click(on_apply_profile, inputs=[profile_json], outputs=[preview, preview]) send_btn.click(respond, inputs=[user_box, chatbot, profile_json], outputs=[chatbot, user_box]) user_box.submit(respond, inputs=[user_box, chatbot, profile_json], outputs=[chatbot, user_box]) clear_chat_btn.click(lambda: [], outputs=[chatbot]) for btn, q in zip(example_btns, EXAMPLE_QUESTIONS): btn.click(on_example_click, inputs=[gr.State(q)], outputs=[user_box]) # 底部信息 info = ( f"**模型:** `{MODEL_ID}` (可通过 Space Variables `MODEL_ID` 覆盖)  ·  " f"**HF_TOKEN:** {'✅ 已配置' if HF_TOKEN else '❌ 未配置(Space Settings → Secrets 中添加)'}  ·  " f"**FAQ:** {len(FAQS)} 条  ·  " f"**示例患者:** {len(SAMPLE_PATIENTS)} 位" ) gr.Markdown(info) gr.Markdown( "> ⚠ 本工具仅为科普/使用引导,不能替代医生;紧急情况请拨打 **120**。" ) return demo demo = build_ui() if __name__ == "__main__": demo.launch()