from __future__ import annotations import logging from typing import Any import httpx logger = logging.getLogger("dscons.llm.openai") class OpenAiBackendMixin: """OpenAI-compatible inference backend mixin (Ollama/vLLM/DeepSeek).""" def get_provider_for_agent(self, agent_code: str | None) -> dict[str, Any]: """Resolve base_url, api_key, model, temperature, and max_tokens for a specific AI agent/persona.""" db_config = None if agent_code: try: from app.core.postgres.erp_client import ErpDatabaseClient client = ErpDatabaseClient() db_config = client.get_ai_agent_model(agent_code) except Exception as e: logger.debug("Không thể đọc cấu hình AI Agent từ DB: %s", e) # Kiểm tra tự động chuyển đổi Fallback nếu mô hình Stealth đã hết hạn if db_config and db_config.get("is_expired"): fallback_model = db_config.get( "fallback_model_name", "anthropic/claude-opus-4.8" ) fallback_prov = db_config.get("fallback_provider", "openrouter") logger.warning( "[STEALTH EXPIRED] Agent %s: Model '%s' đã hết hạn ưu đãi. Tự động chuyển về Fallback '%s' (%s).", agent_code, db_config.get("model_name"), fallback_model, fallback_prov, ) provider_key = fallback_prov model_override = fallback_model else: routing = getattr(self._settings, "agent_provider_routing", {}) provider_key = ( db_config.get("provider") if db_config else None ) or routing.get(agent_code or "default", "openrouter") model_routing = getattr(self._settings, "agent_model_routing", {}) model_override = (db_config.get("model_name") if db_config else None) or ( model_routing.get(agent_code) if agent_code else None ) temperature = (db_config.get("temperature") if db_config else None) or 0.20 max_tokens = ( db_config.get("max_tokens") if db_config else None ) or self._default_max_tokens if provider_key == "lmstudio" and getattr( self._settings, "lmstudio_enabled", True ): return { "base_url": getattr( self._settings, "lmstudio_base_url", "http://127.0.0.1:1234/v1" ).rstrip("/"), "api_key": getattr(self._settings, "lmstudio_api_key", "lm-studio"), "model": model_override or getattr(self._settings, "lmstudio_model_name", "lfm2.5-2.6b"), "provider": "lmstudio", "temperature": temperature, "max_tokens": max_tokens, } elif ( provider_key == "openrouter" and getattr(self._settings, "openrouter_enabled", True) and getattr(self._settings, "openrouter_api_key", "") ): clean_model = model_override or getattr(self._settings, "openrouter_model_name", "stealth/ox-alpha") if clean_model.startswith("gemini-"): clean_model = f"google/{clean_model}" if clean_model in ("google/gemini-3.1-pro", "gemini-3.1-pro"): clean_model = "google/gemini-3.8-flash" return { "base_url": getattr( self._settings, "openrouter_base_url", "https://openrouter.ai/api/v1", ).rstrip("/"), "api_key": getattr(self._settings, "openrouter_api_key", ""), "model": clean_model, "provider": "openrouter", "temperature": temperature, "max_tokens": max_tokens, } elif ( provider_key == "gemini" ) and getattr(self._settings, "gemini_api_key", ""): return { "base_url": getattr( self._settings, "gemini_base_url", "https://generativelanguage.googleapis.com/v1beta/openai", ).rstrip("/"), "api_key": getattr(self._settings, "gemini_api_key", ""), "model": model_override if ( model_override and not model_override.startswith("liquid/") and not model_override.startswith("stealth/") ) else getattr(self._settings, "gemini_model_name", "gemini-3.8-flash"), "provider": "gemini", "temperature": temperature, "max_tokens": max_tokens or 4096, } elif provider_key in ("antigravity", "antigravity_sdk"): import os clean_model = model_override or getattr(self._settings, "gemini_model_name", "gemini-3.8-flash") if clean_model in ("gemini-3.1-pro", "gemini-2.5-flash"): clean_model = "gemini-3.8-flash" return { "base_url": "", "api_key": getattr(self._settings, "gemini_api_key", "") or os.environ.get("GEMINI_API_KEY", ""), "model": clean_model, "provider": "antigravity", "temperature": temperature, "max_tokens": max_tokens or 4096, } # Fallback to default fallback_model = model_override or self._model_path or "stealth/ox-alpha" if "openrouter" in self._api_base_url.lower(): if fallback_model.startswith("gemini-"): fallback_model = f"google/{fallback_model}" if fallback_model in ("google/gemini-3.1-pro", "gemini-3.1-pro"): fallback_model = "google/gemini-3.8-flash" return { "base_url": self._api_base_url, "api_key": self._api_key, "model": fallback_model, "provider": self._active_provider, "temperature": temperature, "max_tokens": max_tokens, } def _strip_multimodal_content( self, messages: list[dict[str, Any]] ) -> list[dict[str, Any]]: """Strip image_url blocks and convert multimodal messages to text-only for text LLMs.""" clean_messages: list[dict[str, Any]] = [] for msg in messages: c = msg.get("content") if isinstance(c, list): text_parts: list[str] = [] for part in c: if isinstance(part, dict): if part.get("type") == "text": text_parts.append(str(part.get("text", ""))) elif part.get("type") == "image_url": text_parts.append( "[Hình ảnh bản vẽ kỹ thuật CAD / PDF thiết kế thi công]" ) elif isinstance(part, str): text_parts.append(part) clean_messages.append( {"role": msg.get("role", "user"), "content": "\n".join(text_parts)} ) else: clean_messages.append(msg) return clean_messages async def _generate_openai_compatible( self, messages: list[dict[str, Any]], *, model: str | None = None, temperature: float = 0.2, max_tokens: int | None = None, extra_body: dict[str, Any] | None = None, agent_code: str | None = None, ) -> dict[str, Any]: """Call OpenAI-compatible HTTP API with per-agent provider resolution.""" provider_info = self.get_provider_for_agent(agent_code) base_url = provider_info["base_url"] api_key = provider_info["api_key"] effective_model = model or provider_info["model"] url = f"{base_url}/chat/completions" payload = { "model": effective_model, "messages": messages, "temperature": temperature if temperature != 0.2 else provider_info.get("temperature", 0.2), "max_tokens": max_tokens or provider_info.get("max_tokens", self._default_max_tokens), } if extra_body: payload.update(extra_body) headers = { "Authorization": f"Bearer {api_key}", "HTTP-Referer": "https://github.com/dscons", "X-Title": "DSCons Operations", } def _normalize_resp(data: dict[str, Any]) -> dict[str, Any]: if "choices" in data and len(data["choices"]) > 0: msg = data["choices"][0].get("message", {}) if not msg.get("content"): if msg.get("reasoning"): msg["content"] = msg["reasoning"] elif ( msg.get("reasoning_details") and len(msg["reasoning_details"]) > 0 ): msg["content"] = msg["reasoning_details"][0].get("text", "") return data async with httpx.AsyncClient(timeout=self._timeout) as client: try: response = await client.post(url, json=payload, headers=headers) response.raise_for_status() return _normalize_resp(response.json()) except Exception as primary_err: has_images = any(isinstance(m.get("content"), list) for m in messages) if has_images: vision_fallbacks = [ "anthropic/claude-opus-4.8", "anthropic/claude-opus-4.8", "google/gemini-3.1-pro-preview", ] for fb_v_model in vision_fallbacks: if fb_v_model == effective_model: continue try: logger.info( "[LLM_CLIENT] Thu lai voi model thi giac '%s'...", fb_v_model, ) fb_payload = dict(payload) fb_payload["model"] = fb_v_model fb_res = await client.post( url, json=fb_payload, headers=headers, timeout=60.0 ) fb_res.raise_for_status() return _normalize_resp(fb_res.json()) except Exception as fb_err: logger.debug( "[LLM_CLIENT] Vision fallback '%s' failed: %s", fb_v_model, fb_err, ) text_fallbacks = [ "google/gemini-3.8-flash", "anthropic/claude-opus-4.8", "anthropic/claude-fable-5", "google/gemini-3.1-pro-preview", ] clean_msgs = self._strip_multimodal_content(messages) for fb_model in text_fallbacks: if fb_model == effective_model: continue try: logger.warning( "Primary model '%s' failed (%s). Retrying with text fallback '%s'...", effective_model, primary_err, fb_model, ) fallback_payload = dict(payload) fallback_payload["model"] = fb_model fallback_payload["messages"] = clean_msgs fb_res = await client.post( url, json=fallback_payload, headers=headers, timeout=60.0 ) fb_res.raise_for_status() return _normalize_resp(fb_res.json()) except Exception as fb_err: logger.debug( "Fallback text model '%s' failed: %s", fb_model, fb_err ) raise primary_err