MedGemma 1.5 (4B)是Google公司于2026年1月14日面向开发者的基础医疗多模态大模型（非临床诊断工具），本文对本地如何部署该模型进行详细说明，并开发了完整的web-UI界面方便使用，代码见下文，支持：

1.纯文本问答

2.单图提问

3.多图连续追问

MedGemma 1.5核心能力与技术突破

• CT（计算机断层扫描）：可处理三维体数据切片序列，识别肺结节、脑出血等
• MRI（磁共振成像）：支持 T1/T2 加权图像分析，如脑肿瘤分割、脊髓病变检测
• 全切片病理图像（WSI）：通过 patch-based 编码器解析数字玻片，辅助癌症分级
• 纵向影像对比：自动比对患者历史 X 光/CT，识别病情进展（如肺炎吸收、骨转移）

官方基准测试结果：

任务	指标	MedGemma 1.5 表现
CT 疾病分类	准确率	61%（↑3%）
MRI 异常检测	准确率	65%（↑14%）
病理报告生成	ROUGE-L	0.49（↑0.47，达 PolyPath 水平）
解剖结构定位	Chest ImaGenome 交叉率	38%（↑35%）
纵向 X 光对比	MS-CXR-T 宏观准确率	66%（↑5%）

多模态融合架构

• 文本 + 图像联合推理：例如输入“这位患者的胸部 CT 显示什么？”+ CT 切片 → 生成自然语言报告
• 基于 SigLIP 的医学图像编码器：在放射科、病理科、皮肤科等专业图像上预训练
• 解码器-only Transformer：继承自 Gemma 3 架构，支持 128K tokens 长上下文

本地部署资源需求

组件	最低要求
GPU	RTX 3090 / A10 / L4（≥24GB 显存）
CPU/RAM	≥32GB 内存
存储	≥20GB SSD
框架	Python ≥3.10, PyTorch ≥2.1, transformers ≥4.38
量化支持	GGUF、AWQ（可在 RTX 4090 上运行）

环境安装

# 准备编译工具
sudo apt update && sudo apt upgrade -y
sudo apt install build-essential -u 

# 创建运行环境
conda create -n MedGemma python=3.10 -y 
conda activate MedGemma

# 克隆项目
git clone https://github.com/Google-Health/F.git
cd medgemma

# 安装依赖组件
pip install --upgrade transformers accelerate bitsandbytes torch pillow gradio

# 准备模型文件，如果由于网络原因，这步会失效，需要手动自己下载
huggingface-clidownloadgoogle/medgemma-4b-it--local-dircheckpoints/medgemma-4b-it

注意：模型虽然开源，但需要授权，在huggingface上点击获取，我部署的模型为 medgemma-1.5-4b-it

基于模型的二次应用开发

应用1：纯文本问答

功能：在命令行输入问题进行回答

"""
MedGemma 交互式问答系统
在终端直接输入问题，获得 MedGemma 模型的回答- Mr.Cun
"""

import os
import sys
import torch
from typing import Optional

# 导入必要的库
try:
    from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
except ImportError as e:
    print(f"缺少必要的依赖库: {e}")
    print("请运行: pip install torch transformers accelerate")
    sys.exit(1)


class MedGemmaChat:
    def __init__(self, model_path: str, use_quantization: bool = False):
        """
        初始化 MedGemma 聊天系统

        Args:
            model_path: 本地模型路径
            use_quantization: 是否使用4位量化（节省显存）
        """
        self.model_path = model_path
        self.use_quantization = use_quantization

        # 设置设备
        if torch.cuda.is_available():
            self.device = "cuda"
            print("✓ 使用 CUDA 设备")
        elif torch.backends.mps.is_available():
            self.device = "mps"
            print("✓ 使用 MPS 设备 (Apple Silicon)")
        else:
            self.device = "cpu"
            print("✓ 使用 CPU 设备")

        # 初始化模型和分词器
        self.model = None
        self.tokenizer = None
        self.pipe = None

        # 对话历史（可选）
        self.conversation_history = []

    def load_model(self, use_pipeline: bool = True):
        """加载模型"""
        print(f"\n正在加载模型: {self.model_path}")

        if not os.path.exists(self.model_path):
            print(f"模型路径不存在: {self.model_path}")
            sys.exit(1)

        # 设置模型加载参数
        model_kwargs = {
            "torch_dtype": torch.bfloat16,
        }

        # 根据设备类型调整配置
        if self.device != "mps":
            model_kwargs["device_map"] = "auto"

        # 量化配置
        if self.use_quantization:
            try:
                quantization_config = BitsAndBytesConfig(
                    load_in_4bit=True,
                    bnb_4bit_compute_dtype=torch.bfloat16,
                    bnb_4bit_use_double_quant=True,
                    bnb_4bit_quant_type="nf4"
                )
                model_kwargs["quantization_config"] = quantization_config
                print("✓ 启用4位量化")
            except ImportError:
                print("未安装 bitsandbytes，将使用标准加载")
                self.use_quantization = False

        try:
            if use_pipeline:
                # 使用 pipeline API（修复了参数冲突问题）
                self.pipe = pipeline(
                    "text-generation",
                    model=self.model_path,
                    model_kwargs=model_kwargs,  # 移除了 trust_remote_code
                )
                # 加载 tokenizer 用于后续配置
                self.tokenizer = self.pipe.tokenizer
                print("✓ 模型加载完成 (pipeline 模式)")
            else:
                # 直接加载模型
                self.model = AutoModelForCausalLM.from_pretrained(
                    self.model_path,
                    trust_remote_code=True,
                    **model_kwargs,
                )
                self.tokenizer = AutoTokenizer.from_pretrained(
                    self.model_path,
                    trust_remote_code=True
                )

                # MPS 设备需要手动移动
                if self.device == "mps":
                    self.model = self.model.to(self.device)

                print("✓ 模型加载完成 (直接加载模式)")

        except Exception as e:
            print(f"模型加载失败: {e}")
            import traceback
            traceback.print_exc()
            sys.exit(1)

    def generate_answer(
            self,
            question: str,
            system_prompt: Optional[str] = None,
            max_new_tokens: int = 1024,
            temperature: float = 0.7,
            top_p: float = 0.9
    ) -> str:
        """
        生成回答

        Args:
            question: 用户问题
            system_prompt: 系统提示词（可选）
            max_new_tokens: 最大生成token数
            temperature: 温度参数（控制随机性）
            top_p: nucleus sampling 参数

        Returns:
            模型的回答
        """
        # 默认系统提示词
        if system_prompt is None:
            system_prompt = "你是一位专业的医学专家，请用中文详细、准确地回答医学问题。"

        # 构建消息格式
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question}
        ]

        # 生成配置
        generation_config = {
            "max_new_tokens": max_new_tokens,
            "do_sample": True,
            "temperature": temperature,
            "top_p": top_p,
            "repetition_penalty": 1.1,
        }

        # 添加 pad_token_id
        if self.tokenizer:
            generation_config["pad_token_id"] = self.tokenizer.eos_token_id

        try:
            if self.pipe:
                # 使用 pipeline
                output = self.pipe(messages, **generation_config)
                response = output[0]["generated_text"][-1]["content"]

            else:
                # 直接使用模型
                inputs = self.tokenizer.apply_chat_template(
                    messages,
                    add_generation_prompt=True,
                    tokenize=True,
                    return_dict=True,
                    return_tensors="pt",
                )

                # 移动到正确的设备
                device = self.model.device if hasattr(self.model, 'device') else self.device
                inputs = {k: v.to(device) for k, v in inputs.items()}

                input_len = inputs["input_ids"].shape[-1]

                with torch.inference_mode():
                    generation = self.model.generate(
                        **inputs,
                        **generation_config
                    )
                    generation = generation[0][input_len:]

                response = self.tokenizer.decode(generation, skip_special_tokens=True)

            return response.strip()

        except Exception as e:
            return f"生成回答时出错: {e}"

    def chat(self):
        """启动交互式聊天"""
        print("\n" + "=" * 60)
        print("MedGemma 医学问答系统")
        print("=" * 60)
        print("\n使用说明:")
        print("  - 直接输入医学问题，按回车获得回答")
        print("  - 输入 'quit' 或 'exit' 退出")
        print("  - 输入 'clear' 清空对话历史")
        print("  - 输入 'help' 查看帮助")
        print("\n" + "=" * 60 + "\n")

        while True:
            try:
                # 获取用户输入
                user_input = input("您的问题: ").strip()

                # 处理特殊命令
                if user_input.lower() in ['quit', 'exit', 'q']:
                    print("\n 感谢使用，再见！")
                    break

                if user_input.lower() == 'clear':
                    self.conversation_history.clear()
                    print("✓ 对话历史已清空\n")
                    continue

                if user_input.lower() == 'help':
                    self._show_help()
                    continue

                if not user_input:
                    print("请输入问题\n")
                    continue

                # 生成回答
                print("\n正在思考...\n")
                answer = self.generate_answer(user_input)

                # 显示回答
                print("💡 回答:")
                print("-" * 60)
                print(answer)
                print("-" * 60)
                print()

                # 保存到历史（可选）
                self.conversation_history.append({
                    "question": user_input,
                    "answer": answer
                })

            except KeyboardInterrupt:
                print("\n\n检测到中断信号，退出程序")
                break

            except Exception as e:
                print(f"\n发生错误: {e}\n")
                continue

    def _show_help(self):
        """显示帮助信息"""
        print("\n" + "=" * 60)
        print("📖 帮助信息")
        print("=" * 60)
        print("""
命令列表:
  quit/exit/q  - 退出程序
  clear        - 清空对话历史
  help         - 显示此帮助信息

使用提示:
  - 尽量提出明确、具体的医学问题
  - 可以询问症状、诊断、治疗等各方面
  - 模型会尽可能详细地回答问题
  - 回答仅供参考，不能替代专业医疗建议

示例问题:
  - 糖尿病有哪些常见症状？
  - 高血压的治疗方法有哪些？
  - 如何预防心血管疾病？
        """)
        print("=" * 60 + "\n")

    def ask_once(self, question: str):
        """
        单次问答（非交互模式）

        Args:
            question: 问题
        """
        print(f"\n❓ 问题: {question}\n")
        print("正在生成回答...\n")

        answer = self.generate_answer(question)

        print(" 回答:")
        print("-" * 60)
        print(answer)
        print("-" * 60)

        return answer


def main():
    """主函数"""
    # ========== 配置区域 ==========
    MODEL_PATH = "/home/aic/reseach/medgemma/checkpoints/medgemma-1.5-4b-it"
    USE_QUANTIZATION = False  # 是否启用量化（节省显存）
    USE_PIPELINE = True  # 是否使用 pipeline API
    # ============================

    try:
        # 初始化聊天系统
        chat_system = MedGemmaChat(
            model_path=MODEL_PATH,
            use_quantization=USE_QUANTIZATION
        )

        # 加载模型
        chat_system.load_model(use_pipeline=USE_PIPELINE)

        # 检查命令行参数
        if len(sys.argv) > 1:
            # 单次问答模式
            question = " ".join(sys.argv[1:])
            chat_system.ask_once(question)
        else:
            # 交互式聊天模式
            chat_system.chat()

    except KeyboardInterrupt:
        print("\n\n程序已中断")
    except Exception as e:
        print(f"\n程序执行出错: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()

终端运行情况，是用markdown方式回复：

============================================================
🏥 MedGemma 医学问答系统
============================================================

使用说明:
  - 直接输入医学问题，按回车获得回答
  - 输入 'quit' 或 'exit' 退出
  - 输入 'clear' 清空对话历史
  - 输入 'help' 查看帮助

============================================================

❓ 您的问题: 如何区分一型和二型糖尿病？
如何区分一型和二型糖尿病？

正在思考...

💡 回答:
------------------------------------------------------------
好的，作为一名医学专家，我将详细且准确地介绍如何区分一型和二型糖尿病：

**一、核心区别：病因与胰岛素分泌情况**

这是区分一型和二型糖尿病最根本的区别：

*   **一型糖尿病 (Type 1 Diabetes):**  属于自身免疫性疾病，患者的自身免疫系统错误地攻击并破坏了胰腺中的β细胞（β-cells），这些β细胞负责产生胰岛素。因此，一型糖尿病患者几乎无法自行产生足够的胰岛素，需要依赖外部胰岛素才能维持生命。

*   **二型糖尿病 (Type 2 Diabetes):** 是一种代谢性疾病，通常是多种因素共同作用的结果，包括遗传倾向、生活方式（饮食不健康、缺乏运动）等。在早期阶段，胰腺可能还能产生足够的胰岛素，但由于身体对胰岛素抵抗（insulin resistance），导致胰岛素难以有效发挥其作用。随着时间的推移，胰腺的功能可能会逐渐下降，最终导致胰岛素分泌不足。

....回复得比较多，我省略了

应用2：基于WebUI构建

用图形界面比较方便使用，因此编写了基于gradio的界面方式，支持图文问答。代码不复杂，直接贴出来：

"""
MedGemma Web UI - Mr.Cun
支持：单图/多图/纯文本 多模态医学问答
"""

import os
import sys
import torch
from typing import Optional, List, Dict, Any, Tuple
from pathlib import Path

# 导入必要的库
try:
    from transformers import (
        pipeline,
        AutoModelForImageTextToText,
        AutoModelForCausalLM,
        AutoProcessor,
        AutoTokenizer,
        BitsAndBytesConfig
    )
    from PIL import Image
    import gradio as gr
except ImportError as e:
    print(f"缺少必要的依赖库: {e}")
    print("请运行: pip install torch transformers accelerate pillow gradio")
    sys.exit(1)


class MedGemmaWebUI:
    def __init__(self, model_path: str, use_quantization: bool = False):
        """
        初始化 MedGemma Web UI

        Args:
            model_path: 本地模型路径
            use_quantization: 是否使用4位量化
        """
        self.model_path = model_path
        self.use_quantization = use_quantization

        # 检测模型类型
        self.is_text_only = "text" in os.path.basename(model_path).lower()

        # 设置设备
        if torch.cuda.is_available():
            self.device = "cuda"
            gpu_name = torch.cuda.get_device_name(0)
            print(f"✓ 使用 GPU: {gpu_name}")
        elif torch.backends.mps.is_available():
            self.device = "mps"
            print("✓ 使用 MPS 设备 (Apple Silicon)")
        else:
            self.device = "cpu"
            print("✓ 使用 CPU 设备")

        self.model = None
        self.processor = None
        self.tokenizer = None
        self.pipe = None

    def load_model(self, use_pipeline: bool = True):
        """加载模型"""
        print(f"\n正在加载模型: {self.model_path}")
        print(f"模型类型: {'纯文本' if self.is_text_only else '多模态（图像+文本）'}")

        if not os.path.exists(self.model_path):
            raise ValueError(f"模型路径不存在: {self.model_path}")

        # 模型加载参数
        model_kwargs = {
            "torch_dtype": torch.bfloat16,
            "device_map": "auto",
        }

        # 量化配置
        if self.use_quantization:
            try:
                quantization_config = BitsAndBytesConfig(
                    load_in_4bit=True,
                    bnb_4bit_compute_dtype=torch.bfloat16,
                    bnb_4bit_use_double_quant=True,
                    bnb_4bit_quant_type="nf4"
                )
                model_kwargs["quantization_config"] = quantization_config
                print("✓ 启用4位量化")
            except ImportError:
                print(" bitsandbytes，使用标准加载")
                self.use_quantization = False

        try:
            if self.is_text_only:
                # 加载纯文本模型
                if use_pipeline:
                    self.pipe = pipeline(
                        "text-generation",
                        model=self.model_path,
                        model_kwargs=model_kwargs,
                    )
                    self.pipe.model.generation_config.do_sample = False
                else:
                    self.model = AutoModelForCausalLM.from_pretrained(
                        self.model_path,
                        **model_kwargs,
                    )
                    self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
                print("✓ 纯文本模型加载完成")
            else:
                # 加载多模态模型
                if use_pipeline:
                    self.pipe = pipeline(
                        "image-text-to-text",
                        model=self.model_path,
                        model_kwargs=model_kwargs,
                    )
                    self.pipe.model.generation_config.do_sample = False
                else:
                    self.model = AutoModelForImageTextToText.from_pretrained(
                        self.model_path,
                        **model_kwargs,
                    )
                    self.processor = AutoProcessor.from_pretrained(self.model_path)
                print("✓ 多模态模型加载完成")

        except Exception as e:
            print(f" 模型加载失败: {e}")
            raise

    def _extract_text_from_history(self, chat_history: List[Dict]) -> List[Dict]:
        """
        从聊天历史中提取纯文本消息

        Args:
            chat_history: Gradio 新格式的聊天历史

        Returns:
            标准消息格式列表
        """
        messages = []
        for msg in chat_history:
            if not isinstance(msg, dict):
                continue

            role = msg.get("role", "")
            content = msg.get("content", "")

            # 清理用户消息中的图片标记
            if role == "user" and isinstance(content, str):
                if content.startswith("[已上传"):
                    content = content.split("] ", 1)[-1] if "] " in content else content

            messages.append({
                "role": role,
                "content": content
            })

        return messages

    def chat_with_context(
            self,
            images: Optional[List],
            question: str,
            chat_history: List[Dict],
            system_prompt: str,
            max_tokens: int,
            progress=gr.Progress()
    ) -> Tuple[List[Dict], str]:
        """
        支持多模态的对话（单图/多图/纯文本）

        Returns:
            (更新后的聊天历史, 空字符串)
        """
        if not question.strip():
            return chat_history, ""

        # 判断是否有图片
        has_images = images is not None and len(images) > 0

        # 如果是纯文本模型但用户上传了图片
        if self.is_text_only and has_images:
            return chat_history + [
                {"role": "user", "content": question},
                {"role": "assistant",
                 "content": "当前模型是纯文本版本，不支持图像输入。请切换到多模态模型或仅输入文本问题。"}
            ], ""

        # 根据是否有图片选择模式
        if not has_images:
            return self._text_only_chat(question, chat_history, system_prompt, max_tokens, progress)
        else:
            return self._multimodal_chat(images, question, chat_history, system_prompt, max_tokens, progress)

    def _text_only_chat(
            self,
            question: str,
            chat_history: List[Dict],
            system_prompt: str,
            max_tokens: int,
            progress
    ) -> Tuple[List[Dict], str]:
        """纯文本对话"""
        progress(0, desc="准备分析...")

        try:
            # 提取历史对话
            extracted_history = self._extract_text_from_history(chat_history)

            # 根据模型类型构建不同的消息格式
            if self.is_text_only:
                # 纯文本模型：使用简单字符串格式
                messages = [{"role": "system", "content": system_prompt}]
                messages.extend(extracted_history)
                messages.append({"role": "user", "content": question})
            else:
                # 多模态模型：content 必须是列表格式
                messages = [
                    {
                        "role": "system",
                        "content": [{"type": "text", "text": system_prompt}]
                    }
                ]

                # 添加历史对话
                for msg in extracted_history:
                    messages.append({
                        "role": msg["role"],
                        "content": [{"type": "text", "text": msg["content"]}]
                    })

                # 添加当前问题
                messages.append({
                    "role": "user",
                    "content": [{"type": "text", "text": question}]
                })

            progress(0.3, desc="生成回答中...")

            # 生成回答
            if self.pipe:
                output = self.pipe(messages, max_new_tokens=max_tokens)
                # Pipeline 返回格式处理
                if isinstance(output, list) and len(output) > 0:
                    generated = output[0]
                    if isinstance(generated, dict):
                        response = generated.get("generated_text", "")
                        # 如果是完整对话，提取最后一条
                        if isinstance(response, list):
                            last_msg = response[-1] if len(response) > 0 else {}
                            if isinstance(last_msg, dict):
                                # 处理多模态格式的回复
                                content = last_msg.get("content", "")
                                if isinstance(content, list):
                                    # 提取文本内容
                                    text_parts = [c.get("text", "") for c in content if
                                                  isinstance(c, dict) and c.get("type") == "text"]
                                    response = " ".join(text_parts)
                                else:
                                    response = content
                            else:
                                response = str(last_msg)
                        elif isinstance(response, str):
                            pass  # 已经是字符串
                    else:
                        response = str(generated)
                else:
                    response = "生成失败"
            else:
                # 直接使用模型
                if self.is_text_only:
                    inputs = self.tokenizer.apply_chat_template(
                        messages,
                        add_generation_prompt=True,
                        tokenize=True,
                        return_tensors="pt",
                    ).to(self.model.device)

                    input_len = inputs.shape[-1]

                    with torch.inference_mode():
                        generation = self.model.generate(
                            inputs,
                            max_new_tokens=max_tokens,
                            do_sample=False
                        )
                        generation = generation[0][input_len:]

                    response = self.tokenizer.decode(generation, skip_special_tokens=True)
                else:
                    # 多模态模型
                    inputs = self.processor.apply_chat_template(
                        messages,
                        add_generation_prompt=True,
                        tokenize=True,
                        return_dict=True,
                        return_tensors="pt",
                    ).to(self.model.device, dtype=torch.bfloat16)

                    input_len = inputs["input_ids"].shape[-1]

                    with torch.inference_mode():
                        generation = self.model.generate(
                            **inputs,
                            max_new_tokens=max_tokens,
                            do_sample=False
                        )
                        generation = generation[0][input_len:]

                    response = self.processor.decode(generation, skip_special_tokens=True)

            progress(1.0, desc="完成")

            # 返回新格式
            return chat_history + [
                {"role": "user", "content": question},
                {"role": "assistant", "content": response.strip()}
            ], ""

        except Exception as e:
            import traceback
            error_detail = traceback.format_exc()
            print(f"文本对话错误: {error_detail}")
            error_msg = f"生成回答时出错: {str(e)}"
            return chat_history + [
                {"role": "user", "content": question},
                {"role": "assistant", "content": error_msg}
            ], ""

    def _multimodal_chat(
            self,
            images: List,
            question: str,
            chat_history: List[Dict],
            system_prompt: str,
            max_tokens: int,
            progress
    ) -> Tuple[List[Dict], str]:
        """多模态对话（带图片）"""
        progress(0, desc="准备分析...")

        try:
            # 处理图像
            processed_images = []
            for img_item in images:
                # img_item 可能是文件路径或 PIL Image
                if isinstance(img_item, str):
                    img = Image.open(img_item)
                elif isinstance(img_item, Image.Image):
                    img = img_item
                else:
                    continue

                if img.mode != "RGB":
                    img = img.convert("RGB")
                processed_images.append(img)

            if not processed_images:
                return chat_history + [
                    {"role": "user", "content": question},
                    {"role": "assistant", "content": "图片加载失败"}
                ], ""

            # 图片说明
            image_count = len(processed_images)
            image_desc = f"[已上传{image_count}张图片] " if image_count > 1 else "[已上传1张图片] "
            enhanced_question = image_desc + question

            # 构建消息
            messages = [
                {
                    "role": "system",
                    "content": [{"type": "text", "text": system_prompt}]
                }
            ]

            # 添加历史对话（简化，只保留文本）
            extracted_history = self._extract_text_from_history(chat_history)
            for msg in extracted_history:
                messages.append({
                    "role": msg["role"],
                    "content": [{"type": "text", "text": msg["content"]}]
                })

            # 添加当前问题（附带所有图片）
            current_content = [{"type": "text", "text": question}]
            for img in processed_images:
                current_content.append({"type": "image", "image": img})

            messages.append({
                "role": "user",
                "content": current_content
            })

            progress(0.3, desc="生成回答中...")

            # 生成回答
            if self.pipe:
                output = self.pipe(
                    text=messages,
                    max_new_tokens=max_tokens
                )
                # Pipeline 返回格式处理
                if isinstance(output, list) and len(output) > 0:
                    generated = output[0]
                    if isinstance(generated, dict):
                        gen_text = generated.get("generated_text", "")
                        # 提取最后一条消息的内容
                        if isinstance(gen_text, list):
                            last_msg = gen_text[-1] if len(gen_text) > 0 else {}
                            response = last_msg.get("content", "") if isinstance(last_msg, dict) else str(last_msg)
                        else:
                            response = str(gen_text)
                    else:
                        response = str(generated)
                else:
                    response = "生成失败"
            else:
                # 直接使用模型
                inputs = self.processor.apply_chat_template(
                    messages,
                    add_generation_prompt=True,
                    tokenize=True,
                    return_dict=True,
                    return_tensors="pt",
                ).to(self.model.device, dtype=torch.bfloat16)

                input_len = inputs["input_ids"].shape[-1]

                with torch.inference_mode():
                    generation = self.model.generate(
                        **inputs,
                        max_new_tokens=max_tokens,
                        do_sample=False
                    )
                    generation = generation[0][input_len:]

                response = self.processor.decode(generation, skip_special_tokens=True)

            progress(1.0, desc="完成")

            # 返回新格式
            return chat_history + [
                {"role": "user", "content": enhanced_question},
                {"role": "assistant", "content": response.strip()}
            ], ""

        except Exception as e:
            import traceback
            error_detail = traceback.format_exc()
            print(f"多模态对话错误: {error_detail}")
            error_msg = f" 生成回答时出错: {str(e)}"

            image_count = len(images) if images else 0
            image_desc = f"[已上传{image_count}张图片] " if image_count > 0 else ""

            return chat_history + [
                {"role": "user", "content": image_desc + question},
                {"role": "assistant", "content": error_msg}
            ], ""

    def create_interface(self):
        """创建 Gradio 界面 - 适配 Gradio 6.x"""

        # 自定义 CSS
        custom_css = """
        .gradio-container {
            font-family: 'Arial', sans-serif;
        }
        .header {
            text-align: center;
            padding: 20px;
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white;
            border-radius: 10px;
            margin-bottom: 20px;
        }
        .warning-box {
            background-color: #fff3cd;
            border: 1px solid #ffc107;
            border-radius: 5px;
            padding: 15px;
            margin: 10px 0;
        }
        .info-box {
            background-color: #d1ecf1;
            border: 1px solid #0c5460;
            border-radius: 5px;
            padding: 15px;
            margin: 10px 0;
        }
        """

        # 创建界面
        with gr.Blocks(css=custom_css, title="MedGemma 多模态医学问答系统") as demo:

            # 标题
            gr.HTML("""
                <div class="header">
                    <h1>🏥 MedGemma 多模态医学问答系统</h1>
                    <p>支持：单图 | 多图 | 纯文本 - 灵活的医学AI助手</p>
                    <p>开发者：寸老师</p>
                </div>
            """)

            # 免责声明
            gr.HTML("""
                <div class="warning-box">
                    <strong> 重要提示：</strong>
                    本系统提供的分析结果仅供参考，不能替代专业医疗诊断。
                    请咨询专业医生获取准确的医疗建议。
                </div>
            """)

            # 功能说明
            model_type = "纯文本" if self.is_text_only else "多模态（图像+文本）"
            gr.HTML(f"""
                <div class="info-box">
                    <strong>当前模型：</strong>{model_type}<br>
                    <strong>使用方式：</strong>
                    <ul style="margin: 5px 0; padding-left: 20px;">
                        <li><strong>单图模式：</strong>上传1张影像 + 提问</li>
                        <li><strong>多图模式：</strong>上传多张影像（如不同切面的CT）+ 提问</li>
                        <li><strong>纯文本模式：</strong>不上传图片，直接咨询医学问题</li>
                    </ul>
                </div>
            """)

            with gr.Row():
                # 左侧：输入区域
                with gr.Column(scale=1):
                    gr.Markdown("### 上传医学影像（可选）")

                    # Gradio 6.x 的 File 组件
                    image_input = gr.File(
                        label="医学影像（可上传多张，Ctrl/Cmd多选）",
                        file_count="multiple",
                        file_types=["image"]
                    )

                    gr.Markdown("""
                    **提示：**
                    - 可以不上传图片，直接进行纯文本医学咨询
                    - 可以上传1张图片进行分析
                    - 可以上传多张图片（Ctrl/Cmd多选）
                    """)

                    # 高级设置
                    with gr.Accordion("⚙️ 高级设置", open=False):
                        system_prompt = gr.Textbox(
                            label="系统提示词",
                            value="你是一位专业的医学专家，请用中文详细、准确地分析医学影像或回答医学问题。",
                            lines=3
                        )

                        max_tokens = gr.Slider(
                            label="最大生成长度",
                            minimum=128,
                            maximum=2048,
                            value=512,
                            step=64
                        )

                # 右侧：对话区域
                with gr.Column(scale=1):
                    gr.Markdown("### 💬 AI 对话")

                    # Gradio 6.x 的 Chatbot（移除 type 参数）
                    chatbot = gr.Chatbot(
                        label="对话历史",
                        height=500
                    )

                    with gr.Row():
                        question_input = gr.Textbox(
                            label="您的问题",
                            placeholder="输入问题后按回车或点击发送...",
                            lines=2,
                            scale=4
                        )
                        send_btn = gr.Button("📤 发送", variant="primary", scale=1)

                    with gr.Row():
                        clear_btn = gr.Button("🗑️ 清空对话")
                        clear_images_btn = gr.Button("🖼️ 清空图片")

            # 示例
            gr.Markdown("### 📚 使用示例")

            with gr.Tabs():
                with gr.Tab("🖼️ 单图/多图分析"):
                    gr.Examples(
                        examples=[
                            ["描述这张X光片的主要发现"],
                            ["对比这几张CT切面，病灶的范围有多大？"],
                            ["这些不同时期的影像显示病情如何变化？"],
                        ],
                        inputs=question_input,
                        label="影像分析示例"
                    )

                with gr.Tab("💬 纯文本咨询"):
                    gr.Examples(
                        examples=[
                            ["什么是肺结节？如何分类？"],
                            ["高血压患者如何选择降压药？"],
                            ["糖尿病的诊断标准是什么？"],
                        ],
                        inputs=question_input,
                        label="医学知识咨询示例"
                    )

            # 处理图片上传（Gradio 6.x 格式）
            def process_uploaded_files(files):
                """处理上传的文件 - Gradio 6.x 版本"""
                if not files:
                    return None

                images = []
                # Gradio 6.x 返回文件路径字符串列表
                if isinstance(files, list):
                    for f in files:
                        try:
                            # f 可能是字符串路径
                            if isinstance(f, str):
                                images.append(Image.open(f))
                            # 或者是字典 {'name': 路径}
                            elif isinstance(f, dict) and 'name' in f:
                                images.append(Image.open(f['name']))
                            # 或者有 name 属性
                            elif hasattr(f, 'name'):
                                images.append(Image.open(f.name))
                        except Exception as e:
                            print(f" 图片加载失败: {e}")
                            continue
                # 单个文件
                elif isinstance(files, str):
                    images.append(Image.open(files))
                elif isinstance(files, dict) and 'name' in files:
                    images.append(Image.open(files['name']))
                elif hasattr(files, 'name'):
                    images.append(Image.open(files.name))

                return images if images else None

            # 事件绑定
            send_btn.click(
                fn=lambda files, q, h, sp, mt: self.chat_with_context(
                    process_uploaded_files(files),
                    q, h, sp, mt
                ),
                inputs=[image_input, question_input, chatbot, system_prompt, max_tokens],
                outputs=[chatbot, question_input]
            )

            question_input.submit(
                fn=lambda files, q, h, sp, mt: self.chat_with_context(
                    process_uploaded_files(files),
                    q, h, sp, mt
                ),
                inputs=[image_input, question_input, chatbot, system_prompt, max_tokens],
                outputs=[chatbot, question_input]
            )

            clear_btn.click(
                fn=lambda: ([], ""),
                inputs=None,
                outputs=[chatbot, question_input]
            )

            clear_images_btn.click(
                fn=lambda: None,
                inputs=None,
                outputs=image_input
            )

        return demo


def main():
    """主函数"""
    # ========== 配置区域 ==========
    MODEL_PATH = "/home/aic/reseach/medgemma/checkpoints/medgemma-1.5-4b-it"
    USE_QUANTIZATION = False
    USE_PIPELINE = True

    # Web UI 配置
    SERVER_NAME = "0.0.0.0"
    SERVER_PORT = 7860
    SHARE = False
    # ============================

    try:
        print("=" * 70)
        print("MedGemma 多模态医学问答系统")
        print("=" * 70)

        # 初始化系统
        web_ui = MedGemmaWebUI(
            model_path=MODEL_PATH,
            use_quantization=USE_QUANTIZATION
        )

        # 加载模型
        web_ui.load_model(use_pipeline=USE_PIPELINE)

        # 创建界面
        demo = web_ui.create_interface()

        # 启动服务
        print("\n" + "=" * 70)
        print("启动 Web 服务器...")
        print("=" * 70)

        demo.launch(
            server_name=SERVER_NAME,
            server_port=SERVER_PORT,
            share=SHARE,
            show_error=True,
            quiet=False
        )

    except KeyboardInterrupt:
        print("\n\n服务器已停止")
    except Exception as e:
        print(f"\n程序执行出错: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()

运行后界面如下：

测试输入CT：

模型输出结论：

好的，根据您提供的CT影像，我将进行详细的分析和解读。

影像特征：

胸廓结构： 胸廓的形态看起来正常，没有明显的骨折、移位或异常。
肺部： 肺部呈正常形态，没有明显的肺气肿、肺水肿或肺结节。
心脏： 心脏的大小和位置看起来正常。
血管： 主动脉、肺动脉等主要血管的走行和形态看起来正常。
胸膜： 胸膜看起来正常，没有明显的积液或胸腔积气。
软组织： 纵隔、胸膜等软组织的形态看起来正常。
异常表现：

右肺下叶可见一个圆形或椭圆形区域，其密度略高于周围组织。 这种区域可能代表着一个小的结节、囊肿或炎症。
可能诊断及依据：

基于以上影像特征和异常表现，以下是一些可能的诊断：

肺结节： 这是最常见的可能性。肺结节是指肺部内单个或多个小的、固定的、圆形或椭圆形的肿块。结节的直径通常小于3厘米。根据结节的形态、大小、密度和生长速度，可以进一步判断其性质。
依据： 结节的密度略高于周围组织，提示其可能是一个固定的肿块。
需要进一步检查： 进一步的影像学检查（如CT增强扫描、PET-CT）和/或活组织检查（如支气管镜活检、经皮穿刺活检）是确诊肺结节性质的必要步骤。
肺囊肿： 肺囊肿是指肺部内形成的空腔，通常是良性的。
依据： 囊肿的圆形或椭圆形区域，其密度略高于周围组织，提示其可能是一个空腔。
需要进一步检查： 进一步的影像学检查（如CT增强扫描）可以帮助区分囊肿和结节。
炎症： 肺部炎症也可能表现为圆形或椭圆形区域，其密度略高于周围组织。
依据： 炎症区域的密度略高于周围组织，提示其可能是一个炎症反应。
需要进一步检查： 进一步的影像学检查（如CT增强扫描）可以帮助评估炎症的程度和范围。
其他可能性： 还有一些其他可能性，例如：
转移瘤： 如果患者有其他癌症病史，则转移瘤的可能性需要考虑。
血管瘤： 血管瘤是指肺部内形成的血管团，通常是良性的。
其他良性或恶性病变： 还有一些其他良性或恶性病变也可能表现为圆形或椭圆形区域。
重要提示：

以上诊断仅为可能性，不能作为最终诊断。
需要结合患者的临床症状、病史、体格检查结果以及其他辅助检查结果进行综合评估，才能做出准确的诊断。
请务必咨询专业的医生，进行进一步的检查和治疗。
建议：

进行CT增强扫描： 增强扫描可以帮助区分结节和囊肿，并评估结节的血供情况。
进行PET-CT： PET-CT可以帮助评估结节的代谢活性，从而判断其性质。
进行活组织检查： 如果影像学检查结果不明确，则需要进行活组织检查，以明确结节或囊肿的性质。
希望以上分析对您有所帮助。请记住，这只是一个初步的分析，最终的诊断和治疗方案需要由专业的医生根据患者的具体情况来决定。

测试一个报告解读

输入提问：我上传了1检查报告单，是一病人的C肽释放试验+胰岛素释放试验结果，请你阅读结果，分析这个病人最可能存在的问题

AI回答：

好的，我将根据您上传的C肽释放试验和胰岛素释放试验结果，对该患者进行分析，并提供详细的解释。

首先，我们来解读一下这些试验的原理和意义：

C肽释放试验 (C-Peptide Stimulation Test): C肽是胰高血糖素样肽-2 (GLP-2) 的前体，由胰腺β细胞分泌。该试验通过注射C肽来刺激胰腺β细胞，评估其功能。C肽的释放情况可以反映β细胞的储备功能和功能状态。
胰岛素释放试验 (Insulin Stimulation Test): 胰岛素是β细胞分泌的激素，用于降低血糖。该试验通过注射胰岛素来刺激胰腺β细胞，评估其功能。胰岛素的释放情况可以反映β细胞的敏感性和功能。
根据您提供的报告单，该患者的C肽释放试验和胰岛素释放试验结果显示：

C肽释放试验： C肽释放量较低，C-P肽/C-肽比值偏高。
胰岛素释放试验： 胰岛素释放量较低，胰岛素/C肽比值偏高。
综合分析：

这些结果提示该患者可能存在胰岛β细胞功能障碍。具体表现为：

β细胞功能减退： 胰岛素和C肽的释放量均低于正常水平，表明β细胞的储备功能和功能状态均受到影响。
β细胞敏感性降低： 胰岛素/C肽比值偏高，提示β细胞对胰岛素的敏感性可能降低。
可能的诊断：

基于以上分析，最可能的诊断是糖尿病前期或早期糖尿病。

诊断依据：

C肽和胰岛素释放试验的异常： 异常的C肽和胰岛素释放试验结果是诊断糖尿病前期或早期糖尿病的重要依据。
其他检查结果： 报告单中没有提供其他相关检查结果，如空腹血糖、餐后2小时血糖、糖化血红蛋白等，这些结果对于确诊糖尿病类型和程度至关重要。
建议：

进一步检查： 建议进行空腹血糖、餐后2小时血糖、糖化血红蛋白等检查，以明确诊断。
生活方式干预： 建议患者进行健康饮食、规律运动、控制体重等生活方式干预，以改善胰岛功能和血糖控制。
随访： 定期随访，监测血糖变化，必要时调整治疗方案。
重要提示：

以上分析仅基于您提供的报告单，不能作为最终诊断的依据。最终诊断需要结合患者的临床症状、病史、其他检查结果以及医生的综合判断。请务必咨询医生，进行详细的评估和诊断。

测试多CT切片输入：

由于隐私保护，本文中我不上传CT片

输入4个肺部CT切片，并且输入提示：这是同一个病人的四个切片影像，请你分析存在什么病情？

AI回复：

影像特征分析：

这些影像都是胸部CT（Computed Tomography）扫描，显示了患者的肺部和胸腔结构。

共同观察到的影像特征：

肺部弥漫性病变： 观察到肺部广泛存在不规则的、低密度（灰暗）区域，这些区域的边界不清晰，呈斑片状或结节状分布。这些病变主要分布在肺叶的下肺段和中下肺段，也可见到部分分布在肺段。
肺气肿： 肺部存在明显的气胸（肺气肿）表现，特别是下肺段，可见到肺纹理增粗、肺野的过度扩张和肺纹理的消失。
胸腔积液： 观察到胸腔内存在少量积液，主要位于胸腔后方，可能与肺部病变有关。
胸膜增厚： 肺叶的胸膜（覆盖在肺脏上的薄层）可见增厚，可能与炎症或纤维化有关。
气胸： 影像中也可见到气胸（肺气肿）的表现，即肺野的过度扩张，肺纹理消失。
胸腔内结构： 观察到胸腔内结构相对清晰，心脏、大血管等主要结构未见明显异常。
异常表现总结：

肺部： 广泛的弥漫性低密度病变（斑片状、结节状），肺气肿，胸膜增厚。
胸腔： 少量积液，气胸。
可能诊断及依据：

基于以上影像特征，结合患者的年龄（根据影像信息，患者年龄为60岁）和可能的临床背景（虽然未提供，但需要考虑），以下是一些可能的诊断方向：

肺部感染性疾病：
慢性感染性肺炎： 如支原体肺炎、非典型肺炎、或慢性化脓性肺炎等，可能导致肺部广泛的弥漫性病变和纤维化。
肺结核： 尽管影像特征不典型，但肺结核也可能表现为弥漫性病变，尤其是在免疫力低下或慢性感染的情况下。
其他感染性疾病： 如肺脓肿、肺气肿等，但其分布和形态可能与此患者影像不完全一致。
肺部纤维化/肺硬化：
间质性肺炎： 各种类型的间质性肺炎（如非特异性间质性肺炎、结节病、肺纤维化等）可能导致肺部弥漫性病变和肺气肿。
慢性肺气肿： 长期吸烟或其他因素导致慢性肺气肿，可能伴有肺部弥漫性病变。
肺癌：
多发性肺癌： 尽管影像特征不典型，但肺癌也可能表现为多发性病变。
肺癌伴肺气肿： 肺癌可能导致肺气肿，同时伴有肺部病变。
其他：
肺部血管疾病： 如肺动脉高压等，可能导致肺部弥漫性病变。
免疫相关疾病： 如系统性红斑狼疮、SLE等，可能导致肺部弥漫性病变。
依据：

影像特征： 弥漫性低密度病变、肺气肿、胸膜增厚、胸腔积液、气胸等是诊断肺部疾病的关键影像学依据。
分布模式： 病变主要分布在下肺段和中下肺段，符合某些慢性感染性肺炎、间质性肺炎或肺癌的常见分布模式。
结合年龄： 60岁患者，肺部疾病的发生风险相对较高。
需要进一步的检查和评估：

临床信息： 患者的症状（如咳嗽、咳痰、胸痛、呼吸困难等）、既往病史（如吸烟史、慢性疾病）、体格检查结果等是诊断的关键。
实验室检查： 血常规、气胸液分析、痰培养、血沉降率等。

继续输入病史：患者基本情况：男，60岁，咳嗽咳痰5年，胸闷气促3年，四年前肺活检提示慢性验证伴肺纤维增生，诊断性抗结核一年半吗，肺部症状无好转，上传的是最近的CT，请你结合病史分析有什么问题？

AI回复：

患者是一位60岁男性，有5年咳嗽咳痰、3年胸闷气促的病史。四年前肺活检提示慢性验证伴肺纤维增生，诊断性抗结核一年半前也未完全治愈。这些病史提示患者存在长期、慢性、且可能进展性的肺部疾病，特别是肺纤维化。

影像特征分析：共同观察到的影像特征：
肺部弥漫性病变： 观察到肺部广泛存在不规则的、低密度（灰暗）区域，这些区域的边界不清晰，呈斑片状或结节状分布。这些病变主要分布在肺叶的下肺段和中下肺段，也可见到部分分布在肺段。
肺气肿： 肺部存在明显的气胸（肺气肿）表现，特别是下肺段，可见到肺纹理增粗、肺野的过度扩张和肺纹理的消失。
胸腔积液： 观察到胸腔内存在少量积液，主要位于胸腔后方，可能与肺部病变有关。
胸膜增厚： 肺叶的胸膜（覆盖在肺脏上的薄层）可见增厚，可能与炎症或纤维化有关。
气胸： 影像中也可见到气胸（肺气肿）的表现，即肺野的过度扩张，肺纹理消失。
胸腔内结构： 观察到胸腔内结构相对清晰，心脏、大血管等主要结构未见明显异常。
异常表现总结： 
肺部： 广泛的弥漫性低密度病变（斑片状、结节状），肺气肿，胸膜增厚。 胸腔： 少量积液，气胸。
结合病史与影像特征的分析：慢性肺纤维化进展： 患者的病史（慢性咳嗽咳痰、胸闷气促、肺活检提示慢性验证伴肺纤维增生）与当前影像特征（弥漫性低密度病变、肺气肿、胸膜增厚）高度吻合。这提示患者可能存在慢性肺纤维化疾病的进展，特别是肺纤维化和肺气肿的共同存在。
肺气肿的严重程度： 影像中肺气肿表现明显，特别是下肺段，可能影响肺功能，并可能与肺纤维化共同导致肺部弥漫性病变。弥漫性病变的性质： 弥漫性病变的主要分布在下肺段和中下肺段，这在慢性肺纤维化、慢性感染性肺炎或肺癌等疾病中都可能见到。需要进一步明确病变性质（如是否为炎症性病变、纤维化病变或肿瘤病变）。 
胸腔积液和胸膜增厚： 胸腔积液和胸膜增厚可能与肺部炎症、纤维化或肿瘤相关，需要结合临床表现和进一步检查来判断其原因。
抗结核治疗效果不佳： 尽管诊断性抗结核一年半前进行，但肺部症状无好转，提示可能存在抗结核治疗效果不佳、药物耐药、或患者存在其他合并肺部疾病（如慢性感染性肺炎、间质性肺炎等）。
可能诊断方向： 慢性肺纤维化（进展期）： 结合病史和影像特征，慢性肺纤维化是首要考虑的诊断。影像特征（弥漫性病变、肺气肿、胸膜增厚）符合此诊断。
慢性感染性肺炎伴肺纤维化： 长期咳嗽咳痰可能提示慢性感染性肺炎，这可能导致肺部弥漫性病变和纤维化。
间质性肺炎（进展期）： 间质性肺炎也可能导致弥漫性病变和肺气肿，需要考虑其类型。
肺癌（伴肺气肿）： 尽管影像特征不典型，但肺癌也可能表现为多发性病变，并伴有肺气肿。

继续追问：有没有可能是非结合分支杆菌感染？

AI回答：

根据您提供的CT影像，以及患者的病史（长期咳嗽咳痰、胸闷气促、肺活检提示慢性验证伴肺纤维增生、抗结核治疗效果不佳），非结合分支杆菌感染（Atypical Mycobacteria infection）是可能的诊断方向之一，但并非唯一可能性。以下是分析：
非结合分支杆菌感染的影像表现：
非结合分支杆菌感染（通常指非结核分枝杆菌，如非结合分支杆菌）的肺部表现可以多样，但常见的影像特征包括：弥漫性病变： 影像中观察到的广泛的、低密度（灰暗）区域，呈斑片状或结节状分布，这与您提供的CT影像中的弥漫性病变特征一致。
肺气肿： 影像中可见的肺气肿表现，也可能与非结合分支杆菌感染相关，尤其是在慢性感染导致肺部结构破坏时。胸腔积液： 影像中观察到的少量胸腔积液，也可能与感染相关的炎症反应有关。
胸膜增厚： 影像中可见的胸膜增厚，可能与慢性炎症或纤维化有关。
其他表现： 影像中也可能观察到肺叶的局限性病变、肺结节、或肺部气胸等。
结合患者病史的考虑：慢性咳嗽咳痰： 患者有5年咳嗽咳痰的病史，这提示可能存在慢性呼吸道感染或炎症。非结合分支杆菌感染可以引起慢性肺部感染，导致长期症状。
抗结核治疗效果不佳： 尽管诊断性抗结核一年半前进行，但肺部症状无好转，这提示可能存在抗结核治疗效果不佳、药物耐药、或患者存在其他合并肺部疾病（如慢性感染性肺炎、间质性肺炎等）。非结合分支杆菌感染可能与抗结核治疗效果不佳有关。
肺活检结果： 肺活检提示慢性验证伴肺纤维增生，这提示患者存在慢性肺部疾病，非结合分支杆菌感染可能导致或加重这种慢性炎症和纤维化过程。
需要注意：
影像特征的非特异性： 弥漫性病变、肺气肿、胸膜增厚等影像特征是多种肺部疾病（包括慢性感染性肺炎、间质性肺炎、肺纤维化、肺癌等）的共同表现，难以完全排除了其他可能性。
诊断的必要性： 仅凭影像特征难以确诊非结合分支杆菌感染。需要结合临床症状、实验室检查（如痰培养、血培养、免疫学检查等）和病史进行综合判断。
其他可能的诊断： 鉴于患者的病史和影像特征，其他可能的诊断方向包括慢性肺纤维化（进展期）、慢性感染性肺炎、间质性肺炎、肺癌等。
总结：
非结合分支杆菌感染是基于您提供的影像特征和患者病史（特别是慢性症状和抗结核治疗效果不佳）而提出的一个可能的诊断方向。然而，需要进行更全面的评估，包括实验室检查和病史详细询问，才能做出最终的诊断。

使用结论

4B模型偏小，判断能力还不是太足，有条件应该下载更大的27B尝试。但4B版本可以给出建议补充检查的思路和方向，这个不错。