cloud_llm_service.py

#!/usr/bin/env python3
"""
Generic Cloud LLM Service supporting multiple providers.

Supports:
- Google Gemini (via google-generativeai SDK)
- Moonshot Kimi (OpenAI-compatible API)
- OpenAI (OpenAI-compatible API)
- Anthropic Claude (OpenAI-compatible API)
- DeepSeek (OpenAI-compatible API)
- OpenRouter (aggregator with many free models)
- Custom OpenAI-compatible endpoints
"""

import json
import logging
import os
from abc import ABC, abstractmethod
from datetime import datetime
from typing import TYPE_CHECKING, Dict, Generator, List, Optional, Union


if TYPE_CHECKING:
    from xray_proxy_manager import XrayProxyManager, XrayProxyManagerWithFallback

import httpx


# Gemini SDK (optional)
try:
    import google.generativeai as genai

    GEMINI_AVAILABLE = True
except ImportError:
    GEMINI_AVAILABLE = False
    genai = None

# VLESS Proxy Manager (optional)
try:
    from xray_proxy_manager import (
        XrayProxyManager,
        XrayProxyManagerWithFallback,
        validate_vless_url,
    )

    XRAY_AVAILABLE = True
except ImportError:
    XRAY_AVAILABLE = False
    XrayProxyManager = None
    XrayProxyManagerWithFallback = None
    validate_vless_url = None

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# Provider type configuration (also defined in db/models.py)
PROVIDER_TYPES = {
    "gemini": {
        "name": "Google Gemini",
        "default_base_url": None,
        "default_models": ["gemini-2.0-flash", "gemini-2.5-flash", "gemini-2.5-pro"],
        "requires_base_url": False,
    },
    "kimi": {
        "name": "Moonshot Kimi",
        "default_base_url": "https://api.moonshot.ai/v1",
        "default_models": ["kimi-k2", "moonshot-v1-8k", "moonshot-v1-32k", "moonshot-v1-128k"],
        "requires_base_url": True,
    },
    "openai": {
        "name": "OpenAI",
        "default_base_url": "https://api.openai.com/v1",
        "default_models": ["gpt-4o", "gpt-4o-mini", "gpt-4-turbo"],
        "requires_base_url": True,
    },
    "claude": {
        "name": "Anthropic Claude",
        "default_base_url": "https://api.anthropic.com/v1",
        "default_models": ["claude-opus-4-5-20251101", "claude-sonnet-4-20250514"],
        "requires_base_url": True,
    },
    "deepseek": {
        "name": "DeepSeek",
        "default_base_url": "https://api.deepseek.com/v1",
        "default_models": ["deepseek-chat", "deepseek-coder"],
        "requires_base_url": True,
    },
    "openrouter": {
        "name": "OpenRouter",
        "default_base_url": "https://openrouter.ai/api/v1",
        "default_models": [
            # Free models (январь 2026)
            "nvidia/nemotron-3-nano-30b-a3b:free",
            "nvidia/nemotron-nano-12b-v2-vl:free",
            "arcee-ai/trinity-large-preview:free",
            "arcee-ai/trinity-mini:free",
            "upstage/solar-pro-3:free",
            "liquid/lfm-2.5-1.2b-instruct:free",
            "allenai/molmo-2-8b:free",
            "tngtech/tng-r1t-chimera:free",
            # Paid (дешёвые)
            "google/gemini-2.0-flash-001",
            "openai/gpt-4o-mini",
            "deepseek/deepseek-chat-v3-0324",
        ],
        "requires_base_url": True,
    },
    "custom": {
        "name": "Custom OpenAI-Compatible",
        "default_base_url": "",
        "default_models": [],
        "requires_base_url": True,
    },
    "claude_bridge": {
        "name": "Claude Bridge (Local CLI)",
        "default_base_url": "http://127.0.0.1:8787/v1",
        "default_models": ["sonnet", "opus", "haiku"],
        "requires_base_url": False,
    },
}


class BaseLLMProvider(ABC):
    """Abstract base class for LLM providers."""

    def __init__(self, config: dict):
        self.config = config
        self.api_key = config.get("api_key", "")
        self.model_name = config.get("model_name", "")
        self.base_url = config.get("base_url", "")
        self.provider_id = config.get("id", "unknown")
        self.provider_type = config.get("provider_type", "custom")

        # Runtime parameters
        self.runtime_params = config.get("config", {}) or {}
        if not self.runtime_params:
            self.runtime_params = {
                "temperature": 0.7,
                "max_tokens": 512,
                "top_p": 0.9,
            }

    @abstractmethod
    def generate_response(
        self, user_message: str, system_prompt: str = None, history: List[Dict] = None
    ) -> str:
        """Generate a response synchronously."""
        pass

    @abstractmethod
    def generate_response_stream(
        self, user_message: str, system_prompt: str = None, history: List[Dict] = None
    ) -> Generator[str, None, None]:
        """Generate a response with streaming."""
        pass

    @abstractmethod
    def generate_response_from_messages(
        self, messages: List[Dict[str, str]], stream: bool = False
    ) -> Union[str, Generator[str, None, None]]:
        """Generate response from OpenAI-format messages."""
        pass

    @abstractmethod
    def is_available(self) -> bool:
        """Check if provider is available."""
        pass

    def set_params(self, **kwargs):
        """Set runtime parameters."""
        for key, value in kwargs.items():
            if value is not None:
                self.runtime_params[key] = value
        logger.info(f"[{self.provider_id}] Parameters updated: {self.runtime_params}")

    def get_params(self) -> Dict:
        """Get runtime parameters."""
        return self.runtime_params.copy()


class OpenAICompatibleProvider(BaseLLMProvider):
    """
    Provider for OpenAI-compatible APIs.
    Supports: Kimi (Moonshot), OpenAI, DeepSeek, Claude*, Custom endpoints.

    *Note: Claude has its own API format, but can be used via OpenAI-compatible proxy.
    """

    def __init__(self, config: dict):
        super().__init__(config)

        # Bridge runs on localhost — must bypass global VLESS/HTTP proxy.
        # GeminiProvider sets HTTP_PROXY globally for xray; httpx picks it up
        # and routes localhost requests through the proxy, which fails.
        # Set NO_PROXY BEFORE creating httpx.Client so it respects it.
        if self.provider_type == "claude_bridge":
            self._ensure_no_proxy_for_localhost()

        # Claude bridge needs longer timeouts: CLI warmup (7-30s) + complex
        # prompt processing can exceed 60s before first token arrives.
        # Bridge itself allows 600s per-chunk; match that on client side.
        if self.provider_type == "claude_bridge":
            self.client = httpx.Client(
                timeout=httpx.Timeout(connect=10.0, read=300.0, write=10.0, pool=10.0)
            )
            # Bridge/Claude can handle much longer responses than default 512
            self.runtime_params.setdefault("max_tokens", 4096)
        else:
            self.client = httpx.Client(timeout=60.0)

        # Validate required fields (bridge uses CLI auth, no API key needed)
        if not self.api_key and self.provider_type != "claude_bridge":
            raise ValueError(f"API key required for provider {self.provider_id}")

        # Set default base URL if not provided
        if not self.base_url:
            default_url = PROVIDER_TYPES.get(self.provider_type, {}).get("default_base_url", "")
            if default_url:
                self.base_url = default_url
            else:
                raise ValueError(f"Base URL required for provider {self.provider_id}")

        logger.info(f"[{self.provider_id}] Initialized OpenAI-compatible provider: {self.base_url}")

    @staticmethod
    def _ensure_no_proxy_for_localhost():
        """Add 127.0.0.1 and localhost to NO_PROXY so httpx bypasses VLESS proxy."""
        bypass = {"127.0.0.1", "localhost"}
        for key in ("NO_PROXY", "no_proxy"):
            current = os.environ.get(key, "")
            existing = {h.strip() for h in current.split(",") if h.strip()}
            missing = bypass - existing
            if missing:
                new_val = ",".join(sorted(existing | bypass))
                os.environ[key] = new_val

    def _get_headers(self) -> dict:
        headers: dict = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        return headers

    def is_available(self) -> bool:
        try:
            response = self.client.get(
                f"{self.base_url}/models", headers=self._get_headers(), timeout=10.0
            )
            # 200 = success, 401/403 = auth issue but API reachable
            return response.status_code in [200, 401, 403]
        except Exception as e:
            logger.warning(f"[{self.provider_id}] Health check failed: {e}")
            return False

    def generate_response(
        self, user_message: str, system_prompt: str = None, history: List[Dict] = None
    ) -> str:
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        if history:
            messages.extend(history)
        messages.append({"role": "user", "content": user_message})

        return self._generate_non_stream(messages)

    def generate_response_stream(
        self, user_message: str, system_prompt: str = None, history: List[Dict] = None
    ) -> Generator[str, None, None]:
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        if history:
            messages.extend(history)
        messages.append({"role": "user", "content": user_message})

        yield from self._generate_stream(messages)

    def generate_response_from_messages(
        self, messages: List[Dict[str, str]], stream: bool = False
    ) -> Union[str, Generator[str, None, None]]:
        if stream:
            return self._generate_stream(messages)
        return self._generate_non_stream(messages)

    def _generate_non_stream(self, messages: List[Dict[str, str]]) -> str:
        try:
            response = self.client.post(
                f"{self.base_url}/chat/completions",
                headers=self._get_headers(),
                json={
                    "model": self.model_name,
                    "messages": messages,
                    "temperature": self.runtime_params.get("temperature", 0.7),
                    "max_tokens": self.runtime_params.get("max_tokens", 512),
                    "top_p": self.runtime_params.get("top_p", 0.9),
                    "stream": False,
                },
            )
            response.raise_for_status()
            result = response.json()
            return result["choices"][0]["message"]["content"].strip()
        except httpx.HTTPStatusError as e:
            logger.error(
                f"[{self.provider_id}] HTTP error: {e.response.status_code} - {e.response.text}"
            )
            error_messages = {
                401: "Invalid API key",
                403: "Access denied - check API key permissions",
                404: f"Model '{self.model_name}' not found - check model name",
                429: "Rate limit exceeded - wait or upgrade plan",
                500: "Provider server error",
                502: "Provider gateway error",
                503: "Provider temporarily unavailable",
            }
            msg = error_messages.get(e.response.status_code, f"HTTP {e.response.status_code}")
            return f"Error: {msg}"
        except Exception as e:
            logger.error(f"[{self.provider_id}] Error: {e}")
            return "Извините, произошла техническая ошибка."

    def _generate_stream(self, messages: List[Dict[str, str]]) -> Generator[str, None, None]:
        try:
            with self.client.stream(
                "POST",
                f"{self.base_url}/chat/completions",
                headers=self._get_headers(),
                json={
                    "model": self.model_name,
                    "messages": messages,
                    "temperature": self.runtime_params.get("temperature", 0.7),
                    "max_tokens": self.runtime_params.get("max_tokens", 512),
                    "top_p": self.runtime_params.get("top_p", 0.9),
                    "stream": True,
                },
            ) as response:
                response.raise_for_status()
                for line in response.iter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        if data == "[DONE]":
                            break
                        try:
                            chunk = json.loads(data)
                            delta = chunk["choices"][0].get("delta", {})
                            content = delta.get("content", "")
                            if content:
                                yield content
                        except json.JSONDecodeError:
                            continue
        except Exception as e:
            logger.error(f"[{self.provider_id}] Stream error: {e}")
            yield "Извините, произошла техническая ошибка."


class GeminiProvider(BaseLLMProvider):
    """
    Provider for Google Gemini API.
    Uses the google-generativeai SDK.
    Supports optional VLESS proxy via xray-core with fallback support.
    """

    def __init__(self, config: dict):
        super().__init__(config)

        if not GEMINI_AVAILABLE:
            raise ImportError(
                "google-generativeai package not installed. Install with: pip install google-generativeai"
            )

        if not self.api_key:
            raise ValueError("API key required for Gemini provider")

        # Initialize VLESS proxy BEFORE configuring genai (so HTTP client uses proxy)
        self.proxy_manager: Optional[Union["XrayProxyManager", "XrayProxyManagerWithFallback"]] = (
            None
        )
        self._setup_vless_proxy()

        # Start proxy and set env vars BEFORE genai.configure()
        # This ensures the HTTP client is created with proxy settings
        if self.proxy_manager:
            self.proxy_manager.start()
            proxy_url = f"http://127.0.0.1:{self.proxy_manager.http_port}"
            os.environ["HTTP_PROXY"] = proxy_url
            os.environ["HTTPS_PROXY"] = proxy_url
            os.environ["http_proxy"] = proxy_url
            os.environ["https_proxy"] = proxy_url
            logger.info(f"[{self.provider_id}] Proxy env set: {proxy_url}")

        genai.configure(api_key=self.api_key)
        self.model = genai.GenerativeModel(model_name=self.model_name or "gemini-2.0-flash")

        logger.info(f"[{self.provider_id}] Initialized Gemini provider: {self.model_name}")

    def _setup_vless_proxy(self):
        """Setup VLESS proxy if configured in runtime_params.

        Supports:
        - vless_url: single URL (backward compatible)
        - vless_urls: list of URLs with automatic fallback
        """
        # Check for list of URLs first (new format with fallback)
        vless_urls = self.runtime_params.get("vless_urls", [])
        vless_url = self.runtime_params.get("vless_url", "")

        # Normalize to list
        if vless_urls:
            urls = vless_urls if isinstance(vless_urls, list) else [vless_urls]
        elif vless_url:
            urls = [vless_url]
        else:
            return

        if not XRAY_AVAILABLE:
            logger.warning(
                f"[{self.provider_id}] VLESS URL configured but xray_proxy_manager not available"
            )
            return

        # Use fallback manager if multiple URLs, otherwise simple manager
        if len(urls) > 1:
            self.proxy_manager = XrayProxyManagerWithFallback()
            count = self.proxy_manager.configure_proxies(urls)
            if count > 0:
                logger.info(
                    f"[{self.provider_id}] VLESS proxy configured with {count} fallback servers"
                )
            else:
                logger.warning(f"[{self.provider_id}] No valid VLESS URLs configured")
                self.proxy_manager = None
        else:
            # Single URL - use simple manager
            is_valid, error = validate_vless_url(urls[0])
            if not is_valid:
                logger.error(f"[{self.provider_id}] Invalid VLESS URL: {error}")
                return

            self.proxy_manager = XrayProxyManager()
            if self.proxy_manager.configure(urls[0]):
                logger.info(f"[{self.provider_id}] VLESS proxy configured")
            else:
                logger.warning(f"[{self.provider_id}] Failed to configure VLESS proxy")
                self.proxy_manager = None

    def get_proxy_status(self) -> dict:
        """Get VLESS proxy status."""
        if not self.proxy_manager:
            return {
                "configured": False,
                "xray_available": XRAY_AVAILABLE and XrayProxyManager is not None,
                "fallback_enabled": False,
            }
        status = self.proxy_manager.get_status()
        status["fallback_enabled"] = isinstance(self.proxy_manager, XrayProxyManagerWithFallback)
        return status

    def test_proxy_connection(self, index: int = -1) -> dict:
        """Test VLESS proxy connection to Google API.

        Args:
            index: Proxy index to test (-1 for current/all)
        """
        if not self.proxy_manager:
            return {"success": False, "error": "No VLESS proxy configured"}

        if isinstance(self.proxy_manager, XrayProxyManagerWithFallback):
            if index >= 0:
                return self.proxy_manager.test_proxy(index)
            return {"results": self.proxy_manager.test_all_proxies()}

        return self.proxy_manager.test_connection("https://generativelanguage.googleapis.com")

    def mark_proxy_failed(self):
        """Mark current proxy as failed and switch to next (fallback mode only)."""
        if isinstance(self.proxy_manager, XrayProxyManagerWithFallback):
            self.proxy_manager.mark_current_failed()

    def reset_proxies(self):
        """Reset all proxies to enabled state (fallback mode only)."""
        if isinstance(self.proxy_manager, XrayProxyManagerWithFallback):
            self.proxy_manager.reset_all_proxies()

    def is_available(self) -> bool:
        """Check if provider is available. Proxy is already running if configured."""
        try:
            # Proxy env vars are set in __init__, just test the API
            list(genai.list_models())
            return True
        except Exception as e:
            logger.warning(f"[{self.provider_id}] Health check failed: {e}")
            return False

    def generate_response(
        self, user_message: str, system_prompt: str = None, history: List[Dict] = None
    ) -> str:
        """Generate response. Proxy is already running if configured."""
        try:
            # Rebuild model with system instruction if provided
            if system_prompt:
                model = genai.GenerativeModel(
                    model_name=self.model_name or "gemini-2.0-flash",
                    system_instruction=system_prompt,
                )
            else:
                model = self.model

            # Convert history to Gemini format
            gemini_history = []
            if history:
                for msg in history:
                    role = "model" if msg["role"] == "assistant" else msg["role"]
                    if role not in ["user", "model"]:
                        continue
                    gemini_history.append({"role": role, "parts": [msg["content"]]})

            chat = model.start_chat(history=gemini_history)
            response = chat.send_message(user_message)
            return response.text.strip()
        except Exception as e:
            logger.error(f"[{self.provider_id}] Error: {e}")
            return "Извините, произошла техническая ошибка."

    def generate_response_stream(
        self, user_message: str, system_prompt: str = None, history: List[Dict] = None
    ) -> Generator[str, None, None]:
        """Generate streaming response. Proxy is already running if configured."""
        try:
            if system_prompt:
                model = genai.GenerativeModel(
                    model_name=self.model_name or "gemini-2.0-flash",
                    system_instruction=system_prompt,
                )
            else:
                model = self.model

            gemini_history = []
            if history:
                for msg in history:
                    role = "model" if msg["role"] == "assistant" else msg["role"]
                    if role not in ["user", "model"]:
                        continue
                    gemini_history.append({"role": role, "parts": [msg["content"]]})

            chat = model.start_chat(history=gemini_history)
            response = chat.send_message(user_message, stream=True)

            for chunk in response:
                if chunk.text:
                    yield chunk.text
        except Exception as e:
            logger.error(f"[{self.provider_id}] Stream error: {e}")
            yield "Извините, произошла техническая ошибка."

    def generate_response_from_messages(
        self, messages: List[Dict[str, str]], stream: bool = False
    ) -> Union[str, Generator[str, None, None]]:
        # Extract system prompt and convert to Gemini format
        system_prompt = None
        history = []
        last_user = ""

        for msg in messages:
            if msg["role"] == "system":
                system_prompt = msg["content"]
            elif msg["role"] == "user":
                last_user = msg["content"]
                history.append({"role": "user", "content": msg["content"]})
            elif msg["role"] == "assistant":
                history.append({"role": "assistant", "content": msg["content"]})

        # Remove last user message from history (will be sent)
        if history and history[-1]["role"] == "user":
            history = history[:-1]

        if stream:
            return self.generate_response_stream(last_user, system_prompt, history)
        return self.generate_response(last_user, system_prompt, history)


class CloudLLMService:
    """
    Main service class for cloud LLM providers.
    Manages provider instances and provides unified interface.

    Compatible with LLMService and VLLMLLMService interfaces.
    """

    # Provider class mapping
    PROVIDER_CLASSES = {
        "gemini": GeminiProvider,
        "kimi": OpenAICompatibleProvider,
        "openai": OpenAICompatibleProvider,
        "claude": OpenAICompatibleProvider,
        "deepseek": OpenAICompatibleProvider,
        "openrouter": OpenAICompatibleProvider,
        "custom": OpenAICompatibleProvider,
        "claude_bridge": OpenAICompatibleProvider,
    }

    def __init__(self, provider_config: dict):
        """
        Initialize with provider configuration from database.

        Args:
            provider_config: Dict with id, provider_type, api_key, base_url, model_name, config
        """
        self.config = provider_config
        self.provider_type = provider_config.get("provider_type", "custom")
        self.provider_id = provider_config.get("id", "unknown")

        # Get provider class and instantiate
        provider_class = self.PROVIDER_CLASSES.get(self.provider_type, OpenAICompatibleProvider)
        self.provider: BaseLLMProvider = provider_class(provider_config)

        # For compatibility with existing code
        self.model_name = provider_config.get("model_name", "")
        self.api_url = provider_config.get("base_url", "")
        self.backend_type = "cloud"  # Отличает от vLLM

        # FAQ (загружается через reload_faq из БД)
        self.faq: Dict[str, str] = {}

        # Conversation history
        self.conversation_history: List[Dict[str, str]] = []

        # System prompt (for secretary persona)
        self.system_prompt = provider_config.get("system_prompt", "")

        logger.info(f"CloudLLMService initialized: {self.provider_id} ({self.provider_type})")

    def _normalize_faq(self, faq_dict: Dict[str, str]) -> Dict[str, str]:
        """Нормализует ключи FAQ (lowercase, strip)"""
        return {k.lower().strip(): v for k, v in faq_dict.items()}

    def _check_faq(self, user_message: str) -> Optional[str]:
        if not self.faq:
            return None
        normalized = user_message.lower().strip().rstrip("?!.,")
        if normalized in self.faq:
            return self._apply_faq_templates(self.faq[normalized])
        for key, response in self.faq.items():
            if key in normalized or normalized in key:
                return self._apply_faq_templates(response)
        return None

    def _apply_faq_templates(self, response: str) -> str:
        now = datetime.now()
        replacements = {
            "{current_time}": now.strftime("%H:%M"),
            "{current_date}": now.strftime("%d.%m.%Y"),
            "{day_of_week}": [
                "понедельник",
                "вторник",
                "среда",
                "четверг",
                "пятница",
                "суббота",
                "воскресенье",
            ][now.weekday()],
        }
        for placeholder, value in replacements.items():
            response = response.replace(placeholder, value)
        return response

    def reload_faq(self, faq_dict: Dict[str, str] = None):
        """
        Перезагружает FAQ (hot reload).

        Args:
            faq_dict: FAQ словарь из БД. Если не передан, FAQ очищается.
        """
        if faq_dict:
            self.faq = self._normalize_faq(faq_dict)
        else:
            self.faq = {}
        logger.info(f"🔄 FAQ перезагружен: {len(self.faq)} записей")

    def get_system_prompt(self) -> str:
        """Return the system prompt configured for this provider."""
        return self.system_prompt or ""

    def is_available(self) -> bool:
        """Check if provider is available."""
        return self.provider.is_available()

    def generate_response(self, user_message: str, use_history: bool = True) -> str:
        """Generate response (compatible with LLMService/VLLMLLMService)."""
        # Check FAQ first
        faq_response = self._check_faq(user_message)
        if faq_response:
            if use_history:
                self.conversation_history.append({"role": "user", "content": user_message})
                self.conversation_history.append({"role": "assistant", "content": faq_response})
            return faq_response

        history = self.conversation_history if use_history else []
        response = self.provider.generate_response(user_message, self.system_prompt, history)

        if use_history:
            self.conversation_history.append({"role": "user", "content": user_message})
            self.conversation_history.append({"role": "assistant", "content": response})

        return response

    def generate_response_stream(
        self, user_message: str, use_history: bool = True
    ) -> Generator[str, None, None]:
        """Generate streaming response (compatible with LLMService/VLLMLLMService)."""
        # Check FAQ first
        faq_response = self._check_faq(user_message)
        if faq_response:
            if use_history:
                self.conversation_history.append({"role": "user", "content": user_message})
                self.conversation_history.append({"role": "assistant", "content": faq_response})
            yield faq_response
            return

        history = self.conversation_history if use_history else []
        full_response = ""

        for chunk in self.provider.generate_response_stream(
            user_message, self.system_prompt, history
        ):
            full_response += chunk
            yield chunk

        if use_history and full_response:
            self.conversation_history.append({"role": "user", "content": user_message})
            self.conversation_history.append({"role": "assistant", "content": full_response})

    def generate_response_from_messages(
        self, messages: List[Dict[str, str]], stream: bool = False
    ) -> Union[str, Generator[str, None, None]]:
        """Generate response from OpenAI-format messages (compatible with orchestrator)."""
        # Check FAQ for single-message requests
        user_messages = [m for m in messages if m.get("role") == "user"]
        if len(user_messages) == 1:
            faq_response = self._check_faq(user_messages[0]["content"])
            if faq_response:
                if stream:

                    def gen():
                        yield faq_response

                    return gen()
                return faq_response

        return self.provider.generate_response_from_messages(messages, stream)

    def reset_conversation(self):
        """Clear conversation history."""
        self.conversation_history = []

    def get_conversation_history(self) -> List[Dict[str, str]]:
        """Get conversation history."""
        return self.conversation_history

    def set_params(self, **kwargs):
        """Set runtime parameters."""
        self.provider.set_params(**kwargs)

    def get_params(self) -> Dict:
        """Get runtime parameters."""
        return self.provider.get_params()

    # For compatibility with VLLMLLMService persona system
    @property
    def runtime_params(self) -> Dict:
        return self.provider.runtime_params