Stage 1: OpenAI-compatible provider layer (llm, vision, audio, metrics, debug)
This commit is contained in:
NSCT Agent
89
src/nsct/api/debug.py
Normal file
89
src/nsct/api/debug.py
Normal file
@@ -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
|
||||
@@ -1 +1,38 @@
|
||||
"""NSCT — providers package init."""
|
||||
"""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",
|
||||
]
|
||||
183
src/nsct/providers/audio.py
Normal file
183
src/nsct/providers/audio.py
Normal file
@@ -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
|
||||
314
src/nsct/providers/llm.py
Normal file
314
src/nsct/providers/llm.py
Normal file
@@ -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
|
||||
83
src/nsct/providers/metrics.py
Normal file
83
src/nsct/providers/metrics.py
Normal file
@@ -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
|
||||
194
src/nsct/providers/vision.py
Normal file
194
src/nsct/providers/vision.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user