Spaces:
Sleeping
Sleeping
| """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}" | |
| 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() | |