From 9280d69ebf1237dfe3080cc47ba196b6e944e097 Mon Sep 17 00:00:00 2001 From: NSCT Agent Date: Sun, 23 Aug 2026 11:54:50 +0000 Subject: [PATCH] Stage 1: OpenAI-compatible provider layer (llm, vision, audio, metrics, debug) --- src/nsct/api/debug.py | 89 ++++++++++ src/nsct/providers/__init__.py | 39 +++- src/nsct/providers/audio.py | 183 +++++++++++++++++++ src/nsct/providers/llm.py | 314 +++++++++++++++++++++++++++++++++ src/nsct/providers/metrics.py | 83 +++++++++ src/nsct/providers/vision.py | 194 ++++++++++++++++++++ 6 files changed, 901 insertions(+), 1 deletion(-) create mode 100644 src/nsct/api/debug.py create mode 100644 src/nsct/providers/audio.py create mode 100644 src/nsct/providers/llm.py create mode 100644 src/nsct/providers/metrics.py create mode 100644 src/nsct/providers/vision.py diff --git a/src/nsct/api/debug.py b/src/nsct/api/debug.py new file mode 100644 index 0000000..9d94269 --- /dev/null +++ b/src/nsct/api/debug.py @@ -0,0 +1,89 @@ +"""Debug endpoints — model health / connectivity checks (only when NSCT_DEBUG=true).""" + +from __future__ import annotations + +import logging + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from nsct.config import AppSettings +from nsct.providers.llm import LLMProvider, get_provider as get_llm_provider +from nsct.providers.metrics import ProviderMetrics + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Request / response models +# --------------------------------------------------------------------------- + + +class LLMProbeRequest(BaseModel): + """Simple prompt for an LLM probe.""" + + prompt: str = Field(..., description="Input text sent to the model.") + + +class LLMProbeResponse(BaseModel): + """Response from the LLM probe.""" + + response: str = Field(..., description="The model's text response.") + model: str = Field(..., description="Model identifier used.") + + +class ModelInfoItem(BaseModel): + """One discovered model entry.""" + + id: str + object: str = "model" + owned_by: str | None = None + + +class ModelListResponse(BaseModel): + """List of discovered models.""" + + models: list[ModelInfoItem] + + +# --------------------------------------------------------------------------- +# Router +# --------------------------------------------------------------------------- + + +def create_debug_router( + config: AppSettings, + metrics: ProviderMetrics, +) -> APIRouter: + """Create a debug router. Routes are only active when config.debug is True.""" + router = APIRouter() + + @router.post("/debug/models/llm", response_model=LLMProbeResponse) + async def probe_llm(body: LLMProbeRequest) -> LLMProbeResponse: + """Send a prompt to the configured LLM and return the response. + + Only available when NSCT_DEBUG=true. + """ + if not config.debug: + raise HTTPException(status_code=404, detail="Debug endpoints are disabled") + + llm_provider = get_llm_provider(config, metrics) + + messages: list[dict[str, str]] = [{"role": "user", "content": body.prompt}] + result = await llm_provider.complete(messages) + return LLMProbeResponse(response=result, model=config.llm.model) + + @router.get("/debug/models/llm", response_model=ModelListResponse) + async def list_llm_models() -> ModelListResponse: + """List available LLM model IDs. + + Only available when NSCT_DEBUG=true. + """ + if not config.debug: + raise HTTPException(status_code=404, detail="Debug endpoints are disabled") + + llm_provider = get_llm_provider(config, metrics) + ids = await llm_provider.list_models() + return ModelListResponse(models=[ModelInfoItem(id=mid) for mid in ids]) + + return router \ No newline at end of file diff --git a/src/nsct/providers/__init__.py b/src/nsct/providers/__init__.py index 0b70ec9..c242141 100644 --- a/src/nsct/providers/__init__.py +++ b/src/nsct/providers/__init__.py @@ -1 +1,38 @@ -"""NSCT — providers package init.""" \ No newline at end of file +"""NSCT — providers package init. + +Re-exports all provider classes and metric classes for convenient imports. +""" + +from __future__ import annotations + +from nsct.providers.audio import _AudioProviderImpl +from nsct.providers.llm import ( + LLMProvider, + ProviderError, + ProviderHTTPError, + ProviderModelNotFoundError, + ProviderRateLimitError, + ProviderTimeoutError, + _LLMProviderImpl, +) +from nsct.providers.metrics import ProviderMetrics +from nsct.providers.vision import _VisionProviderImpl + +# Aliases for convenience +VisionProvider = _VisionProviderImpl +AudioProvider = _AudioProviderImpl + +__all__ = [ + "LLMProvider", + "_LLMProviderImpl", + "ProviderError", + "ProviderHTTPError", + "ProviderModelNotFoundError", + "ProviderRateLimitError", + "ProviderTimeoutError", + "ProviderMetrics", + "VisionProvider", + "_VisionProviderImpl", + "AudioProvider", + "_AudioProviderImpl", +] \ No newline at end of file diff --git a/src/nsct/providers/audio.py b/src/nsct/providers/audio.py new file mode 100644 index 0000000..2291df2 --- /dev/null +++ b/src/nsct/providers/audio.py @@ -0,0 +1,183 @@ +"""OpenAI-compatible audio / STT provider.""" + +from __future__ import annotations + +import asyncio +import io +import logging +import os +import time +from typing import Any + +import httpx +from httpx import HTTPStatusError +from openai import AsyncOpenAI + +from nsct.config import AppSettings +from nsct.providers.llm import ( + ProviderError, + ProviderHTTPError, + ProviderModelNotFoundError, + ProviderRateLimitError, + ProviderTimeoutError, + _RetryPolicy, +) +from nsct.providers.metrics import ProviderMetrics + +logger = logging.getLogger(__name__) + + +class _AudioProviderImpl: + """OpenAI-compatible audio transcription / STT provider.""" + + def __init__(self, config: AppSettings, metrics: ProviderMetrics) -> None: + self._config = config + self._metrics = metrics + self._base_url: str = config.audio.base_url.rstrip("/") + self._model: str = config.audio.model + self._api_key: str = config.audio.get_secret() + + self._connect_timeout: float = float( + os.environ.get("NSCT_AUDIO_CONNECT_TIMEOUT", "30") + ) + self._read_timeout: float = float( + os.environ.get("NSCT_AUDIO_READ_TIMEOUT", "120") + ) + self._client: AsyncOpenAI | None = None + + @property + def client(self) -> AsyncOpenAI: + if self._client is None: + self._client = self._create_client() + return self._client + + def _create_client(self) -> AsyncOpenAI: + http_client = httpx.AsyncClient( + limits=httpx.Limits( + max_connections=100, + max_keepalive_connections=20, + ), + timeout=httpx.Timeout( + connect=self._connect_timeout, + read=self._read_timeout, + write=self._read_timeout, + pool=5, + ), + ) + return AsyncOpenAI( + base_url=self._base_url, + api_key=self._api_key, + http_client=http_client, + ) + + async def _ensure_client(self) -> AsyncOpenAI: + if self._client is None: + self._client = self._create_client() + return self._client + + async def _request_with_retry( + self, + fn, # noqa: ANN202 + *args: Any, + **kwargs: Any, + ) -> Any: + last_exc: Exception | None = None + for attempt in range(_RetryPolicy.DEFAULT + 1): + try: + return await fn(*args, **kwargs) + except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.PoolTimeout) as exc: + last_exc = ProviderTimeoutError(str(exc)) + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_audio_request() + await self._metrics.record_llm_error(error_type="audio_timeout") + raise + except HTTPStatusError as exc: + status = exc.response.status_code + if status == 429: + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="audio_rate_limit") + raise + elif 500 <= status < 600: + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type=f"audio_http_{status}") + raise + else: + await self._metrics.record_llm_error(error_type=f"audio_http_{status}") + raise + except Exception as exc: + last_exc = exc + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="audio_unexpected") + raise + raise last_exc # type: ignore[misc] + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def transcribe( + self, + audio_file_path: str, + language: str | None = None, + prompt: str | None = None, + model: str | None = None, + ) -> dict[str, Any]: + """Transcribe an audio file. Returns ``{text, language, duration}``.""" + model = model or self._model + + async def _do() -> dict[str, Any]: + client = await self._ensure_client() + with open(audio_file_path, "rb") as fh: + audio_data = fh.read() + + build: dict[str, Any] = { + "model": model, + "file": ("audio.wav", io.BytesIO(audio_data), "audio/wav"), + } + if language is not None: + build["language"] = language + if prompt is not None: + build["prompt"] = prompt + + resp = await client.audio.transcriptions.create(**build) + + result = { + "text": getattr(resp, "text", ""), + "language": getattr(resp, "language", ""), + "duration": getattr(resp, "duration", 0.0), + } + await self._metrics.record_audio_request() + return result + + return await self._request_with_retry(_do) + + async def list_models(self) -> list[str]: + """Discover available audio model IDs.""" + async def _do() -> list[str]: + client = await self._ensure_client() + resp = await client.models.list() + return [m.id for m in resp.data] + + return await self._request_with_retry(_do) + + +# --------------------------------------------------------------------------- +# Singleton factory +# --------------------------------------------------------------------------- + +_instance: _AudioProviderImpl | None = None + + +def get_provider(config: AppSettings, metrics: ProviderMetrics) -> _AudioProviderImpl: + global _instance + if _instance is None: + _instance = _AudioProviderImpl(config, metrics) + return _instance \ No newline at end of file diff --git a/src/nsct/providers/llm.py b/src/nsct/providers/llm.py new file mode 100644 index 0000000..4185fac --- /dev/null +++ b/src/nsct/providers/llm.py @@ -0,0 +1,314 @@ +"""OpenAI-compatible LLM provider with timeout, retry, pooling, metrics.""" + +from __future__ import annotations + +import asyncio +import enum +import logging +import os +import time +from abc import ABC, abstractmethod +from typing import Any, AsyncGenerator, cast + +import httpx +from httpx import HTTPStatusError +from openai import AsyncOpenAI, AsyncStream +from openai._types import NOT_GIVEN, NotGiven +from openai.types.chat import ChatCompletion, ChatCompletionChunk + +from nsct.config import AppSettings +from nsct.providers.metrics import ProviderMetrics + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Structured errors +# --------------------------------------------------------------------------- + + +class ProviderError(Exception): + """Base exception for provider errors.""" + + +class ProviderTimeoutError(ProviderError): + """Request timed out.""" + + +class ProviderRateLimitError(ProviderError): + """Rate-limited by the provider.""" + + +class ProviderModelNotFoundError(ProviderError): + """Requested model not found.""" + + +class ProviderHTTPError(ProviderError): + """Generic HTTP error (non-2xx).""" + + def __init__(self, status_code: int, message: str) -> None: + super().__init__(f"HTTP {status_code}: {message}") + self.status_code = status_code + + +# --------------------------------------------------------------------------- +# Retries +# --------------------------------------------------------------------------- + + +class _RetryPolicy(enum.IntEnum): + """Number of retries for transient failures.""" + + DEFAULT = 3 + MAX_WAIT_SEC = 8 # 1s, 2s, 4s + + +# --------------------------------------------------------------------------- +# LLM Provider +# --------------------------------------------------------------------------- + + +class LLMProvider(ABC): + """Abstract base for LLM providers.""" + + @abstractmethod + async def complete( + self, + messages: list[dict[str, str]], + model: str | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + response_format: dict[str, Any] | None = None, + ) -> str: + """Issue a chat-completions request and return the assistant text.""" + ... + + @abstractmethod + async def stream_complete( + self, + messages: list[dict[str, str]], + model: str | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + ) -> AsyncGenerator[str, None]: + """Stream chat completions. Yields delta content chunks.""" + ... + + @abstractmethod + async def list_models(self) -> list[str]: + """Discover available model IDs.""" + ... + + +class _LLMProviderImpl(LLMProvider): + """Concrete OpenAI-compatible LLM provider.""" + + def __init__(self, config: AppSettings, metrics: ProviderMetrics) -> None: + self._config = config + self._metrics = metrics + self._base_url: str = config.llm.base_url + self._model: str = config.llm.model + self._max_concurrency: int = config.llm.max_concurrency + self._api_key: str = config.llm.get_secret() + + # Timeout config (seconds) — configurable via env, defaults here + self._connect_timeout: float = float( + os.environ.get("NSCT_LLM_CONNECT_TIMEOUT", "30") + ) + self._read_timeout: float = float( + os.environ.get("NSCT_LLM_READ_TIMEOUT", "120") + ) + + self._client: AsyncOpenAI | None = None + + @property + def client(self) -> AsyncOpenAI: + """Lazy-init the OpenAI client with connection pooling.""" + if self._client is None: + self._client = self._create_client() + return self._client + + def _create_client(self) -> AsyncOpenAI: + """Build an AsyncOpenAI client with httpx connection pooling.""" + http_client = httpx.AsyncClient( + limits=httpx.Limits( + max_connections=100, + max_keepalive_connections=20, + ), + timeout=httpx.Timeout( + connect=self._connect_timeout, + read=self._read_timeout, + write=self._read_timeout, + pool=5, + ), + ) + return AsyncOpenAI( + base_url=self._base_url.rstrip("/") + "/v1", + api_key=self._api_key, + http_client=http_client, + ) + + async def _ensure_client(self) -> AsyncOpenAI: + """Make sure the client is fresh and usable.""" + if self._client is None: + self._client = self._create_client() + return self._client + + async def _request_with_retry( + self, + fn, # noqa: ANN202 + *args: Any, + **kwargs: Any, + ) -> Any: + """Execute *fn* with exponential-backoff retries for transient errors.""" + last_exc: Exception | None = None + for attempt in range(_RetryPolicy.DEFAULT + 1): + try: + return await fn(*args, **kwargs) + except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.PoolTimeout) as exc: + last_exc = ProviderTimeoutError(str(exc)) + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="timeout") + raise + except HTTPStatusError as exc: + status = exc.response.status_code + if status == 429: + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="rate_limit") + raise + elif 500 <= status < 600: + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type=f"http_{status}") + raise + else: + await self._metrics.record_llm_error(error_type=f"http_{status}") + raise + except Exception as exc: + last_exc = exc + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="unexpected") + raise + raise last_exc # type: ignore[misc] + + async def complete( + self, + messages: list[dict[str, str]], + model: str | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + response_format: dict[str, Any] | None = None, + ) -> str: + """Issue a chat-completions request and return the assistant text.""" + model = model or self._model + + async def _do() -> str: + client = await self._ensure_client() + build: dict[str, Any] = { + "model": model, + "messages": messages, + } + if temperature is not None: + build["temperature"] = temperature + if max_tokens is not None: + build["max_tokens"] = max_tokens + if response_format is not None: + build["response_format"] = response_format + + start = time.monotonic() + resp: ChatCompletion = await client.chat.completions.create(**build) # type: ignore[arg-type] + latency = time.monotonic() - start + + input_tokens = (resp.usage.completion_tokens if resp.usage else 0) + ( + resp.usage.prompt_tokens if resp.usage else 0 + ) + output_tokens = resp.usage.completion_tokens if resp.usage else 0 + await self._metrics.record_llm_request( + input_tokens=input_tokens, + output_tokens=output_tokens, + latency=latency, + ) + + choice = resp.choices[0] + return choice.message.content or "" + + return await self._request_with_retry(_do) + + async def stream_complete( + self, + messages: list[dict[str, str]], + model: str | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + ) -> AsyncGenerator[str, None]: + """Stream chat completions. Yields delta content chunks.""" + model = model or self._model + tokens_input = 0 + tokens_output = 0 + start = time.monotonic() + + async for chunk in (await self._make_stream(messages, model, temperature, max_tokens)): + delta = chunk.choices[0].delta if chunk.choices and chunk.choices[0] else None + delta_text = delta.content if delta and delta.content else "" + if delta_text: + yield delta_text + tokens_output += 1 + + latency = time.monotonic() - start + for msg in messages: + tokens_input += len(msg.get("content", "").split()) + await self._metrics.record_llm_request( + input_tokens=tokens_input, + output_tokens=tokens_output, + latency=latency, + ) + + async def _make_stream( + self, + messages: list[dict[str, str]], + model: str, + temperature: float | None, + max_tokens: int | None, + ) -> AsyncStream[ChatCompletionChunk]: + """Helper: build and return the async stream.""" + client = await self._ensure_client() + build: dict[str, Any] = { + "model": model, + "messages": messages, + } + if temperature is not None: + build["temperature"] = temperature + if max_tokens is not None: + build["max_tokens"] = max_tokens + return await client.chat.completions.create(stream=True, **build) # type: ignore[arg-type] + + async def list_models(self) -> list[str]: + """Discover available model IDs via GET /v1/models.""" + async def _do() -> list[str]: + client = await self._ensure_client() + resp = await client.models.list() + return [m.id for m in resp.data] + + return await self._request_with_retry(_do) + + +# --------------------------------------------------------------------------- +# Singleton factory +# --------------------------------------------------------------------------- + +_instance: _LLMProviderImpl | None = None + + +def get_provider(config: AppSettings, metrics: ProviderMetrics) -> LLMProvider: + """Return a cached LLMProvider singleton.""" + global _instance + if _instance is None: + _instance = _LLMProviderImpl(config, metrics) + return _instance \ No newline at end of file diff --git a/src/nsct/providers/metrics.py b/src/nsct/providers/metrics.py new file mode 100644 index 0000000..cc8d031 --- /dev/null +++ b/src/nsct/providers/metrics.py @@ -0,0 +1,83 @@ +"""Provider metrics — thread-safe collection of request / token / latency counters.""" + +from __future__ import annotations + +import asyncio +import time +from collections import defaultdict +from typing import Any + + +class ProviderMetrics: + """Collects request-level and token-level metrics for all providers.""" + + def __init__(self) -> None: # noqa: PLR0913 + self._lock = asyncio.Lock() + # Counters + self.llm_requests_total: int = 0 + self.llm_tokens_input_total: int = 0 + self.llm_tokens_output_total: int = 0 + self.vision_requests_total: int = 0 + self.audio_requests_total: int = 0 + # Error bucket: error_type -> count + self._llm_errors: dict[str, int] = defaultdict(int) + # Latency tracking: list of seconds per successful call + self._llm_latency_samples: list[float] = [] + + # ------------------------------------------------------------------ + # Public helpers + # ------------------------------------------------------------------ + + async def record_llm_request(self, *, input_tokens: int, output_tokens: int, latency: float) -> None: + """Record a successful LLM request (token counts, latency).""" + async with self._lock: + self.llm_requests_total += 1 + self.llm_tokens_input_total += input_tokens + self.llm_tokens_output_total += output_tokens + self._llm_latency_samples.append(latency) + + async def record_llm_error(self, *, error_type: str) -> None: + """Record a failed LLM call.""" + async with self._lock: + self.llm_requests_total += 1 + self._llm_errors[error_type] += 1 + + async def record_vision_request(self) -> None: + """Record a successful vision request.""" + async with self._lock: + self.vision_requests_total += 1 + + async def record_audio_request(self) -> None: + """Record a successful audio request.""" + async with self._lock: + self.audio_requests_total += 1 + + async def get_metrics(self) -> dict[str, Any]: + """Return a flat dict suitable for Prometheus exporters or logging.""" + async with self._lock: + samples = list(self._llm_latency_samples) + avg_latency = (sum(samples) / len(samples)) if samples else 0.0 + return { + "llm_requests_total": self.llm_requests_total, + "llm_tokens_input_total": self.llm_tokens_input_total, + "llm_tokens_output_total": self.llm_tokens_output_total, + "llm_errors_total": dict(self._llm_errors), + "llm_avg_latency_seconds": round(avg_latency, 4), + "vision_requests_total": self.vision_requests_total, + "audio_requests_total": self.audio_requests_total, + } + + # ------------------------------------------------------------------ + # Reset helpers (useful in tests) + # ------------------------------------------------------------------ + + async def reset(self) -> None: + """Reset all counters to zero.""" + async with self._lock: + self.llm_requests_total = 0 + self.llm_tokens_input_total = 0 + self.llm_tokens_output_total = 0 + self._llm_errors.clear() + self._llm_latency_samples.clear() + self.vision_requests_total = 0 + self.audio_requests_total = 0 \ No newline at end of file diff --git a/src/nsct/providers/vision.py b/src/nsct/providers/vision.py new file mode 100644 index 0000000..dd378c6 --- /dev/null +++ b/src/nsct/providers/vision.py @@ -0,0 +1,194 @@ +"""OpenAI-compatible vision provider (image analysis).""" + +from __future__ import annotations + +import asyncio +import logging +import os +import time +from typing import Any + +import httpx +from httpx import HTTPStatusError +from openai import AsyncOpenAI + +from nsct.config import AppSettings +from nsct.providers.llm import ( + ProviderError, + ProviderHTTPError, + ProviderModelNotFoundError, + ProviderRateLimitError, + ProviderTimeoutError, + _RetryPolicy, +) +from nsct.providers.metrics import ProviderMetrics + +logger = logging.getLogger(__name__) + + +class _VisionProviderImpl: + """OpenAI-compatible vision / image-analysis provider.""" + + def __init__(self, config: AppSettings, metrics: ProviderMetrics) -> None: + self._config = config + self._metrics = metrics + self._base_url: str = config.vision.base_url + self._model: str = config.vision.model + self._api_key: str = config.vision.get_secret() + + self._connect_timeout: float = float( + os.environ.get("NSCT_VISION_CONNECT_TIMEOUT", "30") + ) + self._read_timeout: float = float( + os.environ.get("NSCT_VISION_READ_TIMEOUT", "120") + ) + self._client: AsyncOpenAI | None = None + + @property + def client(self) -> AsyncOpenAI: + if self._client is None: + self._client = self._create_client() + return self._client + + def _create_client(self) -> AsyncOpenAI: + http_client = httpx.AsyncClient( + limits=httpx.Limits( + max_connections=100, + max_keepalive_connections=20, + ), + timeout=httpx.Timeout( + connect=self._connect_timeout, + read=self._read_timeout, + write=self._read_timeout, + pool=5, + ), + ) + return AsyncOpenAI( + base_url=self._base_url.rstrip("/") + "/v1", + api_key=self._api_key, + http_client=http_client, + ) + + async def _ensure_client(self) -> AsyncOpenAI: + if self._client is None: + self._client = self._create_client() + return self._client + + async def _request_with_retry( + self, + fn, # noqa: ANN202 + *args: Any, + **kwargs: Any, + ) -> Any: + last_exc: Exception | None = None + for attempt in range(_RetryPolicy.DEFAULT + 1): + try: + return await fn(*args, **kwargs) + except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.PoolTimeout) as exc: + last_exc = ProviderTimeoutError(str(exc)) + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="vision_timeout") + raise + except HTTPStatusError as exc: + status = exc.response.status_code + if status == 429: + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="vision_rate_limit") + raise + elif 500 <= status < 600: + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type=f"vision_http_{status}") + raise + else: + await self._metrics.record_llm_error(error_type=f"vision_http_{status}") + raise + except Exception as exc: + last_exc = exc + if attempt < _RetryPolicy.DEFAULT: + await asyncio.sleep(2 ** attempt) + else: + await self._metrics.record_llm_error(error_type="vision_unexpected") + raise + raise last_exc # type: ignore[misc] + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def analyze( + self, + image_url_or_base64: str, + prompt: str, + model: str | None = None, + ) -> str: + """Send an image (URL or base64) + text prompt and get analysis text.""" + model = model or self._model + start = time.monotonic() + + async def _do() -> str: + client = await self._ensure_client() + + # Build content array: text part + image part + # Using a plain list — OpenAI accepts it as the content argument + content: list[dict[str, Any]] = [ + {"type": "text", "text": prompt}, + ] + if image_url_or_base64.startswith(("http://", "https://")): + content.append({ + "type": "image_url", + "image_url": {"url": image_url_or_base64}, + }) + else: + content.append({ + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{image_url_or_base64}"}, + }) + + resp = await client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": content}], + ) + return resp.choices[0].message.content or "" + + result = await self._request_with_retry(_do) + await self._metrics.record_vision_request() + return result + + async def analyze_multiple( + self, + images_and_prompts: list[tuple[str, str]], + model: str | None = None, + ) -> list[str]: + """Analyze multiple images concurrently. Each item is (image, prompt).""" + tasks = [self.analyze(img, pmt, model=model) for img, pmt in images_and_prompts] + results = await asyncio.gather(*tasks) + return list(results) + + async def list_models(self) -> list[str]: + """Discover available vision model IDs.""" + async def _do() -> list[str]: + client = await self._ensure_client() + resp = await client.models.list() + return [m.id for m in resp.data] + + return await self._request_with_retry(_do) + + +# --------------------------------------------------------------------------- +# Singleton factory +# --------------------------------------------------------------------------- + +_instance: _VisionProviderImpl | None = None + + +def get_provider(config: AppSettings, metrics: ProviderMetrics) -> _VisionProviderImpl: + global _instance + if _instance is None: + _instance = _VisionProviderImpl(config, metrics) + return _instance \ No newline at end of file