Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion backend/app/api/docs/analytics/monthly.md
Original file line number Diff line number Diff line change
Expand Up @@ -163,5 +163,5 @@ GET /api/analytics/monthly?metric=eval_cost&modality=STT&from_month=2026-01-01
only to the `requests` metric (via the chain count), never to `cost` —
its dollars come from the underlying `llm_call` rows.
- **Cost computed on summed tokens per (provider, model) group**, which is
equivalent to per-row pricing because `estimate_model_cost` is linear in
equivalent to per-row pricing because the pricing function is linear in
token counts.
57 changes: 48 additions & 9 deletions backend/app/crud/model_config.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import logging
from collections.abc import Mapping
from datetime import datetime
from typing import Any, Literal, get_args

Expand Down Expand Up @@ -373,23 +374,40 @@ def is_summary_model(session: Session, provider: Provider, model_name: str) -> b
return "summary" in model.config


def estimate_model_cost(
def list_active_model_pricing(
session: Session,
provider: Provider,
pairs: set[tuple[str, str]],
) -> dict[tuple[str, str], Mapping[str, Any] | None]:
"""Fetch pricing for many (provider, model_name) pairs in one query."""
if not pairs:
return {}
providers = {provider for provider, _ in pairs}
model_names = {model_name for _, model_name in pairs}
stmt = select(
ModelConfig.provider, ModelConfig.model_name, ModelConfig.pricing
).where(
ModelConfig.provider.in_(providers), # type: ignore[attr-defined]
ModelConfig.model_name.in_(model_names), # type: ignore[attr-defined]
ModelConfig.is_active,
)
return {
(provider, model_name): pricing
for provider, model_name, pricing in session.exec(stmt).all()
}


def compute_cost_from_pricing(
pricing: Mapping[str, Any] | None,
provider: str,
model_name: str,
input_tokens: int,
output_tokens: int,
usage_type: Literal["response", "batch"] = "response",
) -> dict[str, Any] | None:
model = get_model_config(session=session, provider=provider, model_name=model_name)
if model is None or model.pricing is None:
return None

if not isinstance(model.pricing, dict):
if not isinstance(pricing, dict):
return None

pricing_source: dict[str, Any] = model.pricing
usage_pricing = pricing_source.get(usage_type)
usage_pricing = pricing.get(usage_type)
if not isinstance(usage_pricing, dict):
return None

Expand All @@ -415,3 +433,24 @@ def estimate_model_cost(
"total_cost": input_cost + output_cost,
"currency": "USD",
}


def estimate_model_cost(
session: Session,
provider: Provider,
model_name: str,
input_tokens: int,
output_tokens: int,
usage_type: Literal["response", "batch"] = "response",
) -> dict[str, Any] | None:
model = get_model_config(session=session, provider=provider, model_name=model_name)
if model is None:
return None
return compute_cost_from_pricing(
pricing=model.pricing,
provider=provider,
model_name=model_name,
input_tokens=input_tokens,
output_tokens=output_tokens,
usage_type=usage_type,
)
47 changes: 30 additions & 17 deletions backend/app/services/analytics/aggregation.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@
from app.crud.model_config import (
KNOWN_PROVIDERS,
NATIVE_PROVIDER_SUFFIX,
estimate_model_cost,
compute_cost_from_pricing,
list_active_model_pricing,
)
from app.models.analytics import Modality
from app.models.config.version import ConfigVersion
Expand Down Expand Up @@ -103,11 +104,12 @@ def _aggregate_llm_calls(
) -> None:
"""GROUP BY llm_call → merge counts, tokens, and per-group cost into buckets.

Cost is computed per (provider, model) group via `estimate_model_cost`
using the summed tokens. The pricing function is linear in tokens, so
summing first and pricing once is equivalent to per-row pricing. Skipped
for providers outside `KNOWN_PROVIDERS` (the `model_config.provider`
enum) since the lookup would raise InvalidTextRepresentation.
Cost is computed per (provider, model) group from the summed tokens. The
pricing function is linear in tokens, so summing first and pricing once
is equivalent to per-row pricing. Pricing for every group is fetched in
one query up front — a per-group lookup was an N+1. Providers outside
`KNOWN_PROVIDERS` (the `model_config.provider` enum) are excluded from
the lookup since Postgres would raise InvalidTextRepresentation.
"""
month_col = (
sa.func.date_trunc("month", LlmCall.inserted_at).cast(sa.Date).label("month")
Expand Down Expand Up @@ -149,7 +151,17 @@ def _aggregate_llm_calls(
if provider_filter is not None:
stmt = stmt.where(LlmCall.provider == provider_filter)

for row in session.exec(stmt).all():
rows = session.exec(stmt).all()
pricing_by_model = list_active_model_pricing(
session,
{
(row.provider, row.model)
for row in rows
if row.provider in KNOWN_PROVIDERS and row.model is not None
},
)

for row in rows:
modality_enum = Modality(row.modality)
if modality_filter is not None and modality_enum is not modality_filter:
continue
Expand All @@ -160,16 +172,17 @@ def _aggregate_llm_calls(
output_tokens = int(row.output_tokens or 0)
bucket["input_tokens"] += input_tokens
bucket["output_tokens"] += output_tokens
if (input_tokens or output_tokens) and row.provider in KNOWN_PROVIDERS:
estimate = estimate_model_cost(
session=session,
provider=row.provider, # type: ignore[arg-type]
model_name=row.model,
input_tokens=input_tokens,
output_tokens=output_tokens,
)
if estimate is not None:
bucket["cost_usd"] += Decimal(str(estimate.get("total_cost", 0)))
if not (input_tokens or output_tokens):
continue
estimate = compute_cost_from_pricing(
pricing=pricing_by_model.get((row.provider, row.model)),
provider=row.provider,
model_name=row.model,
input_tokens=input_tokens,
output_tokens=output_tokens,
)
if estimate is not None:
bucket["cost_usd"] += Decimal(str(estimate.get("total_cost", 0)))


def _aggregate_chains(
Expand Down
59 changes: 59 additions & 0 deletions backend/app/tests/api/routes/test_analytics.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from uuid import uuid4

import pytest
from sqlalchemy import event
from fastapi.testclient import TestClient
from sqlmodel import Session

Expand Down Expand Up @@ -211,6 +212,64 @@ def test_metric_cost_uses_estimate_model_cost(
total = sum(float(p["value"]) for p in data)
assert total == pytest.approx(2.0)

def test_metric_cost_fetches_pricing_in_single_query(
self,
client: TestClient,
db: Session,
user_api_key: TestAuthContext,
user_api_key_header: dict[str, str],
llm_job: Job,
):
model_names = [f"gpt-analytics-n1-{uuid4().hex[:8]}" for _ in range(3)]
for name in model_names:
db.add(
ModelConfig(
provider="openai",
model_name=name,
completion_type=["text"],
pricing={
"response": {"input_token_cost": 1.0, "output_token_cost": 2.0}
},
is_active=True,
)
)
db.commit()
for name in model_names:
_make_llm_call(
db,
job_id=llm_job.id,
project_id=user_api_key.project_id,
organization_id=user_api_key.organization_id,
provider="openai",
model=name,
input_tokens=1_000_000,
output_tokens=500_000,
)

model_config_queries: list[str] = []

def _capture(conn, cursor, statement, parameters, context, executemany):
if "model_config" in statement:
model_config_queries.append(statement)

engine = db.get_bind()
event.listen(engine, "before_cursor_execute", _capture)
try:
data = _ok(
client.get(
MONTHLY_URL,
params={"metric": "cost", "provider": "openai"},
headers=user_api_key_header,
)
)
finally:
event.remove(engine, "before_cursor_execute", _capture)

total = sum(float(p["value"]) for p in data)
assert total == pytest.approx(6.0)
# One pricing lookup for llm_call + one DISTINCT ON subquery for evals.
assert len(model_config_queries) <= 2

@pytest.mark.parametrize(
"input_type,output_type,expected_modality",
[
Expand Down
2 changes: 1 addition & 1 deletion docs/wiki/modules/platform.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ All paths relative to `backend/app/`.
| Feature flags | `api/routes/features.py` | `feature_flag` (`models/feature_flag.py`) | `core/feature_flags/`, `crud/feature_flag.py` |
| Languages | `api/routes/languages.py` | `global.languages` (`models/language.py`) | `crud/language.py` |
| Credentials | `api/routes/credentials.py` | `credential` (`models/credentials.py`) | `crud/credentials.py`; provider keys per org/project; envelope encryption (KMS-wrapped data key + AES-GCM), prefix-versioned ciphertexts |
| Model config | `api/routes/model_config.py` | `model_config` (`models/model_config.py`) | `crud/model_config.py` |
| Model config | `api/routes/model_config.py` | `model_config` (`models/model_config.py`) | `crud/model_config.py`; cost: `estimate_model_cost` (single lookup) vs `list_active_model_pricing` + `compute_cost_from_pricing` (one batched lookup — use in loops/aggregations, e.g. `services/analytics/aggregation.py`) |
| Bucket providers | — | reuses `credential` (`google-gcp`) | `services/buckets/` — global registry + resolver (`providers/`), GCS V4 signed + bulk-signed URLs with a 24h expiry cap (`providers/gcs.py`, cap in `providers/base.py`), attachment path selection + URL resolution (`attachments.py`) |
| Cron | `api/routes/cron.py` | — | triggers batch polling (`crud/evaluations/cron.py`, `crud/assessment/cron.py` — the latter polls `method == RUN` assessments only; API-client BATCH runs are driven by their own Celery self-re-enqueue) |
| Jobs | — | `job` (`models/job.py`), `batch_job` (`models/batch_job.py`) | `crud/jobs.py`, `crud/job/`, `services/job_monitoring.py` |
Expand Down
Loading