70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Abstract repository contract for LLM call metrics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class LlmMetricRecord:
|
|
"""In-memory shape of a metric row before persistence."""
|
|
|
|
id: str
|
|
created_at: datetime
|
|
user_id: str | None = None
|
|
thread_id: str | None = None
|
|
run_id: str | None = None
|
|
agent_name: str | None = None
|
|
model_name: str | None = None
|
|
duration_ms: int = 0
|
|
input_tokens: int | None = None
|
|
output_tokens: int | None = None
|
|
total_tokens: int | None = None
|
|
tokens_per_sec: float | None = None
|
|
status: str = "success"
|
|
error_type: str | None = None
|
|
error_message: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class LlmMetricsQuery:
|
|
"""Filter set for the admin list / export endpoints."""
|
|
|
|
user_id: str | None = None
|
|
# Multiple user ids — set when a fuzzy username search resolves to several
|
|
# accounts. Takes precedence over ``user_id`` when non-empty.
|
|
user_ids: list[str] | None = None
|
|
model_name: str | None = None
|
|
status: str | None = None
|
|
since: datetime | None = None
|
|
until: datetime | None = None
|
|
limit: int = 100
|
|
offset: int = 0
|
|
|
|
|
|
class LlmMetricsStore(ABC):
|
|
"""Persistence contract for the metrics middleware + admin API."""
|
|
|
|
@abstractmethod
|
|
async def record(self, metric: LlmMetricRecord) -> None:
|
|
"""Persist one metric row. Must never raise on best-effort writes."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def list(self, query: LlmMetricsQuery) -> list[dict[str, Any]]:
|
|
"""Return the matching rows (newest first), paginated by ``query``."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def count(self, query: LlmMetricsQuery) -> int:
|
|
"""Return the unpaginated row count for ``query``."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def iter_all(self, query: LlmMetricsQuery):
|
|
"""Yield rows matching ``query`` without pagination — used by export."""
|
|
raise NotImplementedError
|