leocrm/app/ai/llm_client.py

"""Configurable LLM client — supports LiteLLM (100+ providers) or mock/stub mode.

Reads AI_MODEL, AI_API_KEY, AI_PROVIDER from environment. If not set, uses mock mode
which returns predefined actions based on keyword matching. This allows
tests to run without external API dependencies.

LiteLLM provides a unified interface to OpenAI, Anthropic, Google, Azure,
AWS Bedrock, Ollama, and many more providers.
"""

from __future__ import annotations

import json
import logging
import os
from typing import Any

import litellm

logger = logging.getLogger(__name__)


class LLMResponse:
    """Structured LLM response containing proposed actions."""

    def __init__(
        self, message: str, proposed_actions: list[dict[str, Any]], confidence: float = 0.8
    ):
        self.message = message
        self.proposed_actions = proposed_actions
        self.confidence = confidence

    def to_dict(self) -> dict[str, Any]:
        return {
            "message": self.message,
            "proposed_actions": self.proposed_actions,
            "confidence": self.confidence,
        }


class LLMClient:
    """LLM client that translates natural language to proposed API actions.

    Modes:
    - If AI_MODEL and AI_API_KEY are set: calls LiteLLM chat completions API
    - Otherwise: mock/stub mode with keyword-based action mapping

    LiteLLM model format: "provider/model_name" (e.g. "openai/gpt-4o", "anthropic/claude-3-sonnet", "ollama/llama3")
    """

    def __init__(
        self, model: str | None = None, api_key: str | None = None, api_base: str | None = None,
        provider: str | None = None,
    ) -> None:
        self.model = model or os.environ.get("AI_MODEL", "")
        self.api_key = api_key or os.environ.get("AI_API_KEY", "")
        self.api_base = api_base or os.environ.get("AI_API_BASE", "")
        self.provider = provider or os.environ.get("AI_PROVIDER", "openai")
        self.is_mock = not bool(self.model and self.api_key)

    async def generate(self, user_query: str, context: dict[str, Any] | None = None) -> LLMResponse:
        """Generate proposed actions from natural language query.

        Args:
            user_query: Natural language input from user
            context: Optional context (e.g. current page, selected entity)

        Returns:
            LLMResponse with message and proposed_actions list
        """
        if self.is_mock:
            return await self._mock_generate(user_query, context or {})
        return await self._api_generate(user_query, context or {})

    async def _mock_generate(self, query: str, context: dict[str, Any]) -> LLMResponse:
        """Mock/stub mode — keyword-based action mapping for tests."""
        from app.ai.action_mapper import map_query_to_actions

        actions = map_query_to_actions(query, context)
        if actions:
            return LLMResponse(
                message=f"I found {len(actions)} possible action(s) based on your request.",
                proposed_actions=actions,
                confidence=0.85,
            )
        return LLMResponse(
            message="I couldn't determine a specific action from your request. Could you be more specific?",
            proposed_actions=[],
            confidence=0.3,
        )

    async def _api_generate(self, query: str, context: dict[str, Any]) -> LLMResponse:
        """Call LLM via LiteLLM unified interface.

        Supports 100+ providers through a single API:
        - OpenAI: "openai/gpt-4o"
        - Anthropic: "anthropic/claude-3-sonnet"
        - Google: "gemini/gemini-pro"
        - Azure: "azure/<deployment-name>"
        - Ollama: "ollama/llama3"
        - And many more.
        """
        system_prompt = self._build_system_prompt(context)
        user_prompt = f"User request: {query}\n\nRespond with proposed actions as JSON."

        # Build LiteLLM model string: "provider/model" or just "model" for OpenAI compat
        if self.provider and self.provider != "openai":
            litellm_model = f"{self.provider}/{self.model}"
        else:
            litellm_model = self.model

        # Build kwargs for litellm.acompletion
        kwargs: dict[str, Any] = {
            "model": litellm_model,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            "temperature": 0.3,
            "max_tokens": 1000,
        }

        # Add API key if set
        if self.api_key:
            kwargs["api_key"] = self.api_key

        # Add API base if set (for self-hosted or custom endpoints)
        if self.api_base:
            kwargs["api_base"] = self.api_base

        try:
            response = await litellm.acompletion(**kwargs)
            content = response.choices[0].message.content
            return self._parse_llm_response(content)
        except Exception as e:
            logger.error("LiteLLM API call failed: %s", e)
            # Fall back to mock mode on API error
            return LLMResponse(
                message=f"LLM API call failed: {e}. Falling back to keyword matching.",
                proposed_actions=[],
                confidence=0.1,
            )

    def _build_system_prompt(self, context: dict[str, Any]) -> str:
        """Build system prompt describing available API actions."""
        available_apis = [
            {"method": "GET", "path": "/api/v1/contacts", "description": "List contacts (persons and companies)"},
            {"method": "POST", "path": "/api/v1/contacts", "description": "Create a contact (person or company)"},
            {
                "method": "GET",
                "path": "/api/v1/contacts/{id}",
                "description": "Get contact details",
            },
            {
                "method": "PATCH",
                "path": "/api/v1/contacts/{id}",
                "description": "Update a contact",
            },
            {
                "method": "DELETE",
                "path": "/api/v1/contacts/{id}",
                "description": "Delete a contact",
            },
            {"method": "GET", "path": "/api/v1/workflows", "description": "List workflows"},
            {"method": "POST", "path": "/api/v1/workflows", "description": "Create a workflow"},
            {"method": "GET", "path": "/api/v1/calendar/entries", "description": "List calendar entries"},
            {"method": "POST", "path": "/api/v1/calendar/entries", "description": "Create a calendar entry"},
            {"method": "GET", "path": "/api/v1/dms/files", "description": "List DMS files"},
        ]
        context_str = json.dumps(context) if context else "{}"
        return (
            "You are an AI copilot for LeoCRM. Based on the user's natural language request, "
            "propose one or more API actions. Always respond with a JSON object containing: "
            '"message": a human-readable summary, '
            '"proposed_actions": an array of {method, path, body, description, confidence}. '
            f"Available API endpoints: {json.dumps(available_apis)}. "
            f"Current context: {context_str}. "
            "Never execute actions directly — only propose them for user confirmation."
        )

    def _parse_llm_response(self, content: str) -> LLMResponse:
        """Parse LLM JSON response into LLMResponse."""
        try:
            parsed = json.loads(content)
            return LLMResponse(
                message=parsed.get("message", "Here are the proposed actions."),
                proposed_actions=parsed.get("proposed_actions", []),
                confidence=parsed.get("confidence", 0.8),
            )
        except (json.JSONDecodeError, KeyError):
            logger.warning("Failed to parse LLM response as JSON: %s", content[:200])
            return LLMResponse(
                message=content,
                proposed_actions=[],
                confidence=0.3,
            )


# Global client instance
_client: LLMClient | None = None


def get_llm_client() -> LLMClient:
    """Get or create the global LLM client instance."""
    global _client
    if _client is None:
        _client = LLMClient()
    return _client


def reset_llm_client() -> None:
    """Reset the global client (for testing)."""
    global _client
    _client = None