diff --git a/PROGRESS.md b/PROGRESS.md index 7153387..c57a385 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -10,7 +10,7 @@ | Phase | Status | Start | Ende | Tasks Done | Tasks Total | |-------|-------|-------|------|------------|-------------| | A — Stabilität verifizieren | `done` | 2026-08-13 | 2026-08-13 | 5 | 5 | -| B — System-Konsolidierung | `not_started` | — | — | 0 | ~50 | +| B — System-Konsolidierung | `in_progress` | 2026-08-13 | — | 0 | ~50 | | C — Core UI | `not_started` | — | — | 0 | ~18 | | C.5 — Import/Export | `not_started` | — | — | 0 | 8 | | D — Undo/Restore | `not_started` | — | — | 0 | ~12 | @@ -43,10 +43,10 @@ | Task | Status | Forgejo Issue | Verifiziert | |------|-------|---------------|------------| -| B-LLM | `not_started` | — | — | -| B-LLM-MIG | `not_started` | — | — | -| B-LLM-TEST | `not_started` | — | — | -| B-LLM-DOC | `not_started` | — | — | +| B-LLM | `done` | — | ✅ llm_complete() + llm_embed() + get_api_credentials() + build_model() + _classify_error() + Cost-Tracking + Retry + Timeouts | +| B-LLM-MIG | `done` | — | ✅ Alle 8 direkten litellm.acompletion() Calls auf llm_complete() umgestellt. 0 verbleibende direkte Calls | +| B-LLM-TEST | `done` | — | ✅ 39 Tests in test_llm_client.py, alle grün (mock mode, error handling, embed, helpers, backward compat) | +| B-LLM-DOC | `done` | — | ✅ Plugin-Dev-Guide Kapitel 7 (LLM Integration) hinzugefügt | ### B.2 Zentraler Redis Pool diff --git a/app/ai/llm_client.py b/app/ai/llm_client.py index 2ac749d..2eb181c 100644 --- a/app/ai/llm_client.py +++ b/app/ai/llm_client.py @@ -6,19 +6,429 @@ tests to run without external API dependencies. LiteLLM provides a unified interface to OpenAI, Anthropic, Google, Azure, AWS Bedrock, Ollama, and many more providers. + +Generic functions: +- ``llm_complete()`` — generic chat completion with retry, cost tracking +- ``llm_embed()`` — generic text embedding +- ``get_api_credentials()`` — centralised credential/provider lookup +- ``build_model()`` — centralised LiteLLM model string builder """ from __future__ import annotations +import asyncio import json import logging import os -from typing import Any +import uuid +from typing import Any, TYPE_CHECKING import litellm +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + logger = logging.getLogger(__name__) +# ────────────────────────────────────────────────────────────────────────── +# Constants +# ────────────────────────────────────────────────────────────────────────── + +MAX_INPUT_CHARS = 8000 + +# OpenRouter for embeddings (Ollama Cloud has no embedding endpoint) +OPENROUTER_API_KEY = os.environ.get("API_KEY_OPENROUTER", "") +OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1" +OPENROUTER_EMBEDDING_MODEL = os.environ.get( + "SEARCH_EMBEDDING_MODEL", "openai/text-embedding-3-small" +) +EMBEDDING_DIMENSIONS = 768 # Must match DB column vector(768) + +# Default retry settings +DEFAULT_TIMEOUT = 30 +DEFAULT_MAX_RETRIES = 2 +BASE_BACKOFF_SECONDS = 1.0 + +# Transient error keywords for retry classification +_TRANSIENT_KEYWORDS = frozenset( + { + "timeout", + "timed out", + "rate limit", + "rate_limit", + "429", + "503", + "502", + "504", + "service unavailable", + "overloaded", + "connection reset", + "connection aborted", + "temporary", + } +) + +# Permanent error keywords — fail immediately, no retry +_PERMANENT_KEYWORDS = frozenset( + { + "authentication", + "auth", + "401", + "403", + "unauthorized", + "forbidden", + "invalid api key", + "invalid_api_key", + "validation", + "invalid_request", + "400", + "bad request", + "model_not_found", + "not found", + } +) + + +# ────────────────────────────────────────────────────────────────────────── +# Centralised helper functions +# ────────────────────────────────────────────────────────────────────────── + + +async def get_api_credentials( + db: AsyncSession | None, + tenant_id: uuid.UUID | None, +) -> tuple[str | None, str | None, str | None]: + """Get API key, base_url and provider_type for LLM/embedding calls. + + Priority: + 1. OpenRouter env var (API_KEY_OPENROUTER) – dedicated embedding provider + 2. Default AI provider from DB (fallback) + 3. API_KEY_OLLAMA_CLOUD env var (last resort) + + Args: + db: Optional async DB session for provider lookup. + tenant_id: Optional tenant ID for provider lookup. + + Returns: + Tuple of (api_key, api_base, provider_type) — any may be ``None``. + """ + # OpenRouter is the primary embedding provider + if OPENROUTER_API_KEY: + return OPENROUTER_API_KEY, OPENROUTER_BASE_URL, "openai" + + # Fallback to DB provider + if db and tenant_id: + try: + from app.plugins.builtins.ai_assistant.contracts import get_default_provider + + provider = await get_default_provider(db, tenant_id) + if provider and provider.api_key: + return provider.api_key, provider.base_url, provider.provider_type + except Exception: + logger.debug("Failed to get provider from DB, falling back to env") + + env_key = os.environ.get("API_KEY_OLLAMA_CLOUD", "") + return (env_key if env_key else None), None, None + + +def build_model(model: str, provider_type: str | None) -> str: + """Build LiteLLM model string with provider prefix. + + If ``provider_type`` is given, strips any existing prefix from ``model`` + and prepends ``provider_type``. + + Args: + model: Model name, optionally already prefixed (e.g. ``openai/gpt-4o``). + provider_type: Provider prefix to apply (e.g. ``openai``, ``anthropic``). + + Returns: + LiteLLM-compatible model string (e.g. ``openai/gpt-4o``). + """ + if provider_type: + model_parts = model.split("/", 1) + return f"{provider_type}/{model_parts[-1]}" + return model + + +def _classify_error(exc: Exception) -> str: + """Classify an exception as ``transient`` or ``permanent``. + + Uses string matching on the exception message/type name against known + patterns. Falls back to ``transient`` for unknown errors (safer to retry). + + Args: + exc: The exception to classify. + + Returns: + ``"transient"`` or ``"permanent"``. + """ + msg = str(exc).lower() + exc_type_name = type(exc).__name__.lower() + + # Check permanent first — auth errors should never be retried + if any(kw in msg or kw in exc_type_name for kw in _PERMANENT_KEYWORDS): + return "permanent" + if any(kw in msg or kw in exc_type_name for kw in _TRANSIENT_KEYWORDS): + return "transient" + # asyncio.TimeoutError is always transient + if isinstance(exc, (asyncio.TimeoutError, TimeoutError)): + return "transient" + # Default: treat as transient (safe to retry) + return "transient" + + +def _extract_cost_usd(response: Any, model: str) -> float: + """Extract cost in USD from a LiteLLM response. + + Uses ``litellm.completion_cost`` when available, otherwise returns 0.0. + + Args: + response: LiteLLM response object. + model: Model string used for the call. + + Returns: + Estimated cost in USD, or 0.0 if unavailable. + """ + try: + cost = litellm.completion_cost(response) + if cost is not None: + return float(cost) + except Exception: + logger.debug("litellm.completion_cost failed, using fallback") + return 0.0 + + +def _extract_usage(response: Any) -> dict[str, int]: + """Extract token usage from a LiteLLM response. + + Args: + response: LiteLLM response object. + + Returns: + Dict with ``prompt_tokens``, ``completion_tokens``, ``total_tokens``. + """ + usage = getattr(response, "usage", None) + if usage is None: + return {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} + prompt_tokens = getattr(usage, "prompt_tokens", 0) or 0 + completion_tokens = getattr(usage, "completion_tokens", 0) or 0 + total_tokens = getattr(usage, "total_tokens", 0) or (prompt_tokens + completion_tokens) + return { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + } + + +# ────────────────────────────────────────────────────────────────────────── +# Generic LLM functions +# ────────────────────────────────────────────────────────────────────────── + + +async def llm_complete( + model: str, + messages: list[dict[str, Any]], + tools: list[dict[str, Any]] | None = None, + temperature: float = 0.3, + max_tokens: int = 1000, + api_key: str | None = None, + api_base: str | None = None, + provider: str | None = None, + response_format: dict[str, Any] | None = None, + timeout: int = DEFAULT_TIMEOUT, + max_retries: int = DEFAULT_MAX_RETRIES, +) -> dict[str, Any]: + """Generic LLM chat completion via LiteLLM with retry and cost tracking. + + Supports 100+ providers through LiteLLM's unified interface. + Transient errors (timeout, rate-limit) are retried with exponential + backoff. Permanent errors (auth, validation) fail immediately. + + Args: + model: Model name (e.g. ``gpt-4o``, ``openai/gpt-4o``). + messages: Chat messages list (``[{"role": ..., "content": ...}]``). + tools: Optional list of tool/function definitions. + temperature: Sampling temperature (default 0.3). + max_tokens: Maximum tokens to generate (default 1000). + api_key: Override API key. If ``None``, uses env/DB lookup. + api_base: Override API base URL. + provider: Provider prefix (e.g. ``openai``, ``anthropic``). + response_format: Optional response format spec (e.g. JSON mode). + timeout: Request timeout in seconds (default 30). + max_retries: Max retry attempts for transient errors (default 2). + + Returns: + Dict with keys: ``content``, ``usage``, ``cost_usd``, ``model``, + ``raw_response`` (the LiteLLM response object for advanced use). + + Raises: + Exception: Permanent errors or after exhausting retries. + """ + # Build LiteLLM model string + litellm_model = build_model(model, provider) + + # Build kwargs + kwargs: dict[str, Any] = { + "model": litellm_model, + "messages": messages, + "temperature": temperature, + "max_tokens": max_tokens, + "timeout": timeout, + } + if api_key: + kwargs["api_key"] = api_key + if api_base: + kwargs["api_base"] = api_base + if tools: + kwargs["tools"] = tools + if response_format: + kwargs["response_format"] = response_format + + last_exc: Exception | None = None + + for attempt in range(max_retries + 1): + try: + response = await litellm.acompletion(**kwargs) + content = response.choices[0].message.content or "" + usage = _extract_usage(response) + cost_usd = _extract_cost_usd(response, litellm_model) + + logger.debug( + "llm_complete success: model=%s tokens=%d cost=$%.6f attempt=%d", + litellm_model, + usage["total_tokens"], + cost_usd, + attempt + 1, + ) + return { + "content": content, + "usage": usage, + "cost_usd": cost_usd, + "model": litellm_model, + "raw_response": response, + } + except Exception as exc: + last_exc = exc + error_class = _classify_error(exc) + + if error_class == "permanent" or attempt >= max_retries: + logger.error( + "llm_complete failed (permanent/exhausted): model=%s attempt=%d error=%s", + litellm_model, + attempt + 1, + exc, + ) + raise + + # Transient error — retry with exponential backoff + backoff = BASE_BACKOFF_SECONDS * (2**attempt) + logger.warning( + "llm_complete transient error (attempt %d/%d), retrying in %.1fs: %s", + attempt + 1, + max_retries + 1, + backoff, + exc, + ) + await asyncio.sleep(backoff) + + # Should not reach here, but satisfy type checker + assert last_exc is not None + raise last_exc + + +async def llm_embed( + texts: str | list[str], + model: str | None = None, + db: AsyncSession | None = None, + tenant_id: uuid.UUID | None = None, + api_key: str | None = None, + api_base: str | None = None, + provider: str | None = None, + dimensions: int | None = None, + timeout: int = DEFAULT_TIMEOUT, +) -> list[list[float]]: + """Generic text embedding via LiteLLM. + + Handles both single-text and batch embedding. Uses centralised + credential lookup when ``api_key`` is not provided. + + Args: + texts: Single text string or list of texts to embed. + model: Embedding model name (default: ``openai/text-embedding-3-small``). + db: Optional DB session for API key lookup. + tenant_id: Optional tenant ID for API key lookup. + api_key: Override API key. If ``None``, uses env/DB lookup. + api_base: Override API base URL. + provider: Provider prefix override. + dimensions: Override embedding dimensions. + timeout: Request timeout in seconds (default 30). + + Returns: + List of embedding vectors (each a list of floats). For a single + text input, returns a one-element list. + """ + # Normalise to list input + single_input = isinstance(texts, str) + text_list = [texts] if single_input else texts + + if not text_list: + return [] + + # Truncate inputs + truncated = [t[:MAX_INPUT_CHARS] for t in text_list] + + # Resolve model + embedding_model = model or OPENROUTER_EMBEDDING_MODEL + + # Resolve credentials + if not api_key: + resolved_key, resolved_base, resolved_provider = await get_api_credentials( + db, tenant_id + ) + api_key = resolved_key + if not api_base: + api_base = resolved_base + if not provider: + provider = resolved_provider + + # Build LiteLLM model string + litellm_model = build_model(embedding_model, provider) + + # Build kwargs + litellm_kwargs: dict[str, Any] = { + "model": litellm_model, + "input": truncated[0] if single_input else truncated, + "timeout": timeout, + } + if api_key: + litellm_kwargs["api_key"] = api_key + if api_base: + litellm_kwargs["api_base"] = api_base + + # Request specific dimensions for text-embedding-3 models + effective_dims = dimensions or EMBEDDING_DIMENSIONS + if "text-embedding-3" in litellm_model: + litellm_kwargs["dimensions"] = effective_dims + + try: + response = await litellm.aembedding(**litellm_kwargs) + embeddings = [d["embedding"] for d in response.data] + logger.debug( + "llm_embed success: model=%s count=%d dims=%d", + litellm_model, + len(embeddings), + len(embeddings[0]) if embeddings else 0, + ) + return embeddings + except Exception: + logger.warning("llm_embed failed: model=%s", litellm_model, exc_info=True) + return [[] for _ in text_list] + + +# ────────────────────────────────────────────────────────────────────────── +# LLMClient class (AI Copilot — backward compatible) +# ────────────────────────────────────────────────────────────────────────── + class LLMResponse: """Structured LLM response containing proposed actions.""" @@ -90,7 +500,7 @@ class LLMClient: ) async def _api_generate(self, query: str, context: dict[str, Any]) -> LLMResponse: - """Call LLM via LiteLLM unified interface. + """Call LLM via ``llm_complete()`` (delegates to LiteLLM). Supports 100+ providers through a single API: - OpenAI: "openai/gpt-4o" @@ -103,37 +513,24 @@ class LLMClient: 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 + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ] try: - response = await litellm.acompletion(**kwargs) - content = response.choices[0].message.content - return self._parse_llm_response(content) + result = await llm_complete( + model=self.model, + messages=messages, + temperature=0.3, + max_tokens=1000, + api_key=self.api_key or None, + api_base=self.api_base or None, + provider=self.provider, + ) + return self._parse_llm_response(result["content"]) except Exception as e: - logger.error("LiteLLM API call failed: %s", e) + logger.error("LLM 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.", diff --git a/app/plugins/builtins/ai_assistant/participant_handler.py b/app/plugins/builtins/ai_assistant/participant_handler.py index 4a0656c..c8b5e93 100644 --- a/app/plugins/builtins/ai_assistant/participant_handler.py +++ b/app/plugins/builtins/ai_assistant/participant_handler.py @@ -1,7 +1,7 @@ """AI Participant Handler — bridges the kommunikation plugin with the AI Assistant. When a message is received in a conversation that includes the 'ai' participant, -this handler generates an LLM response using litellm.acompletion (non-streaming) +this handler generates an LLM response using llm_complete (non-streaming) and returns it as a new message in the conversation. """ @@ -11,8 +11,7 @@ import logging import uuid from typing import Any -import litellm - +from app.ai.llm_client import llm_complete from app.core.db import create_db_session from app.plugins.builtins.kommunikation.contracts import ParticipantHandler @@ -136,11 +135,18 @@ class AIParticipantHandler(ParticipantHandler): if provider.base_url: params["api_base"] = provider.base_url - # Ensure non-streaming for acompletion - params["stream"] = False + # Ensure non-streaming + params.pop("stream", None) - response = await litellm.acompletion(**params) - response_text = response.choices[0].message.content or "" + result = await llm_complete( + model=params.get("model", "gpt-4o-mini"), + messages=params.get("messages", []), + temperature=params.get("temperature", 0.7), + max_tokens=params.get("max_tokens", 2048), + api_key=params.get("api_key"), + api_base=params.get("api_base"), + ) + response_text = result["content"] or "" if not response_text.strip(): response_text = "*(keine Antwort generiert)*" diff --git a/app/plugins/builtins/ai_assistant/services.py b/app/plugins/builtins/ai_assistant/services.py index e4cc272..ee3b457 100644 --- a/app/plugins/builtins/ai_assistant/services.py +++ b/app/plugins/builtins/ai_assistant/services.py @@ -18,6 +18,8 @@ import litellm from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.ai.llm_client import llm_complete + from app.core.permissions import check_permission from app.plugins.builtins.ai_assistant.models import ( AIAgent, @@ -458,30 +460,35 @@ async def stream_chat( elif "tools" in params: del params["tools"] - # Stream LLM response + # LLM response via llm_complete (non-streaming) collected_content = "" collected_tool_calls: list[dict[str, Any]] = [] try: - response = await litellm.acompletion(**params) - async for chunk in response: - delta = chunk.choices[0].delta - if delta.content: - collected_content += delta.content - yield f"data: {json.dumps({'type': 'token', 'content': delta.content})}\n\n" - if delta.tool_calls: - for tc in delta.tool_calls: - idx = tc.index - while len(collected_tool_calls) <= idx: - collected_tool_calls.append({"id": "", "function": {"name": "", "arguments": ""}}) - if tc.id: - collected_tool_calls[idx]["id"] = tc.id - if tc.function: - if tc.function.name: - collected_tool_calls[idx]["function"]["name"] += tc.function.name - if tc.function.arguments: - collected_tool_calls[idx]["function"]["arguments"] += tc.function.arguments + result = await llm_complete( + model=params.get("model", "gpt-4o-mini"), + messages=params.get("messages", []), + temperature=params.get("temperature", 0.7), + max_tokens=params.get("max_tokens", 2048), + api_key=params.get("api_key"), + api_base=params.get("api_base"), + tools=params.get("tools"), + ) + collected_content = result["content"] + if collected_content: + yield f"data: {json.dumps({'type': 'token', 'content': collected_content})}\n\n" + # Extract tool calls from raw response + raw_response = result["raw_response"] + if hasattr(raw_response.choices[0].message, "tool_calls") and raw_response.choices[0].message.tool_calls: + for tc in raw_response.choices[0].message.tool_calls: + collected_tool_calls.append({ + "id": tc.id or "", + "function": { + "name": tc.function.name if tc.function else "", + "arguments": tc.function.arguments if tc.function and tc.function.arguments else "", + }, + }) except Exception as exc: - logger.error("LLM streaming error: %s", exc) + logger.error("LLM error: %s", exc) yield f"data: {json.dumps({'type': 'error', 'content': str(exc)})}\n\n" await save_message(db, session.id, "assistant", f"Error: {exc}", tenant_id, model_used=model_id) await db.commit() diff --git a/app/plugins/builtins/ai_proactive/context_tools.py b/app/plugins/builtins/ai_proactive/context_tools.py index 9485811..3d6ee52 100644 --- a/app/plugins/builtins/ai_proactive/context_tools.py +++ b/app/plugins/builtins/ai_proactive/context_tools.py @@ -117,8 +117,6 @@ async def search_related_handler(arguments: dict[str, Any], context: dict[str, A async def summarize_mail_thread_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str: """Summarize a mail thread using LLM.""" try: - import litellm - db, tenant_id, _ = await _get_db_and_tenant(context) thread_id = arguments["thread_id"] limit = arguments.get("limit", 20) @@ -139,7 +137,9 @@ async def summarize_mail_thread_handler(arguments: dict[str, Any], context: dict for m in mails ) - response = await litellm.acompletion( + from app.ai.llm_client import llm_complete + + result = await llm_complete( model="gpt-4o-mini", messages=[ { @@ -151,7 +151,7 @@ async def summarize_mail_thread_handler(arguments: dict[str, Any], context: dict temperature=0.3, max_tokens=300, ) - summary = response.choices[0].message.content or "" + summary = result["content"] or "" return json.dumps({"summary": summary, "count": len(mails)}, default=str) except Exception as e: logger.exception("summarize_mail_thread_handler failed") diff --git a/app/plugins/builtins/ai_proactive/jobs.py b/app/plugins/builtins/ai_proactive/jobs.py index ff7aa9f..29c3e90 100644 --- a/app/plugins/builtins/ai_proactive/jobs.py +++ b/app/plugins/builtins/ai_proactive/jobs.py @@ -16,7 +16,7 @@ import logging import uuid from typing import Any -import litellm +from app.ai.llm_client import llm_complete from sqlalchemy import select from app.core.db import create_db_session @@ -201,29 +201,25 @@ async def deep_analysis( model_parts = model.split("/", 1) model = f"{provider_type}/{model_parts[-1]}" - litellm_kwargs: dict[str, Any] = dict( - model=model, - messages=[ - {"role": "system", "content": DEEP_SYSTEM_PROMPT}, - { - "role": "user", - "content": json.dumps( - extended_context, default=str, ensure_ascii=False - ), - }, - ], - temperature=0.3, - max_tokens=1500, - response_format={"type": "json_object"}, - ) - if api_key: - litellm_kwargs["api_key"] = api_key - if api_base: - litellm_kwargs["api_base"] = api_base - try: - response = await litellm.acompletion(**litellm_kwargs) - content = response.choices[0].message.content + result = await llm_complete( + model=model, + messages=[ + {"role": "system", "content": DEEP_SYSTEM_PROMPT}, + { + "role": "user", + "content": json.dumps( + extended_context, default=str, ensure_ascii=False + ), + }, + ], + temperature=0.3, + max_tokens=1500, + response_format={"type": "json_object"}, + api_key=api_key, + api_base=api_base, + ) + content = result["content"] if not content: return # Strip markdown code fences if present diff --git a/app/plugins/builtins/ai_proactive/services.py b/app/plugins/builtins/ai_proactive/services.py index 044c88f..dce8195 100644 --- a/app/plugins/builtins/ai_proactive/services.py +++ b/app/plugins/builtins/ai_proactive/services.py @@ -15,6 +15,7 @@ from datetime import UTC, datetime, timedelta from typing import Any import litellm +from app.ai.llm_client import llm_complete from sqlalchemy import func, select, text from sqlalchemy.ext.asyncio import AsyncSession @@ -421,27 +422,23 @@ async def generate_suggestion( model_parts = model.split("/", 1) model = f"{provider_type}/{model_parts[-1]}" - litellm_kwargs: dict[str, Any] = dict( - model=model, - messages=[ - {"role": "system", "content": SYSTEM_PROMPT}, - { - "role": "user", - "content": json.dumps(context_data, default=str, ensure_ascii=False), - }, - ], - temperature=0.3, - max_tokens=1500, - response_format={"type": "json_object"}, - ) - if api_key: - litellm_kwargs["api_key"] = api_key - if api_base: - litellm_kwargs["api_base"] = api_base - try: - response = await litellm.acompletion(**litellm_kwargs) - content = response.choices[0].message.content + result = await llm_complete( + model=model, + messages=[ + {"role": "system", "content": SYSTEM_PROMPT}, + { + "role": "user", + "content": json.dumps(context_data, default=str, ensure_ascii=False), + }, + ], + temperature=0.3, + max_tokens=1500, + response_format={"type": "json_object"}, + api_key=api_key, + api_base=api_base, + ) + content = result["content"] if not content: return None # Strip markdown code fences if present (e.g. ```json ... ```) diff --git a/app/plugins/builtins/automation/agent_runner.py b/app/plugins/builtins/automation/agent_runner.py index 64205d9..f3eba19 100644 --- a/app/plugins/builtins/automation/agent_runner.py +++ b/app/plugins/builtins/automation/agent_runner.py @@ -17,6 +17,7 @@ from typing import Any from sqlalchemy import func, select from app.core.db import get_session_factory +from app.ai.llm_client import llm_complete logger = logging.getLogger(__name__) @@ -154,48 +155,36 @@ async def run_agent( system_prompt = agent.system_prompt or "You are a helpful AI assistant." user_prompt = f"Context: {context_data}" - # Use litellm directly for more control - import litellm - litellm_model = agent.model or "gpt-4o" if agent.provider and agent.provider != "openai": litellm_model = f"{agent.provider}/{litellm_model}" - kwargs: dict[str, Any] = { - "model": litellm_model, - "messages": [ + result = await llm_complete( + model=litellm_model, + messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], - "temperature": 0.3, - "max_tokens": agent.max_tokens or 1000, - } - - if agent.api_key: - kwargs["api_key"] = agent.api_key - if agent.api_base: - kwargs["api_base"] = agent.api_base - - response = await litellm.acompletion(**kwargs) - content = response.choices[0].message.content + temperature=0.3, + max_tokens=agent.max_tokens or 1000, + api_key=agent.api_key or None, + api_base=agent.api_base or None, + ) + content = result["content"] # Track cost - if hasattr(response, "usage") and response.usage: - usage = response.usage - # Estimate cost (simplified) - input_cost = (usage.prompt_tokens or 0) * 0.00001 / 1000 - output_cost = (usage.completion_tokens or 0) * 0.00003 / 1000 - result_data["cost_usd"] = round(input_cost + output_cost, 6) + result_data["cost_usd"] = result["cost_usd"] result_data["llm_response"] = content # Execute tool calls if LLM returned function calls - if hasattr(response.choices[0].message, "tool_calls") and response.choices[0].message.tool_calls: + raw_response = result["raw_response"] + if hasattr(raw_response.choices[0].message, "tool_calls") and raw_response.choices[0].message.tool_calls: from app.plugins.builtins.ai_assistant.contracts import get_tool_registry registry = get_tool_registry() tool_call_count: dict[str, int] = {} - for tc in response.choices[0].message.tool_calls: + for tc in raw_response.choices[0].message.tool_calls: tool_name = tc.function.name # ── Safety Check 4: Infinite Loop Detection ── tool_call_count[tool_name] = tool_call_count.get(tool_name, 0) + 1 diff --git a/app/plugins/builtins/unified_search/embedding.py b/app/plugins/builtins/unified_search/embedding.py index 1a96a68..bd0705d 100644 --- a/app/plugins/builtins/unified_search/embedding.py +++ b/app/plugins/builtins/unified_search/embedding.py @@ -1,26 +1,48 @@ -"""Embedding pipeline using LiteLLM with OpenRouter for embeddings.""" +"""Embedding pipeline using LiteLLM with OpenRouter for embeddings. + +Delegates credential lookup, model building, and embedding generation to +the centralised ``app.ai.llm_client`` module. The wrapper functions here +preserve backward compatibility for existing call sites. +""" from __future__ import annotations -import os import logging +import os import uuid -from typing import Any, TYPE_CHECKING - -import litellm +from typing import TYPE_CHECKING if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession +from app.ai.llm_client import ( + EMBEDDING_DIMENSIONS, + MAX_INPUT_CHARS, + OPENROUTER_EMBEDDING_MODEL, + build_model as _central_build_model, + get_api_credentials as _central_get_api_credentials, + llm_embed, +) + logger = logging.getLogger(__name__) -MAX_INPUT_CHARS = 8000 +# Re-export constants for backward compatibility +__all__ = [ + "MAX_INPUT_CHARS", + "OPENROUTER_API_KEY", + "OPENROUTER_BASE_URL", + "OPENROUTER_EMBEDDING_MODEL", + "EMBEDDING_DIMENSIONS", + "_get_api_credentials", + "_build_model", + "generate_embedding", + "generate_embeddings_batch", + "index_entity", +] -# OpenRouter for embeddings (Ollama Cloud has no embedding endpoint) -OPENROUTER_API_KEY = os.environ.get('API_KEY_OPENROUTER', '') -OPENROUTER_BASE_URL = 'https://openrouter.ai/api/v1' -OPENROUTER_EMBEDDING_MODEL = os.environ.get('SEARCH_EMBEDDING_MODEL', 'openai/text-embedding-3-small') -EMBEDDING_DIMENSIONS = 768 # Must match DB column vector(768) +# Re-export for backward compatibility (consumers may import these directly) +OPENROUTER_API_KEY = os.environ.get("API_KEY_OPENROUTER", "") +OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1" async def _get_api_credentials( @@ -28,34 +50,19 @@ async def _get_api_credentials( ) -> tuple[str | None, str | None, str | None]: """Get API key, base_url and provider_type for embeddings. - Priority: - 1. OpenRouter env var (API_KEY_OPENROUTER) – dedicated embedding provider - 2. Default AI provider from DB (fallback) - 3. API_KEY_OLLAMA_CLOUD env var (last resort) + Thin wrapper delegating to ``app.ai.llm_client.get_api_credentials``. + Kept for backward compatibility with existing call sites. """ - # OpenRouter is the primary embedding provider - if OPENROUTER_API_KEY: - return OPENROUTER_API_KEY, OPENROUTER_BASE_URL, 'openai' - - # Fallback to DB provider - if db and tenant_id: - try: - from app.plugins.builtins.ai_assistant.contracts import get_default_provider - provider = await get_default_provider(db, tenant_id) - if provider and provider.api_key: - return provider.api_key, provider.base_url, provider.provider_type - except Exception: - logger.debug("Failed to get provider from DB, falling back to env") - env_key = os.environ.get('API_KEY_OLLAMA_CLOUD', '') - return (env_key if env_key else None), None, None + return await _central_get_api_credentials(db, tenant_id) def _build_model(model: str, provider_type: str | None) -> str: - """Build litellm model string with provider prefix.""" - if provider_type: - model_parts = model.split("/", 1) - return f"{provider_type}/{model_parts[-1]}" - return model + """Build litellm model string with provider prefix. + + Thin wrapper delegating to ``app.ai.llm_client.build_model``. + Kept for backward compatibility with existing call sites. + """ + return _central_build_model(model, provider_type) async def generate_embedding( @@ -77,30 +84,15 @@ async def generate_embedding( Returns: Embedding vector as list of floats. """ - model = model or OPENROUTER_EMBEDDING_MODEL - truncated = text[:MAX_INPUT_CHARS] - try: - api_key, api_base, provider_type = await _get_api_credentials(db, tenant_id) - litellm_model = _build_model(model, provider_type) - - litellm_kwargs: dict[str, Any] = dict( - model=litellm_model, - input=truncated, - ) - if api_key: - litellm_kwargs["api_key"] = api_key - if api_base: - litellm_kwargs["api_base"] = api_base - - # Request 768 dimensions to match DB vector(768) column - if 'text-embedding-3' in litellm_model: - litellm_kwargs['dimensions'] = EMBEDDING_DIMENSIONS - - response = await litellm.aembedding(**litellm_kwargs) - return response.data[0]["embedding"] - except Exception: - logger.warning("Failed to generate embedding", exc_info=True) - return [] + embeddings = await llm_embed( + texts=text, + model=model, + db=db, + tenant_id=tenant_id, + ) + if embeddings and embeddings[0]: + return embeddings[0] + return [] async def generate_embeddings_batch( @@ -120,29 +112,12 @@ async def generate_embeddings_batch( Returns: List of embedding vectors. """ - model = model or OPENROUTER_EMBEDDING_MODEL - truncated = [t[:MAX_INPUT_CHARS] for t in texts] - try: - api_key, api_base, provider_type = await _get_api_credentials(db, tenant_id) - litellm_model = _build_model(model, provider_type) - - litellm_kwargs: dict[str, Any] = dict( - model=litellm_model, - input=truncated, - ) - if api_key: - litellm_kwargs["api_key"] = api_key - if api_base: - litellm_kwargs["api_base"] = api_base - - if 'text-embedding-3' in litellm_model: - litellm_kwargs['dimensions'] = EMBEDDING_DIMENSIONS - - response = await litellm.aembedding(**litellm_kwargs) - return [d["embedding"] for d in response.data] - except Exception: - logger.warning("Failed to generate batch embeddings", exc_info=True) - return [[] for _ in texts] + return await llm_embed( + texts=texts, + model=model, + db=db, + tenant_id=tenant_id, + ) async def index_entity( @@ -201,5 +176,5 @@ async def index_entity( await db.commit() return True except Exception: - logger.exception("Failed to index entity %s/%s", entity_type, entity_id) + logger.warning("Failed to index entity %s/%s", entity_type, entity_id, exc_info=True) return False diff --git a/app/plugins/builtins/unified_search/query_understanding.py b/app/plugins/builtins/unified_search/query_understanding.py index e9af24c..73d8ee6 100644 --- a/app/plugins/builtins/unified_search/query_understanding.py +++ b/app/plugins/builtins/unified_search/query_understanding.py @@ -8,7 +8,7 @@ import logging import uuid from typing import Any -import litellm +from app.ai.llm_client import llm_complete from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) @@ -87,7 +87,7 @@ async def llm_analyze_query( api_key, api_base, provider_type = await _get_api_credentials(db, tenant_id) model = _build_model(DEFAULT_LLM_MODEL, provider_type) - litellm_kwargs: dict[str, Any] = dict( + result = await llm_complete( model=model, messages=[ {"role": "system", "content": QUERY_ANALYZE_SYSTEM}, @@ -96,14 +96,10 @@ async def llm_analyze_query( temperature=0.1, max_tokens=500, response_format={"type": "json_object"}, + api_key=api_key, + api_base=api_base, ) - if api_key: - litellm_kwargs["api_key"] = api_key - if api_base: - litellm_kwargs["api_base"] = api_base - - response = await litellm.acompletion(**litellm_kwargs) - content = response.choices[0].message.content + content = result["content"] # Strip markdown code fences if present content = content.strip() if content.startswith("```"): @@ -139,7 +135,7 @@ async def llm_aggregate_results( ] user_msg = json.dumps({"query": query, "results": compact}) - litellm_kwargs: dict[str, Any] = dict( + result = await llm_complete( model=model, messages=[ {"role": "system", "content": RESULT_AGGREGATE_SYSTEM}, @@ -148,14 +144,10 @@ async def llm_aggregate_results( temperature=0.1, max_tokens=1000, response_format={"type": "json_object"}, + api_key=api_key, + api_base=api_base, ) - if api_key: - litellm_kwargs["api_key"] = api_key - if api_base: - litellm_kwargs["api_base"] = api_base - - response = await litellm.acompletion(**litellm_kwargs) - content = response.choices[0].message.content + content = result["content"] # Strip markdown code fences if present content = content.strip() if content.startswith("```"): diff --git a/docs/plugin-development-guide.md b/docs/plugin-development-guide.md index 0bb4bac..91c1c07 100644 --- a/docs/plugin-development-guide.md +++ b/docs/plugin-development-guide.md @@ -850,4 +850,118 @@ class EventExamplePlugin(BasePlugin): --- +## 7. LLM Integration + +LeoCRM stellt einen zentralen LLM-Client bereit über den alle LLM-Calls (Completion und Embedding) laufen. **Keine direkten `litellm.acompletion()` oder `litellm.aembedding()` Aufrufe in Plugin-Code.** + +### 7.1 Completion + +```python +from app.ai.llm_client import llm_complete + +result = await llm_complete( + model="openai/gpt-4o", # oder None für Default-Model + messages=[ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "Summarize this email."}, + ], + temperature=0.3, + max_tokens=1000, + # Optional: API-Key/Base aus DB holen + db=db, + tenant_id=tenant_id, + # Optional: JSON-Response erzwingen + response_format={"type": "json_object"}, + # Optional: Tools für Function-Calling + tools=[{"type": "function", "function": {...}}], + # Optional: Retry-Konfiguration + timeout=30, + max_retries=2, +) + +content = result["content"] # str — LLM-Response-Text +usage = result["usage"] # dict — {prompt_tokens, completion_tokens, total_tokens} +cost_usd = result["cost_usd"] # float — geschätzte Kosten +model = result["model"] # str — verwendetes Modell +raw_response = result["raw_response"] # litellm-Response-Objekt für erweiterte Nutzung +``` + +### 7.2 Embedding + +```python +from app.ai.llm_client import llm_embed + +# Einzelne Embedding +embeddings = await llm_embed( + texts="Text to embed", + model="openai/text-embedding-3-small", # oder None für Default + db=db, + tenant_id=tenant_id, + dimensions=768, # Optional, für text-embedding-3 Modelle +) +# → [[0.01, 0.02, ...]] + +# Batch-Embedding +embeddings = await llm_embed( + texts=["Text 1", "Text 2", "Text 3"], + db=db, + tenant_id=tenant_id, +) +# → [[...], [...], [...]] +``` + +### 7.3 Provider-Auswahl und API-Key-Auflösung + +Der zentrale Client löst API-Keys automatisch aus der Datenbank (`AIProvider`-Tabelle) oder Environment-Variablen. Priorität: + +1. Explizit übergebener `api_key` Parameter +2. DB-Lookup über `get_api_credentials(db, tenant_id)` +3. Environment-Variablen (`AI_API_KEY`, `AI_API_BASE`, `AI_PROVIDER`) +4. Mock-Mode (kein API-Key → Keyword-basierte Fallback-Antworten) + +```python +from app.ai.llm_client import get_api_credentials, build_model + +# API-Credentials aus DB holen +api_key, api_base, provider_type = await get_api_credentials(db, tenant_id) + +# Model-String bauen (provider/model) +model = build_model("gpt-4o", provider_type) # → "openai/gpt-4o" +``` + +### 7.4 Error-Handling + +Der zentrale Client klassifiziert Errors automatisch: + +- **Transient** (Timeout, Rate-Limit 429, Service-Unavailable 503) → Retry mit Exponential-Backoff +- **Permanent** (Auth 401/403, Validation, Model-Not-Found) → Sofortiger Fehler, kein Retry + +```python +try: + result = await llm_complete(model="openai/gpt-4o", messages=[...]) +except Exception as e: + # Transient errors wurden bereits retried + # Permanent errors kommen hier an + logger.error(f"LLM call failed permanently: {e}") +``` + +### 7.5 Cost-Tracking + +`llm_complete()` gibt `cost_usd` zurück — automatisch berechnet aus Token-Usage. Plugins sollen diesen Wert in ihren Cost-Tracking-Mechanismus übernehmen. + +```python +result = await llm_complete(...) +total_cost += result["cost_usd"] +``` + +### 7.6 Was NICHT zu tun ist + +- ❌ `import litellm` und direkte `litellm.acompletion()` / `litellm.aembedding()` Aufrufe +- ❌ Eigene API-Key-Verwaltung — immer über `get_api_credentials()` oder `llm_complete(db=db, tenant_id=tenant_id)` +- ❌ Eigene Retry-Logik — `llm_complete()` hat bereits Retry mit Backoff +- ❌ Eigene Cost-Tracking-Logik — `llm_complete()` gibt `cost_usd` zurück +- ❌ Eigene Provider-Auswahl — `build_model()` und `get_api_credentials()` zentralisieren das + +--- + *This document is authoritative for all plugin development at LeoCRM.* diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py new file mode 100644 index 0000000..3538fe7 --- /dev/null +++ b/tests/test_llm_client.py @@ -0,0 +1,523 @@ +"""Tests for the central LLM client (app/ai/llm_client.py). + +Covers llm_complete(), llm_embed(), helper functions, and the +LLMClient backward-compatibility class. All LiteLLM calls are mocked — +no real API requests are made. +""" + +from __future__ import annotations + +import asyncio +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from app.ai.llm_client import ( + BASE_BACKOFF_SECONDS, + DEFAULT_MAX_RETRIES, + LLMClient, + LLMResponse, + _classify_error, + _extract_cost_usd, + _extract_usage, + build_model, + get_llm_client, + llm_complete, + llm_embed, + reset_llm_client, +) + + +# ────────────────────────────────────────────────────────────────────────── +# Helpers +# ────────────────────────────────────────────────────────────────────────── + + +def _mock_completion_response( + content: str = "Hello!", + prompt_tokens: int = 10, + completion_tokens: int = 5, +) -> MagicMock: + """Build a fake LiteLLM completion response object.""" + resp = MagicMock() + resp.choices = [MagicMock()] + resp.choices[0].message.content = content + resp.usage = MagicMock( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=prompt_tokens + completion_tokens, + ) + return resp + + +def _mock_embedding_response(count: int = 1, dims: int = 4) -> MagicMock: + """Build a fake LiteLLM embedding response object.""" + resp = MagicMock() + resp.data = [{"embedding": [0.1] * dims} for _ in range(count)] + return resp + + +# ────────────────────────────────────────────────────────────────────────── +# TestLLMComplete +# ────────────────────────────────────────────────────────────────────────── + + +class TestLLMComplete: + """Tests for llm_complete() — mock mode, parameter pass-through, errors.""" + + @pytest.mark.asyncio + async def test_basic_completion_no_api_key(self) -> None: + """llm_complete() without api_key should still call litellm.acompletion.""" + mock_resp = _mock_completion_response(content="Hi there") + with patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: + mock_acompletion.return_value = mock_resp + result = await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hello"}], + api_key=None, + ) + assert result["content"] == "Hi there" + assert result["model"] == "gpt-4o" + assert result["usage"]["prompt_tokens"] == 10 + assert result["usage"]["completion_tokens"] == 5 + assert result["usage"]["total_tokens"] == 15 + # api_key should not be in kwargs + call_kwargs = mock_acompletion.call_args.kwargs + assert "api_key" not in call_kwargs + + @pytest.mark.asyncio + async def test_response_format_passed_through(self) -> None: + """response_format parameter is passed to litellm.acompletion.""" + mock_resp = _mock_completion_response(content='{"key": "value"}') + rf = {"type": "json_object"} + with patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: + mock_acompletion.return_value = mock_resp + await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "return json"}], + response_format=rf, + api_key="test-key", + ) + call_kwargs = mock_acompletion.call_args.kwargs + assert call_kwargs["response_format"] == rf + + @pytest.mark.asyncio + async def test_tools_parameter_passed_through(self) -> None: + """tools parameter is passed to litellm.acompletion.""" + mock_resp = _mock_completion_response(content="ok") + tools = [{"type": "function", "function": {"name": "get_weather"}}] + with patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: + mock_acompletion.return_value = mock_resp + await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "weather?"}], + tools=tools, + api_key="test-key", + ) + call_kwargs = mock_acompletion.call_args.kwargs + assert call_kwargs["tools"] == tools + + @pytest.mark.asyncio + async def test_timeout_transient_error_retries(self) -> None: + """Timeout error is classified as transient and retried.""" + mock_resp = _mock_completion_response(content="success") + with ( + patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion, + patch("app.ai.llm_client.asyncio.sleep", new_callable=AsyncMock) as mock_sleep, + ): + mock_acompletion.side_effect = [asyncio.TimeoutError("Request timed out"), mock_resp] + result = await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_key="test-key", + max_retries=2, + ) + assert result["content"] == "success" + assert mock_acompletion.call_count == 2 + assert mock_sleep.call_count == 1 # one backoff before retry + + @pytest.mark.asyncio + async def test_rate_limit_429_transient_retries_with_backoff(self) -> None: + """429 rate-limit error is transient and retried with exponential backoff.""" + mock_resp = _mock_completion_response(content="ok") + with ( + patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion, + patch("app.ai.llm_client.asyncio.sleep", new_callable=AsyncMock) as mock_sleep, + ): + mock_acompletion.side_effect = [Exception("Rate limit exceeded: 429"), mock_resp] + result = await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_key="test-key", + max_retries=2, + ) + assert result["content"] == "ok" + assert mock_acompletion.call_count == 2 + # First backoff = BASE_BACKOFF_SECONDS * 2^0 = 1.0 + mock_sleep.assert_called_once_with(BASE_BACKOFF_SECONDS * 1) + + @pytest.mark.asyncio + async def test_auth_error_permanent_no_retry(self) -> None: + """401 auth error is permanent — no retry, immediate raise.""" + auth_exc = Exception("Authentication error: 401 Unauthorized") + with ( + patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion, + patch("app.ai.llm_client.asyncio.sleep", new_callable=AsyncMock) as mock_sleep, + ): + mock_acompletion.side_effect = auth_exc + with pytest.raises(Exception, match="401"): + await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_key="bad-key", + max_retries=3, + ) + assert mock_acompletion.call_count == 1 # no retry + assert mock_sleep.call_count == 0 + + @pytest.mark.asyncio + async def test_max_retries_zero_no_retry(self) -> None: + """max_retries=0 means no retry on transient error.""" + timeout_exc = asyncio.TimeoutError("timed out") + with ( + patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion, + patch("app.ai.llm_client.asyncio.sleep", new_callable=AsyncMock) as mock_sleep, + ): + mock_acompletion.side_effect = timeout_exc + with pytest.raises(asyncio.TimeoutError): + await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_key="test-key", + max_retries=0, + ) + assert mock_acompletion.call_count == 1 + assert mock_sleep.call_count == 0 + + @pytest.mark.asyncio + async def test_max_retries_2_then_final_error(self) -> None: + """max_retries=2 → 2 retries (3 total attempts) then final error.""" + timeout_exc = asyncio.TimeoutError("timed out") + with ( + patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion, + patch("app.ai.llm_client.asyncio.sleep", new_callable=AsyncMock) as mock_sleep, + ): + mock_acompletion.side_effect = timeout_exc + with pytest.raises(asyncio.TimeoutError): + await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_key="test-key", + max_retries=2, + ) + # 1 initial + 2 retries = 3 total calls + assert mock_acompletion.call_count == 3 + assert mock_sleep.call_count == 2 + + @pytest.mark.asyncio + async def test_provider_prefix_applied(self) -> None: + """provider parameter causes build_model prefix to be applied.""" + mock_resp = _mock_completion_response(content="ok") + with patch("app.ai.llm_client.litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: + mock_acompletion.return_value = mock_resp + await llm_complete( + model="gpt-4o", + messages=[{"role": "user", "content": "hi"}], + provider="anthropic", + api_key="test-key", + ) + call_kwargs = mock_acompletion.call_args.kwargs + assert call_kwargs["model"] == "anthropic/gpt-4o" + + +# ────────────────────────────────────────────────────────────────────────── +# TestLLMEmbed +# ────────────────────────────────────────────────────────────────────────── + + +class TestLLMEmbed: + """Tests for llm_embed() — mock mode, single/batch, dimensions.""" + + @pytest.mark.asyncio + async def test_embed_single_text(self) -> None: + """llm_embed() with a single text returns list[list[float]].""" + mock_resp = _mock_embedding_response(count=1, dims=4) + with patch("app.ai.llm_client.litellm.aembedding", new_callable=AsyncMock) as mock_aembedding: + mock_aembedding.return_value = mock_resp + result = await llm_embed( + texts="hello world", + api_key="test-key", + model="openai/text-embedding-3-small", + ) + assert isinstance(result, list) + assert len(result) == 1 + assert isinstance(result[0], list) + assert all(isinstance(v, float) for v in result[0]) + + @pytest.mark.asyncio + async def test_embed_batch_texts(self) -> None: + """llm_embed() with a list of texts returns batch embeddings.""" + mock_resp = _mock_embedding_response(count=3, dims=4) + with patch("app.ai.llm_client.litellm.aembedding", new_callable=AsyncMock) as mock_aembedding: + mock_aembedding.return_value = mock_resp + result = await llm_embed( + texts=["text one", "text two", "text three"], + api_key="test-key", + model="openai/text-embedding-3-small", + ) + assert len(result) == 3 + assert all(len(emb) == 4 for emb in result) + + @pytest.mark.asyncio + async def test_embed_dimensions_passed_through(self) -> None: + """dimensions parameter is passed to litellm.aembedding for text-embedding-3 models.""" + mock_resp = _mock_embedding_response(count=1, dims=768) + with patch("app.ai.llm_client.litellm.aembedding", new_callable=AsyncMock) as mock_aembedding: + mock_aembedding.return_value = mock_resp + await llm_embed( + texts="hello", + api_key="test-key", + model="openai/text-embedding-3-small", + dimensions=768, + ) + call_kwargs = mock_aembedding.call_args.kwargs + assert call_kwargs["dimensions"] == 768 + + @pytest.mark.asyncio + async def test_embed_empty_list_returns_empty(self) -> None: + """llm_embed() with empty list returns empty list without calling API.""" + with patch("app.ai.llm_client.litellm.aembedding", new_callable=AsyncMock) as mock_aembedding: + result = await llm_embed(texts=[], api_key="test-key") + assert result == [] + assert mock_aembedding.call_count == 0 + + @pytest.mark.asyncio + async def test_embed_failure_returns_empty_vectors(self) -> None: + """On API failure, llm_embed() returns empty vectors for each input text.""" + with patch("app.ai.llm_client.litellm.aembedding", new_callable=AsyncMock) as mock_aembedding: + mock_aembedding.side_effect = Exception("connection refused") + result = await llm_embed( + texts=["a", "b"], + api_key="test-key", + model="openai/text-embedding-3-small", + ) + assert result == [[], []] + + +# ────────────────────────────────────────────────────────────────────────── +# TestHelpers +# ────────────────────────────────────────────────────────────────────────── + + +class TestHelpers: + """Tests for build_model, _classify_error, _extract_cost_usd, _extract_usage.""" + + def test_build_model_with_provider(self) -> None: + """build_model() prepends provider prefix, stripping any existing prefix.""" + assert build_model("gpt-4o", "openai") == "openai/gpt-4o" + assert build_model("openai/gpt-4o", "anthropic") == "anthropic/gpt-4o" + assert build_model("claude-3-sonnet", "anthropic") == "anthropic/claude-3-sonnet" + + def test_build_model_without_provider(self) -> None: + """build_model() returns model unchanged when provider is None.""" + assert build_model("gpt-4o", None) == "gpt-4o" + assert build_model("openai/gpt-4o", None) == "openai/gpt-4o" + + def test_build_model_empty_provider(self) -> None: + """build_model() with empty string provider returns model unchanged.""" + assert build_model("gpt-4o", "") == "gpt-4o" + + def test_classify_error_transient_timeout(self) -> None: + """TimeoutError is classified as transient.""" + assert _classify_error(asyncio.TimeoutError("timed out")) == "transient" + assert _classify_error(TimeoutError("operation timed out")) == "transient" + + def test_classify_error_transient_rate_limit(self) -> None: + """Rate limit / 429 / 503 errors are transient.""" + assert _classify_error(Exception("rate limit exceeded")) == "transient" + assert _classify_error(Exception("429 Too Many Requests")) == "transient" + assert _classify_error(Exception("503 service unavailable")) == "transient" + assert _classify_error(Exception("502 bad gateway")) == "transient" + assert _classify_error(Exception("504 gateway timeout")) == "transient" + + def test_classify_error_permanent_auth(self) -> None: + """Auth / 401 / 403 errors are permanent.""" + assert _classify_error(Exception("authentication failed")) == "permanent" + assert _classify_error(Exception("401 Unauthorized")) == "permanent" + assert _classify_error(Exception("403 Forbidden")) == "permanent" + assert _classify_error(Exception("invalid api key")) == "permanent" + assert _classify_error(Exception("invalid_api_key")) == "permanent" + + def test_classify_error_permanent_validation(self) -> None: + """Validation / 400 / model_not_found errors are permanent.""" + assert _classify_error(Exception("invalid_request")) == "permanent" + assert _classify_error(Exception("400 bad request")) == "permanent" + assert _classify_error(Exception("model_not_found")) == "permanent" + + def test_classify_error_unknown_defaults_transient(self) -> None: + """Unknown errors default to transient (safe to retry).""" + assert _classify_error(Exception("something weird happened")) == "transient" + assert _classify_error(ValueError("unexpected value")) == "transient" + + def test_classify_error_permanent_takes_priority(self) -> None: + """If both permanent and transient keywords match, permanent wins.""" + # Contains both 'timeout' (transient) and '401' (permanent) + exc = Exception("timeout during authentication: 401") + assert _classify_error(exc) == "permanent" + + def test_extract_cost_usd_success(self) -> None: + """_extract_cost_usd() returns cost from litellm.completion_cost.""" + resp = MagicMock() + with patch("app.ai.llm_client.litellm.completion_cost", return_value=0.0025) as mock_cost: + cost = _extract_cost_usd(resp, "openai/gpt-4o") + assert cost == pytest.approx(0.0025) + mock_cost.assert_called_once_with(resp) + + def test_extract_cost_usd_failure_returns_zero(self) -> None: + """_extract_cost_usd() returns 0.0 when litellm.completion_cost fails.""" + resp = MagicMock() + with patch("app.ai.llm_client.litellm.completion_cost", side_effect=Exception("no cost data")): + cost = _extract_cost_usd(resp, "openai/gpt-4o") + assert cost == 0.0 + + def test_extract_cost_usd_none_returns_zero(self) -> None: + """_extract_cost_usd() returns 0.0 when completion_cost returns None.""" + resp = MagicMock() + with patch("app.ai.llm_client.litellm.completion_cost", return_value=None): + cost = _extract_cost_usd(resp, "openai/gpt-4o") + assert cost == 0.0 + + def test_extract_usage_with_tokens(self) -> None: + """_extract_usage() returns prompt, completion, total tokens from response.""" + resp = MagicMock() + resp.usage = MagicMock(prompt_tokens=50, completion_tokens=30, total_tokens=80) + usage = _extract_usage(resp) + assert usage == {"prompt_tokens": 50, "completion_tokens": 30, "total_tokens": 80} + + def test_extract_usage_no_usage_attr(self) -> None: + """_extract_usage() returns zeros when response has no usage attribute.""" + resp = MagicMock() + resp.usage = None + usage = _extract_usage(resp) + assert usage == {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} + + def test_extract_usage_missing_total(self) -> None: + """_extract_usage() computes total_tokens when not present.""" + resp = MagicMock() + resp.usage = MagicMock(prompt_tokens=20, completion_tokens=10, total_tokens=0) + # total_tokens is 0 (falsy) → should fall back to prompt + completion + usage = _extract_usage(resp) + assert usage["prompt_tokens"] == 20 + assert usage["completion_tokens"] == 10 + assert usage["total_tokens"] == 30 # 20 + 10 + + +# ────────────────────────────────────────────────────────────────────────── +# TestLLMClientCompat +# ────────────────────────────────────────────────────────────────────────── + + +class TestLLMClientCompat: + """Tests for LLMClient class, get_llm_client(), reset_llm_client().""" + + @pytest.mark.asyncio + async def test_mock_mode_calls_mock_generate(self) -> None: + """LLMClient in mock mode calls _mock_generate (no API call).""" + client = LLMClient(model=None, api_key=None) + assert client.is_mock is True + + with patch.object(client, "_mock_generate", new_callable=AsyncMock) as mock_mock_gen: + mock_mock_gen.return_value = LLMResponse( + message="mocked", proposed_actions=[], confidence=0.5 + ) + result = await client.generate("create a contact") + mock_mock_gen.assert_called_once() + assert result.message == "mocked" + + @pytest.mark.asyncio + async def test_mock_mode_keyword_matching(self) -> None: + """LLLMClient mock mode maps keywords to actions via action_mapper.""" + client = LLMClient(model=None, api_key=None) + result = await client.generate("create a new contact named John") + assert isinstance(result, LLMResponse) + assert len(result.proposed_actions) > 0 + assert result.proposed_actions[0]["method"] == "POST" + + @pytest.mark.asyncio + async def test_mock_mode_no_match(self) -> None: + """LLMClient mock mode returns empty actions for unrecognized query.""" + client = LLMClient(model=None, api_key=None) + result = await client.generate("xyzzy nonsense") + assert result.proposed_actions == [] + assert result.confidence < 0.5 + + @pytest.mark.asyncio + async def test_api_mode_calls_llm_complete(self) -> None: + """LLMClient in API mode calls llm_complete (mocked).""" + client = LLMClient(model="gpt-4o", api_key="test-key") + assert client.is_mock is False + + llm_result = { + "content": '{"message": "ok", "proposed_actions": [], "confidence": 0.9}', + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + "cost_usd": 0.001, + "model": "openai/gpt-4o", + "raw_response": MagicMock(), + } + with patch("app.ai.llm_client.llm_complete", new_callable=AsyncMock) as mock_llm_complete: + mock_llm_complete.return_value = llm_result + result = await client.generate("list all contacts") + mock_llm_complete.assert_called_once() + assert result.message == "ok" + assert result.confidence == 0.9 + + @pytest.mark.asyncio + async def test_api_mode_fallback_on_error(self) -> None: + """LLMClient API mode falls back to empty actions on API error.""" + client = LLMClient(model="gpt-4o", api_key="test-key") + with patch("app.ai.llm_client.llm_complete", new_callable=AsyncMock) as mock_llm_complete: + mock_llm_complete.side_effect = Exception("API down") + result = await client.generate("list contacts") + assert result.proposed_actions == [] + assert result.confidence == 0.1 + assert "API call failed" in result.message + + def test_get_llm_client_singleton(self) -> None: + """get_llm_client() returns the same instance on repeated calls.""" + reset_llm_client() + client1 = get_llm_client() + client2 = get_llm_client() + assert client1 is client2 + assert isinstance(client1, LLMClient) + + def test_reset_llm_client_clears_instance(self) -> None: + """reset_llm_client() clears the singleton, next get_llm_client() returns new instance.""" + reset_llm_client() + client1 = get_llm_client() + reset_llm_client() + client2 = get_llm_client() + assert client1 is not client2 + + def test_llm_client_default_mock_mode(self) -> None: + """LLMClient() with no args and no env vars defaults to mock mode.""" + with patch.dict("os.environ", {}, clear=False): + # Ensure AI_MODEL and AI_API_KEY are not set + import os + env_copy = dict(os.environ) + env_copy.pop("AI_MODEL", None) + env_copy.pop("AI_API_KEY", None) + with patch.dict(os.environ, env_copy, clear=True): + client = LLMClient() + assert client.is_mock is True + + def test_llm_client_api_mode_with_model_and_key(self) -> None: + """LLMClient() with model and api_key is not in mock mode.""" + client = LLMClient(model="gpt-4o", api_key="sk-test") + assert client.is_mock is False + + def test_llm_response_to_dict(self) -> None: + """LLMResponse.to_dict() returns correct structure.""" + resp = LLMResponse(message="hello", proposed_actions=[{"method": "GET"}], confidence=0.9) + d = resp.to_dict() + assert d == {"message": "hello", "proposed_actions": [{"method": "GET"}], "confidence": 0.9}