# ============================================================================= # modeling_medhemo_earcp.py — MedHemo EARCP 1.1 # ============================================================================= import os, base64, tempfile import numpy as np import torch from io import BytesIO from typing import Optional, List, Dict, Any from transformers import ( AutoTokenizer, AutoModelForCausalLM, AutoProcessor, LlavaForConditionalGeneration, WhisperForConditionalGeneration, WhisperProcessor, PreTrainedModel, PretrainedConfig, ) class MedHemoEARCPConfig(PretrainedConfig): model_type = "medhemo_earcp" def __init__( self, text_expert_id: str = "google/medgemma-1.5-4b-it", vision_expert_id: str = "llava-hf/llava-1.5-7b-hf", audio_expert_id: str = "openai/whisper-large-v3", earcp_alpha: float = 0.5, earcp_beta: float = 0.3, earcp_gamma: float = 0.2, max_new_tokens: int = 512, **kwargs, ): super().__init__(**kwargs) self.text_expert_id = text_expert_id self.vision_expert_id = vision_expert_id self.audio_expert_id = audio_expert_id self.earcp_alpha = earcp_alpha self.earcp_beta = earcp_beta self.earcp_gamma = earcp_gamma self.max_new_tokens = max_new_tokens self.architectures = ["MedHemoEARCPModel"] self.model_version = "1.1" self.model_family = "MedHemo" class MedHemoEARCPModel(PreTrainedModel): """MedHemo EARCP 1.1 — MedGemma 1.5 4B + LLaVA 1.5 7B + Whisper Large V3""" config_class = MedHemoEARCPConfig _no_split_modules = ["text_expert", "vision_expert", "audio_expert"] def __init__(self, config: MedHemoEARCPConfig): super().__init__(config) from huggingface_hub import get_token tok = get_token() dtype = torch.float16 if torch.cuda.is_available() else torch.float32 dmap = "auto" print("🔄 Tokenizers / Processors...") self.text_tokenizer = AutoTokenizer.from_pretrained( config.text_expert_id, token=tok, trust_remote_code=True) if self.text_tokenizer.pad_token is None: self.text_tokenizer.pad_token = self.text_tokenizer.eos_token self.vision_processor = AutoProcessor.from_pretrained(config.vision_expert_id) self.audio_processor = WhisperProcessor.from_pretrained(config.audio_expert_id) print("✅ Processors prêts") print(f"🔄 Text Expert : {config.text_expert_id}") self.text_expert = AutoModelForCausalLM.from_pretrained( config.text_expert_id, dtype=dtype, device_map=dmap, token=tok, trust_remote_code=True, low_cpu_mem_usage=True) self.text_expert.eval() print("✅ Text Expert chargé") print(f"🔄 Vision Expert : {config.vision_expert_id}") self.vision_expert = LlavaForConditionalGeneration.from_pretrained( config.vision_expert_id, dtype=dtype, device_map=dmap, low_cpu_mem_usage=True) self.vision_expert.eval() print("✅ Vision Expert chargé") print(f"🔄 Audio Expert : {config.audio_expert_id}") self.audio_expert = WhisperForConditionalGeneration.from_pretrained( config.audio_expert_id, dtype=dtype, device_map=dmap, low_cpu_mem_usage=True) self.audio_expert.eval() print("✅ Audio Expert chargé") self.earcp_weights = torch.nn.Parameter( torch.tensor([config.earcp_alpha, config.earcp_beta, config.earcp_gamma], dtype=torch.float32), requires_grad=True) print("\n🏁 MedHemo EARCP 1.1 prêt !") @staticmethod def _decode_b64_image(b64): from PIL import Image return Image.open(BytesIO(base64.b64decode(b64))).convert("RGB") @staticmethod def _decode_b64_audio(b64): import soundfile as sf, librosa raw = base64.b64decode(b64) with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: f.write(raw); tmp = f.name audio, sr = sf.read(tmp); os.unlink(tmp) if sr != 16000: audio = librosa.resample(audio, orig_sr=sr, target_sr=16000) if audio.ndim > 1: audio = audio.mean(axis=1) return audio.astype(np.float32) def _earcp_fuse(self, text_ctx, vision_ctx, audio_ctx): w = torch.softmax(self.earcp_weights, dim=0) a, b, g = w[0].item(), w[1].item(), w[2].item() parts = [] if audio_ctx: parts.append(f"[AUDIO w={g:.2f}] {audio_ctx}") if vision_ctx: parts.append(f"[VISION w={b:.2f}] {vision_ctx}") if text_ctx: parts.append(f"[TEXTE w={a:.2f}] {text_ctx}") fused = "\n\n".join(parts) fused += "\n\nFournis une réponse médicale précise et structurée :" return fused, {"alpha": a, "beta": b, "gamma": g} def _process_vision(self, image_b64, text_hint=""): image = self._decode_b64_image(image_b64) prompt = (f"USER: \nDécris cette image médicale." f"{(' Contexte : ' + text_hint) if text_hint else ''}\nASSISTANT:") inputs = self.vision_processor(text=prompt, images=image, return_tensors="pt").to(self.vision_expert.device) with torch.no_grad(): out = self.vision_expert.generate(**inputs, max_new_tokens=256, do_sample=False) decoded = self.vision_processor.decode(out[0], skip_special_tokens=True) return decoded.split("ASSISTANT:")[-1].strip() if "ASSISTANT:" in decoded else decoded.strip() def _process_audio(self, audio_b64): arr = self._decode_b64_audio(audio_b64) inputs = self.audio_processor(arr, sampling_rate=16000, return_tensors="pt").to(self.audio_expert.device) with torch.no_grad(): ids = self.audio_expert.generate( inputs["input_features"], language="fr", task="transcribe") return self.audio_processor.batch_decode(ids, skip_special_tokens=True)[0].strip() def _generate_text_response(self, prompt): inputs = self.text_tokenizer(prompt, return_tensors="pt", truncation=True, max_length=2048, padding=True, ).to(self.text_expert.device) with torch.no_grad(): out = self.text_expert.generate( **inputs, max_new_tokens=self.config.max_new_tokens, do_sample=True, temperature=0.7, top_p=0.9, repetition_penalty=1.1, pad_token_id=self.text_tokenizer.pad_token_id) new_tok = out[0][inputs["input_ids"].shape[1]:] return self.text_tokenizer.decode(new_tok, skip_special_tokens=True).strip() def forward(self, text=None, image_b64=None, audio_b64=None, history=None, **kwargs): """ Interface unifiée MedHemo EARCP 1.1. Returns : {response, earcp_weights, modalities, vision_desc, transcription} """ result = {"response": "", "earcp_weights": {}, "modalities": [], "vision_desc": None, "transcription": None} vision_ctx = audio_ctx = None if audio_b64: print("🎙️ Whisper...") audio_ctx = self._process_audio(audio_b64) result["transcription"] = audio_ctx result["modalities"].append("audio") if image_b64: print("👁️ LLaVA...") vision_ctx = self._process_vision(image_b64, text_hint=text or "") result["vision_desc"] = vision_ctx result["modalities"].append("vision") if text: result["modalities"].append("text") if not result["modalities"]: result["response"] = "Aucune entrée fournie." return result text_input = text or "" if history: hist = "\n".join([f"{m['role'].upper()}: {m['content']}" for m in history[-4:]]) text_input = f"Historique :\n{hist}\n\nQuestion : {text_input}" print("⚙️ Fusion EARCP...") fused, earcp_w = self._earcp_fuse(text_input, vision_ctx, audio_ctx) result["earcp_weights"] = earcp_w print("🧠 MedGemma...") result["response"] = self._generate_text_response(fused) print("✅ Réponse générée !") return result