multi_model_adapter.py

import os
import json
import requests
from typing import Dict, Any, Optional

class MultiModelAdapter:
    """
    Enterprise-grade Multi-Model Adapter.
    Prioritizes Gemini API (using gemini-1.5-flash endpoint).
    Supports fallback to OpenAI, Anthropic, or Local Open-Source models.
    """
    
    def __init__(self, default_model: str = "gemini-3.1-flash"):
        self.default_model = default_model
        
        # Mapping of internal model names to provider endpoints
        self.providers = {
            "gemini-3.1-flash": self._call_gemini,
            "gpt-4o": self._call_openai,
            "claude-3-5-sonnet": self._call_anthropic,
            "qwen-2-vl-local": self._call_local_vision
        }
        
    def generate_action(self, prompt: str, system_prompt: str, image_path: Optional[str] = None) -> str:
        """
        Routes the prompt to the highest priority available model.
        Returns the raw string output to be parsed by Pydantic.
        """
        try:
            # Attempt primary model
            print(f"🧠 [ADAPTER] Routing request to primary model: {self.default_model}")
            return self.providers[self.default_model](prompt, system_prompt, image_path)
        except Exception as e:
            print(f"⚠️ [ADAPTER WARNING] Primary model {self.default_model} failed: {str(e)}. Attempting Fallback...")
            return self._fallback_routing(prompt, system_prompt, image_path)
            
    def _fallback_routing(self, prompt: str, system_prompt: str, image_path: Optional[str]) -> str:
        fallback_chain = ["gpt-4o", "claude-3-5-sonnet", "qwen-2-vl-local"]
        
        for model in fallback_chain:
            try:
                print(f"🔄 Trying fallback model: {model}...")
                return self.providers[model](prompt, system_prompt, image_path)
            except Exception as e:
                print(f"❌ Fallback {model} failed. ({str(e)})")
                
        raise RuntimeError("CRITICAL: All models in the adapter chain failed.")

    def _call_gemini(self, prompt: str, system_prompt: str, image_path: Optional[str]) -> str:
        """
        Makes a live HTTP request to the Gemini API.
        """
        # Note: We use the single centralized .env in the project root/new_implemetion if available,
        # but os.getenv is sufficient since the main script will load dotenv.
        api_key = os.getenv("GEMINI_API_KEY")
        if not api_key:
            raise ValueError("GEMINI_API_KEY environment variable is not set. Cannot call Gemini.")
            
        url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={api_key}"
        
        # Ensure we request JSON back
        headers = {"Content-Type": "application/json"}
        
        full_prompt = f"{system_prompt}\n\nUser Request: {prompt}"
        
        payload = {
            "contents": [{
                "parts": [{"text": full_prompt}]
            }],
            "generationConfig": {
                "response_mime_type": "application/json"
            }
        }
        
        # If image_path is provided, we would encode it and add it to parts here.
        # For simplicity in this text-based implementation, we just send text.
        
        response = requests.post(url, headers=headers, json=payload)
        
        if response.status_code != 200:
            raise Exception(f"Gemini API Error: {response.text}")
            
        res_json = response.json()
        try:
            raw_text = res_json["candidates"][0]["content"]["parts"][0]["text"]
            return raw_text
        except KeyError:
            raise Exception("Malformed response from Gemini API.")
        
    def _call_openai(self, prompt: str, system_prompt: str, image_path: Optional[str]) -> str:
        raise NotImplementedError("OpenAI fallback not configured yet.")
        
    def _call_anthropic(self, prompt: str, system_prompt: str, image_path: Optional[str]) -> str:
        raise NotImplementedError("Anthropic fallback not configured yet.")
        
    def _call_local_vision(self, prompt: str, system_prompt: str, image_path: Optional[str]) -> str:
        raise NotImplementedError("Local Vision fallback not configured yet.")