mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-05 07:37:50 +00:00
Merge remote-tracking branch 'origin/main' into litellm_fix_concurrent_single_agent_publish
This commit is contained in:
commit
e3f2d5c52e
24 changed files with 2040 additions and 20 deletions
1
.github/workflows/test-unit.yml
vendored
1
.github/workflows/test-unit.yml
vendored
|
|
@ -137,6 +137,7 @@ jobs:
|
|||
include:
|
||||
- shard: core-utils
|
||||
test-path: |-
|
||||
tests/unit/accounting
|
||||
tests/unit/batch_completion
|
||||
tests/unit/decisions
|
||||
tests/unit/extraction
|
||||
|
|
|
|||
|
|
@ -4,7 +4,9 @@ needs an approving review from the spend-log owner before it can merge.
|
|||
CODEOWNERS can only match file paths, and every schema change edits the same schema.prisma copies,
|
||||
so a path rule would put every migration on the owner. This gate reads the diff instead: when no
|
||||
added line in the watched files mentions the table it passes at once, otherwise it passes only once
|
||||
the owner's latest review on the pull request is an approval. The workflow re-runs on every review.
|
||||
the owner's latest review on the pull request is an approval that still covers every spend-log line
|
||||
in the head (approvals survive pushes, so a later commit may not add spend-log lines the owner never
|
||||
saw). The workflow re-runs on every review.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -13,7 +15,7 @@ import argparse
|
|||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from collections.abc import Sequence
|
||||
from collections.abc import Callable, Sequence
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
|
|
@ -37,6 +39,7 @@ class Review(BaseModel):
|
|||
model_config = ConfigDict(frozen=True, extra="ignore")
|
||||
state: str
|
||||
user: Reviewer
|
||||
commit_id: str | None = None
|
||||
|
||||
|
||||
_REVIEWS: Final = TypeAdapter(list[Review])
|
||||
|
|
@ -75,6 +78,24 @@ def approved_by(reviews: Sequence[Review], login: str) -> bool:
|
|||
return bool(states) and states[-1] == "APPROVED"
|
||||
|
||||
|
||||
def latest_approval(reviews: Sequence[Review], login: str) -> Review | None:
|
||||
"""`login`'s most recent APPROVED review on the pull request, the one GitHub keeps in force after pushes."""
|
||||
own: Final = tuple(review for review in reviews if review.user.login == login and review.state == "APPROVED")
|
||||
return own[-1] if own else None
|
||||
|
||||
|
||||
def approval_covers(
|
||||
approval: Review, head: str, hits: Sequence[str], hits_at: Callable[[str], Sequence[str] | None]
|
||||
) -> bool:
|
||||
"""True when the approval is on the head, or the head adds no spend-log line the approved commit did not."""
|
||||
if approval.commit_id is None:
|
||||
return False
|
||||
if approval.commit_id == head:
|
||||
return True
|
||||
reviewed: Final = hits_at(approval.commit_id)
|
||||
return reviewed is not None and set(hits) <= set(reviewed)
|
||||
|
||||
|
||||
def added_diff(base: str, head: str) -> str:
|
||||
result: Final = subprocess.run(
|
||||
("git", "diff", "--unified=0", "--no-color", f"{base}..{head}", "--", *WATCHED, MIGRATIONS_PREFIX),
|
||||
|
|
@ -85,6 +106,15 @@ def added_diff(base: str, head: str) -> str:
|
|||
return result.stdout
|
||||
|
||||
|
||||
def fetched(commit: str) -> bool:
|
||||
result: Final = subprocess.run(("git", "fetch", "--no-tags", "--depth=1", "origin", commit), check=False)
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
def reviewed_hits(base: str, commit: str) -> tuple[str, ...] | None:
|
||||
return spend_log_hits(added_diff(base, commit)) if fetched(commit) else None
|
||||
|
||||
|
||||
class Arguments(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
base: str
|
||||
|
|
@ -104,14 +134,21 @@ def main(argv: Sequence[str]) -> int:
|
|||
if not hits:
|
||||
sys.stdout.write("No added line in schema.prisma or a migration names LiteLLM_SpendLogs; nothing to review.\n")
|
||||
return 0
|
||||
if approved_by(_REVIEWS.validate_json(args.reviews.read_bytes()), args.approver):
|
||||
reviews: Final = _REVIEWS.validate_json(args.reviews.read_bytes())
|
||||
approval: Final = latest_approval(reviews, args.approver) if approved_by(reviews, args.approver) else None
|
||||
if approval is not None and approval_covers(
|
||||
approval, args.head, hits, lambda commit: reviewed_hits(args.base, commit)
|
||||
):
|
||||
sys.stdout.write(f"LiteLLM_SpendLogs change approved by {args.approver}: {', '.join(hits)}\n")
|
||||
return 0
|
||||
listed: Final = "".join(f" {hit}\n" for hit in hits)
|
||||
sys.stdout.write(
|
||||
reason: Final = (
|
||||
f"This pull request changes LiteLLM_SpendLogs and needs an approving review from @{args.approver}:\n"
|
||||
f"{listed}The check re-runs when a review is submitted.\n"
|
||||
if approval is None
|
||||
else f"LiteLLM_SpendLogs lines changed after @{args.approver} approved {approval.commit_id}, "
|
||||
f"a new approving review is needed:\n"
|
||||
)
|
||||
sys.stdout.write(f"{reason}{listed}The check re-runs when a review is submitted.\n")
|
||||
return 1
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ from litellm.integrations.custom_guardrail import (
|
|||
CustomGuardrail,
|
||||
log_guardrail_information,
|
||||
)
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import effective_skip_tool_message_for_guardrail
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
from litellm.proxy.guardrails._content_utils import iter_tool_json_fields, map_tool_json_text, walk_user_text
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
|
|
@ -753,7 +754,12 @@ class _ENTERPRISE_SecretDetection(CustomGuardrail):
|
|||
_masked_entity_count.set({})
|
||||
|
||||
# Covers multimodal list content + Responses-API input.
|
||||
inspected = walk_user_text(data, self.redact_text, self.redact_tool_payload) + self._redact_prompt(data)
|
||||
inspected = walk_user_text(
|
||||
data,
|
||||
self.redact_text,
|
||||
self.redact_tool_payload,
|
||||
skip_tool_message=effective_skip_tool_message_for_guardrail(self),
|
||||
) + self._redact_prompt(data)
|
||||
|
||||
if inspected == 0:
|
||||
# Image-only, empty-text, and unsupported payloads inspected
|
||||
|
|
|
|||
|
|
@ -0,0 +1,47 @@
|
|||
SET LOCAL lock_timeout = '10s';
|
||||
|
||||
CREATE TYPE "AccountingEventKind" AS ENUM ('inbound_usage', 'outbound_usage', 'cost');
|
||||
|
||||
CREATE TYPE "AccountingDeliveryState" AS ENUM ('pending', 'applied', 'parked');
|
||||
|
||||
CREATE TYPE "AccountingDeliveryTarget" AS ENUM ('LiteLLM_VerificationToken', 'LiteLLM_UserTable', 'LiteLLM_TeamTable', 'LiteLLM_TeamMembership', 'LiteLLM_OrganizationTable', 'LiteLLM_OrganizationMembership', 'LiteLLM_EndUserTable', 'LiteLLM_AgentsTable', 'LiteLLM_TagTable', 'LiteLLM_ProjectTable', 'LiteLLM_ModelAccessGroupBudgetTable', 'LiteLLM_BudgetTable', 'LiteLLM_BudgetWindowSpend', 'LiteLLM_DailyUserSpend', 'LiteLLM_DailyTeamSpend', 'LiteLLM_DailyOrganizationSpend', 'LiteLLM_DailyEndUserSpend', 'LiteLLM_DailyAgentSpend', 'LiteLLM_DailyTagSpend', 'LiteLLM_DailyModelUsage', 'LiteLLM_DailyToolSpend', 'LiteLLM_AutoRouterDailySpend');
|
||||
|
||||
-- CreateTable
|
||||
CREATE TABLE IF NOT EXISTS "LiteLLM_AccountingEvent" (
|
||||
"event_id" TEXT NOT NULL,
|
||||
"kind" "AccountingEventKind" NOT NULL,
|
||||
"source_event_id" TEXT,
|
||||
"schema_version" INTEGER NOT NULL,
|
||||
"payload_hash" TEXT NOT NULL,
|
||||
"occurred_at" TIMESTAMP(3) NOT NULL,
|
||||
"recorded_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"payload" JSONB NOT NULL,
|
||||
|
||||
CONSTRAINT "LiteLLM_AccountingEvent_schema_version_check" CHECK ("schema_version" > 0),
|
||||
CONSTRAINT "LiteLLM_AccountingEvent_pkey" PRIMARY KEY ("event_id")
|
||||
);
|
||||
|
||||
-- CreateTable
|
||||
CREATE TABLE IF NOT EXISTS "LiteLLM_AccountingDelivery" (
|
||||
"event_id" TEXT NOT NULL,
|
||||
"target" "AccountingDeliveryTarget" NOT NULL,
|
||||
"version" INTEGER NOT NULL,
|
||||
"state" "AccountingDeliveryState" NOT NULL DEFAULT 'pending',
|
||||
"attempts" INTEGER NOT NULL DEFAULT 0,
|
||||
"next_attempt_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"applied_at" TIMESTAMP(3),
|
||||
"last_error" TEXT,
|
||||
|
||||
CONSTRAINT "LiteLLM_AccountingDelivery_version_check" CHECK ("version" > 0),
|
||||
CONSTRAINT "LiteLLM_AccountingDelivery_attempts_check" CHECK ("attempts" >= 0),
|
||||
CONSTRAINT "LiteLLM_AccountingDelivery_pkey" PRIMARY KEY ("event_id","target","version")
|
||||
);
|
||||
|
||||
-- CreateIndex
|
||||
CREATE INDEX IF NOT EXISTS "LiteLLM_AccountingEvent_recorded_at_idx" ON "LiteLLM_AccountingEvent"("recorded_at");
|
||||
|
||||
-- CreateIndex
|
||||
CREATE INDEX IF NOT EXISTS "LiteLLM_AccountingEvent_kind_source_event_id_idx" ON "LiteLLM_AccountingEvent"("kind", "source_event_id");
|
||||
|
||||
-- CreateIndex
|
||||
CREATE INDEX IF NOT EXISTS "LiteLLM_AccountingDelivery_target_state_next_attempt_at_idx" ON "LiteLLM_AccountingDelivery"("target", "state", "next_attempt_at");
|
||||
|
|
@ -2061,3 +2061,73 @@ model LiteLLM_TelemetryReport {
|
|||
@@index([window_end])
|
||||
@@index([created_at])
|
||||
}
|
||||
|
||||
enum AccountingEventKind {
|
||||
inbound_usage
|
||||
outbound_usage
|
||||
cost
|
||||
}
|
||||
|
||||
enum AccountingDeliveryState {
|
||||
pending
|
||||
applied
|
||||
parked
|
||||
}
|
||||
|
||||
enum AccountingDeliveryTarget {
|
||||
LiteLLM_VerificationToken
|
||||
LiteLLM_UserTable
|
||||
LiteLLM_TeamTable
|
||||
LiteLLM_TeamMembership
|
||||
LiteLLM_OrganizationTable
|
||||
LiteLLM_OrganizationMembership
|
||||
LiteLLM_EndUserTable
|
||||
LiteLLM_AgentsTable
|
||||
LiteLLM_TagTable
|
||||
LiteLLM_ProjectTable
|
||||
LiteLLM_ModelAccessGroupBudgetTable
|
||||
LiteLLM_BudgetTable
|
||||
LiteLLM_BudgetWindowSpend
|
||||
LiteLLM_DailyUserSpend
|
||||
LiteLLM_DailyTeamSpend
|
||||
LiteLLM_DailyOrganizationSpend
|
||||
LiteLLM_DailyEndUserSpend
|
||||
LiteLLM_DailyAgentSpend
|
||||
LiteLLM_DailyTagSpend
|
||||
LiteLLM_DailyModelUsage
|
||||
LiteLLM_DailyToolSpend
|
||||
LiteLLM_AutoRouterDailySpend
|
||||
}
|
||||
|
||||
// Durable accounting ledger. One row per usage or cost event; payload holds the
|
||||
// private event envelope versioned by schema_version. Usage rows retain request-scoped
|
||||
// pricing context; cost rows include resolved rates and the finalized pricing snapshot.
|
||||
model LiteLLM_AccountingEvent {
|
||||
event_id String @id
|
||||
kind AccountingEventKind
|
||||
source_event_id String?
|
||||
schema_version Int
|
||||
payload_hash String
|
||||
occurred_at DateTime
|
||||
recorded_at DateTime @default(now())
|
||||
payload Json
|
||||
|
||||
@@index([recorded_at])
|
||||
@@index([kind, source_event_id])
|
||||
}
|
||||
|
||||
// One row per (event, target table, version): pending work for the projector and,
|
||||
// once applied in the same transaction as the target write, the receipt that stops a retry.
|
||||
model LiteLLM_AccountingDelivery {
|
||||
event_id String
|
||||
target AccountingDeliveryTarget
|
||||
version Int
|
||||
state AccountingDeliveryState @default(pending)
|
||||
attempts Int @default(0)
|
||||
next_attempt_at DateTime @default(now())
|
||||
applied_at DateTime?
|
||||
last_error String?
|
||||
|
||||
@@id([event_id, target, version])
|
||||
@@index([target, state, next_attempt_at])
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ const FINISH_REASONS: &[(&str, &str)] = &[
|
|||
("max_tokens", "length"),
|
||||
("refusal", "content_filter"),
|
||||
("compaction", "length"),
|
||||
("model_context_window_exceeded", "length"),
|
||||
("guardrail_intervened", "content_filter"),
|
||||
("content_filtered", "content_filter"),
|
||||
("content_filter", "content_filter"),
|
||||
|
|
@ -74,6 +75,7 @@ mod tests {
|
|||
#[case::end_turn("end_turn", "stop")]
|
||||
#[case::stop_sequence("stop_sequence", "stop")]
|
||||
#[case::max_tokens("max_tokens", "length")]
|
||||
#[case::model_context_window_exceeded("model_context_window_exceeded", "length")]
|
||||
#[case::refusal("refusal", "content_filter")]
|
||||
#[case::guardrail_intervened("guardrail_intervened", "content_filter")]
|
||||
#[case::content_filtered("content_filtered", "content_filter")]
|
||||
|
|
|
|||
62
litellm/accounting/__init__.py
Normal file
62
litellm/accounting/__init__.py
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
"""Public accounting contracts. Importing these values does not enable accounting."""
|
||||
|
||||
from litellm.accounting.measurements import measurements_for
|
||||
from litellm.accounting.protocols import BudgetEnforcer, CostAggregator, CostCompiler, UsageMeter
|
||||
from litellm.accounting.types import (
|
||||
AdmissionDecision,
|
||||
AggregateQuery,
|
||||
AggregateRow,
|
||||
CostEvent,
|
||||
DeliveryState,
|
||||
DeliveryTarget,
|
||||
Dimension,
|
||||
DimensionFilter,
|
||||
DimensionName,
|
||||
EventKind,
|
||||
Granularity,
|
||||
InboundUsageEvent,
|
||||
Measurement,
|
||||
Metric,
|
||||
MetricName,
|
||||
Multiplier,
|
||||
NamedCost,
|
||||
OutboundUsageEvent,
|
||||
Outcome,
|
||||
PricedQuantity,
|
||||
PricingSnapshot,
|
||||
Rate,
|
||||
Unpriced,
|
||||
validate_aggregate_rows,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"AdmissionDecision",
|
||||
"AggregateQuery",
|
||||
"AggregateRow",
|
||||
"BudgetEnforcer",
|
||||
"CostAggregator",
|
||||
"CostCompiler",
|
||||
"CostEvent",
|
||||
"DeliveryState",
|
||||
"DeliveryTarget",
|
||||
"Dimension",
|
||||
"DimensionFilter",
|
||||
"DimensionName",
|
||||
"EventKind",
|
||||
"Granularity",
|
||||
"InboundUsageEvent",
|
||||
"Measurement",
|
||||
"Metric",
|
||||
"MetricName",
|
||||
"Multiplier",
|
||||
"NamedCost",
|
||||
"OutboundUsageEvent",
|
||||
"Outcome",
|
||||
"PricedQuantity",
|
||||
"PricingSnapshot",
|
||||
"Rate",
|
||||
"Unpriced",
|
||||
"UsageMeter",
|
||||
"measurements_for",
|
||||
"validate_aggregate_rows",
|
||||
]
|
||||
13
litellm/accounting/measurements.py
Normal file
13
litellm/accounting/measurements.py
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
from typing import Final
|
||||
|
||||
from litellm.accounting.types import CostEvent, InboundUsageEvent, Measurement, OutboundUsageEvent
|
||||
|
||||
|
||||
def measurements_for(event: InboundUsageEvent | OutboundUsageEvent | CostEvent) -> tuple[Measurement, ...]:
|
||||
"""Return the complete contribution set for one event, ready for one apply call."""
|
||||
metrics: Final = event.charges if isinstance(event, CostEvent) else event.metrics
|
||||
occurred_at: Final = event.occurred_at if isinstance(event, CostEvent) else event.started_at
|
||||
return tuple(
|
||||
Measurement(source_event_id=event.event_id, occurred_at=occurred_at, dimensions=event.dimensions, metric=metric)
|
||||
for metric in metrics or ()
|
||||
)
|
||||
85
litellm/accounting/protocols.py
Normal file
85
litellm/accounting/protocols.py
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
"""Contracts only. Implementations own persistence, pricing, projections and live budgets.
|
||||
|
||||
The event row's ``payload`` is a private envelope versioned by the row's ``schema_version``.
|
||||
Usage envelopes retain the public event and request-scoped pricing context: quantities with units,
|
||||
model/provider, requested and billed tier/region when known, and per-request/deployment overrides.
|
||||
These inputs must survive request teardown. Cost-map rates are resolved at compilation, not capture.
|
||||
Cost envelopes retain the public CostEvent, including its required finalized pricing snapshot.
|
||||
T5/T6 own envelope serialization; this public API does not introduce a second persisted schema.
|
||||
"""
|
||||
|
||||
from typing import Protocol
|
||||
|
||||
from litellm.accounting.types import (
|
||||
AdmissionDecision,
|
||||
AggregateQuery,
|
||||
AggregateRow,
|
||||
CostEvent,
|
||||
Dimension,
|
||||
InboundUsageEvent,
|
||||
Measurement,
|
||||
Metric,
|
||||
OutboundUsageEvent,
|
||||
Unpriced,
|
||||
)
|
||||
|
||||
|
||||
class UsageMeter(Protocol):
|
||||
async def record(self, event: InboundUsageEvent | OutboundUsageEvent) -> None:
|
||||
"""Acknowledge commit, no-op identical IDs and reject conflicting content.
|
||||
|
||||
Persist the event and its request-scoped pricing context before acknowledging.
|
||||
HTTP acknowledgement is a separate policy. Unknown usage remains None.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class CostCompiler(Protocol):
|
||||
async def compile(self, event: OutboundUsageEvent) -> CostEvent | Unpriced:
|
||||
"""Reuse the committed charge or finalize one charge and its delivery intent.
|
||||
|
||||
Resolve captured request-scoped context by event identity and cost-map rates
|
||||
at compilation. Persist the resolved inputs in the required snapshot. Replay
|
||||
reuses the committed result and never consumes free allowance again.
|
||||
Missing pricing is Unpriced, never an inferred zero.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class CostAggregator(Protocol):
|
||||
async def apply(self, measurements: tuple[Measurement, ...]) -> None:
|
||||
"""Apply one source event's complete contribution set per target/version.
|
||||
|
||||
Each target commits all contributions and its receipt atomically. A receipt
|
||||
for the first metric must not suppress the rest. Usage need not wait for cost.
|
||||
"""
|
||||
...
|
||||
|
||||
async def query(self, query: AggregateQuery) -> tuple[AggregateRow, ...]:
|
||||
"""Return disjoint [start, end) rows per group using existing daily projections.
|
||||
|
||||
Filters combine with AND, values within a filter with IN. DAY emits daily
|
||||
buckets; None emits whole-interval totals. Reject unsupported metric/scope
|
||||
combinations. Different groups may overlap, including non-additive tags.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class BudgetEnforcer(Protocol):
|
||||
async def reserve(
|
||||
self, reservation_id: str, dimensions: tuple[Dimension, ...], estimated: tuple[Metric, ...]
|
||||
) -> AdmissionDecision:
|
||||
"""Apply existing admission policies and estimates; duplicate IDs add no hold."""
|
||||
...
|
||||
|
||||
async def settle(self, reservation_id: str | None, cost_event: CostEvent) -> None:
|
||||
"""Settle the committed charge with atomic deduplication and hold mutation.
|
||||
|
||||
Preserve scope/window attribution and retry unfinished settlement after
|
||||
restart. SQL projection receipts do not provide live-counter atomicity.
|
||||
"""
|
||||
...
|
||||
|
||||
async def release(self, reservation_id: str) -> None:
|
||||
"""Idempotently release unused exposure at the request boundary, preserving charges."""
|
||||
...
|
||||
428
litellm/accounting/types.py
Normal file
428
litellm/accounting/types.py
Normal file
|
|
@ -0,0 +1,428 @@
|
|||
"""Accounting values use UTC instants, decimal strings on the wire and USD charges."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime, timezone
|
||||
from decimal import Decimal
|
||||
from enum import Enum
|
||||
from typing import Annotated, Final, Literal, TypeAlias
|
||||
|
||||
from pydantic import AfterValidator, AwareDatetime, BaseModel, ConfigDict, Field, model_validator
|
||||
from typing_extensions import Self
|
||||
|
||||
from litellm.types.utils import CallTypes
|
||||
|
||||
_CONFIG: Final = ConfigDict(frozen=True, extra="forbid", validate_default=True)
|
||||
Identifier: TypeAlias = Annotated[str, Field(strict=True, min_length=1, pattern=r"\S")]
|
||||
Quantity: TypeAlias = Annotated[Decimal, Field(strict=True, ge=0, allow_inf_nan=False)]
|
||||
|
||||
|
||||
def _utc(value: datetime) -> datetime:
|
||||
return value.astimezone(timezone.utc)
|
||||
|
||||
|
||||
Instant: TypeAlias = Annotated[AwareDatetime, AfterValidator(_utc)]
|
||||
|
||||
|
||||
class MetricName(str, Enum):
|
||||
"""Reasoning tokens are included in completion tokens; above-1hr writes are included in cache creation tokens."""
|
||||
|
||||
REQUESTS = "requests"
|
||||
SUCCESSFUL_REQUESTS = "successful_requests"
|
||||
FAILED_REQUESTS = "failed_requests"
|
||||
PROMPT_TOKENS = "prompt_tokens"
|
||||
COMPLETION_TOKENS = "completion_tokens"
|
||||
REASONING_TOKENS = "reasoning_tokens"
|
||||
TOTAL_TOKENS = "total_tokens"
|
||||
CACHE_READ_INPUT_TOKENS = "cache_read_input_tokens"
|
||||
CACHE_CREATION_INPUT_TOKENS = "cache_creation_input_tokens"
|
||||
CACHE_CREATION_INPUT_TOKENS_ABOVE_1HR = "cache_creation_input_tokens_above_1hr"
|
||||
SPEND = "spend"
|
||||
|
||||
|
||||
class DimensionName(str, Enum):
|
||||
KEY = "key"
|
||||
USER = "user"
|
||||
TEAM = "team"
|
||||
ORGANIZATION = "organization"
|
||||
END_USER = "end_user"
|
||||
MODEL = "model"
|
||||
MODEL_GROUP = "model_group"
|
||||
DEPLOYMENT = "deployment"
|
||||
PROVIDER = "provider"
|
||||
TOOL = "tool"
|
||||
AGENT = "agent"
|
||||
PROJECT = "project"
|
||||
MODEL_ACCESS_GROUP = "model_access_group"
|
||||
API_BASE = "api_base"
|
||||
SERVICE_TIER = "service_tier"
|
||||
TAG = "tag"
|
||||
|
||||
|
||||
class Outcome(str, Enum):
|
||||
SUCCEEDED = "succeeded"
|
||||
FAILED = "failed"
|
||||
CANCELLED = "cancelled"
|
||||
|
||||
|
||||
class Granularity(str, Enum):
|
||||
DAY = "day"
|
||||
|
||||
|
||||
class EventKind(str, Enum):
|
||||
INBOUND_USAGE = "inbound_usage"
|
||||
OUTBOUND_USAGE = "outbound_usage"
|
||||
COST = "cost"
|
||||
|
||||
|
||||
class DeliveryState(str, Enum):
|
||||
PENDING = "pending"
|
||||
APPLIED = "applied"
|
||||
PARKED = "parked"
|
||||
|
||||
|
||||
class DeliveryTarget(str, Enum):
|
||||
"""One member per existing table the projector writes; each is applied and receipted independently."""
|
||||
|
||||
KEY_SPEND = "LiteLLM_VerificationToken"
|
||||
USER_SPEND = "LiteLLM_UserTable"
|
||||
TEAM_SPEND = "LiteLLM_TeamTable"
|
||||
TEAM_MEMBER_SPEND = "LiteLLM_TeamMembership"
|
||||
ORGANIZATION_SPEND = "LiteLLM_OrganizationTable"
|
||||
ORGANIZATION_MEMBER_SPEND = "LiteLLM_OrganizationMembership"
|
||||
END_USER_SPEND = "LiteLLM_EndUserTable"
|
||||
AGENT_SPEND = "LiteLLM_AgentsTable"
|
||||
TAG_SPEND = "LiteLLM_TagTable"
|
||||
PROJECT_SPEND = "LiteLLM_ProjectTable"
|
||||
MODEL_ACCESS_GROUP_SPEND = "LiteLLM_ModelAccessGroupBudgetTable"
|
||||
BUDGET_SPEND = "LiteLLM_BudgetTable"
|
||||
BUDGET_WINDOW_SPEND = "LiteLLM_BudgetWindowSpend"
|
||||
DAILY_USER_SPEND = "LiteLLM_DailyUserSpend"
|
||||
DAILY_TEAM_SPEND = "LiteLLM_DailyTeamSpend"
|
||||
DAILY_ORGANIZATION_SPEND = "LiteLLM_DailyOrganizationSpend"
|
||||
DAILY_END_USER_SPEND = "LiteLLM_DailyEndUserSpend"
|
||||
DAILY_AGENT_SPEND = "LiteLLM_DailyAgentSpend"
|
||||
DAILY_TAG_SPEND = "LiteLLM_DailyTagSpend"
|
||||
DAILY_MODEL_USAGE = "LiteLLM_DailyModelUsage"
|
||||
DAILY_TOOL_SPEND = "LiteLLM_DailyToolSpend"
|
||||
AUTOROUTER_DAILY_SPEND = "LiteLLM_AutoRouterDailySpend"
|
||||
|
||||
|
||||
_SUBSET_PAIRS: Final = (
|
||||
(MetricName.REASONING_TOKENS, MetricName.COMPLETION_TOKENS),
|
||||
(MetricName.CACHE_CREATION_INPUT_TOKENS_ABOVE_1HR, MetricName.CACHE_CREATION_INPUT_TOKENS),
|
||||
)
|
||||
|
||||
|
||||
def _check_subsets(values: Mapping[str, Decimal]) -> None:
|
||||
for part, whole in _SUBSET_PAIRS:
|
||||
if part.value in values and whole.value in values and values[part.value] > values[whole.value]:
|
||||
raise ValueError(f"{part.value} cannot exceed {whole.value}")
|
||||
|
||||
|
||||
class Metric(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
name: MetricName
|
||||
value: Quantity
|
||||
unit: Literal["count", "USD"] = "count"
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_unit(self) -> Self:
|
||||
if (self.name is MetricName.SPEND) != (self.unit == "USD"):
|
||||
raise ValueError("SPEND requires USD; usage metrics require count")
|
||||
if self.unit == "count" and self.value != self.value.to_integral_value():
|
||||
raise ValueError("Count metrics require integral values")
|
||||
return self
|
||||
|
||||
|
||||
class Dimension(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
name: DimensionName
|
||||
value: Identifier
|
||||
|
||||
|
||||
_REPEATABLE_DIMENSIONS: Final = frozenset({DimensionName.TAG, DimensionName.MODEL_ACCESS_GROUP})
|
||||
|
||||
|
||||
def _dimensions(values: tuple[Dimension, ...]) -> tuple[Dimension, ...]:
|
||||
if len(frozenset(values)) != len(values):
|
||||
raise ValueError("Duplicate dimension values")
|
||||
single_names: Final = tuple(value.name for value in values if value.name not in _REPEATABLE_DIMENSIONS)
|
||||
if len(frozenset(single_names)) != len(single_names):
|
||||
raise ValueError("Only tag and model access group dimensions may have multiple values")
|
||||
return values
|
||||
|
||||
|
||||
Dimensions: TypeAlias = Annotated[tuple[Dimension, ...], AfterValidator(_dimensions)]
|
||||
_CALLER_METRICS: Final = frozenset((MetricName.REQUESTS, MetricName.SUCCESSFUL_REQUESTS, MetricName.FAILED_REQUESTS))
|
||||
|
||||
|
||||
def _metrics(values: tuple[Metric, ...]) -> tuple[Metric, ...]:
|
||||
if len(frozenset(value.name for value in values)) != len(values):
|
||||
raise ValueError("Duplicate metric names")
|
||||
return values
|
||||
|
||||
|
||||
Metrics: TypeAlias = Annotated[tuple[Metric, ...], AfterValidator(_metrics)]
|
||||
|
||||
|
||||
def _interval(start: datetime, end: datetime, *, allow_empty: bool = False) -> None:
|
||||
if end < start or (end == start and not allow_empty):
|
||||
raise ValueError("End must follow start")
|
||||
|
||||
|
||||
class InboundUsageEvent(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
event_id: Identifier
|
||||
request_id: Identifier
|
||||
call_type: CallTypes
|
||||
started_at: Instant
|
||||
ended_at: Instant
|
||||
outcome: Outcome
|
||||
dimensions: Dimensions
|
||||
metrics: Metrics
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_caller_activity(self) -> Self:
|
||||
_interval(self.started_at, self.ended_at, allow_empty=True)
|
||||
if any(metric.name not in _CALLER_METRICS for metric in self.metrics):
|
||||
raise ValueError("Inbound events own only caller request metrics")
|
||||
if not any(metric.name is MetricName.REQUESTS and metric.value == 1 for metric in self.metrics):
|
||||
raise ValueError("Inbound events count exactly one caller request")
|
||||
for metric in self.metrics:
|
||||
if metric.name is MetricName.SUCCESSFUL_REQUESTS and metric.value != int(self.outcome is Outcome.SUCCEEDED):
|
||||
raise ValueError("Successful request count must match outcome")
|
||||
if metric.name is MetricName.FAILED_REQUESTS and metric.value != int(self.outcome is not Outcome.SUCCEEDED):
|
||||
raise ValueError("Failed request count must match outcome")
|
||||
return self
|
||||
|
||||
|
||||
class OutboundUsageEvent(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
event_id: Identifier
|
||||
attempt_id: Identifier
|
||||
request_id: Identifier | None
|
||||
retry_of_attempt_id: Identifier | None
|
||||
call_type: CallTypes
|
||||
started_at: Instant
|
||||
ended_at: Instant
|
||||
outcome: Outcome
|
||||
dimensions: Dimensions
|
||||
metrics: Metrics | None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_attempt_usage(self) -> Self:
|
||||
_interval(self.started_at, self.ended_at, allow_empty=True)
|
||||
if self.retry_of_attempt_id == self.attempt_id:
|
||||
raise ValueError("An attempt cannot retry itself")
|
||||
if self.metrics is not None and any(
|
||||
metric.name in _CALLER_METRICS or metric.name is MetricName.SPEND for metric in self.metrics
|
||||
):
|
||||
raise ValueError("Outbound events own only attempt usage metrics")
|
||||
if self.metrics is not None:
|
||||
_check_subsets({metric.name.value: metric.value for metric in self.metrics})
|
||||
return self
|
||||
|
||||
|
||||
PricingUnit: TypeAlias = Literal["token", "second", "character", "image", "request"]
|
||||
|
||||
|
||||
class Rate(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
measure: Identifier
|
||||
unit: PricingUnit
|
||||
usd_per_unit: Quantity
|
||||
|
||||
|
||||
class PricedQuantity(BaseModel):
|
||||
"""A pricing input in its billing unit. Only seconds may be fractional."""
|
||||
|
||||
model_config = _CONFIG
|
||||
|
||||
measure: Identifier
|
||||
unit: PricingUnit
|
||||
value: Quantity
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_discrete_quantity(self) -> Self:
|
||||
if self.unit != "second" and self.value != self.value.to_integral_value():
|
||||
raise ValueError("Discrete priced quantities require integral values")
|
||||
return self
|
||||
|
||||
|
||||
class Multiplier(BaseModel):
|
||||
"""A regional or tier factor applied on top of the priced key, e.g. data_residency=eu or vertex_location=us-east5."""
|
||||
|
||||
model_config = _CONFIG
|
||||
|
||||
name: Identifier
|
||||
value: Identifier
|
||||
factor: Annotated[Decimal, Field(strict=True, gt=0, allow_inf_nan=False)]
|
||||
|
||||
|
||||
class NamedCost(BaseModel):
|
||||
"""A free-form extra charge by name, e.g. azure_model_router_flat_cost."""
|
||||
|
||||
model_config = _CONFIG
|
||||
|
||||
name: Identifier
|
||||
amount: Quantity
|
||||
|
||||
|
||||
class PricingSnapshot(BaseModel):
|
||||
"""Retained inputs and result of authoritative compilation, reproducible after cost-map rates change.
|
||||
|
||||
Request-scoped inputs and overrides come from the usage row's versioned payload. Cost-map rates are
|
||||
resolved at compilation, not frozen at capture. Retries reuse the committed charge and snapshot.
|
||||
Component names match ``CostBreakdown``: reasoning_cost is part of output_cost, not an additional charge.
|
||||
Component sums and float tolerances belong to reconciliation, not this structural contract.
|
||||
"""
|
||||
|
||||
model_config = _CONFIG
|
||||
|
||||
schema_version: Literal[1] = 1
|
||||
calculator_version: Identifier
|
||||
model: Identifier
|
||||
provider: Identifier | None
|
||||
pricing_source: Literal["cost_map", "deployment", "per_request", "upstream_header"]
|
||||
priced_model_key: Identifier
|
||||
multipliers: tuple[Multiplier, ...] = ()
|
||||
rates: tuple[Rate, ...] = ()
|
||||
usage: tuple[PricedQuantity, ...] = ()
|
||||
input_cost: Quantity | None = None
|
||||
output_cost: Quantity | None = None
|
||||
cache_read_cost: Quantity | None = None
|
||||
cache_creation_cost: Quantity | None = None
|
||||
reasoning_cost: Quantity | None = None
|
||||
tool_usage_cost: Quantity | None = None
|
||||
guardrail_cost: Quantity | None = None
|
||||
additional_costs: tuple[NamedCost, ...] = ()
|
||||
original_cost: Quantity | None = None
|
||||
discount_percent: Quantity | None = None
|
||||
discount_amount: Quantity | None = None
|
||||
margin_percent: Quantity | None = None
|
||||
margin_fixed_amount: Quantity | None = None
|
||||
margin_total_amount: Quantity | None = None
|
||||
upstream_reported_cost: Quantity | None = None
|
||||
total_cost: Quantity
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_snapshot(self) -> Self:
|
||||
rate_keys: Final = frozenset((rate.measure, rate.unit) for rate in self.rates)
|
||||
quantity_keys: Final = frozenset((quantity.measure, quantity.unit) for quantity in self.usage)
|
||||
if len(rate_keys) != len(self.rates):
|
||||
raise ValueError("Duplicate rate measures and units")
|
||||
if len(quantity_keys) != len(self.usage):
|
||||
raise ValueError("Duplicate priced quantity measures and units")
|
||||
if not rate_keys.issubset(quantity_keys):
|
||||
raise ValueError("Every rate requires one priced quantity with the same measure and unit")
|
||||
if len(frozenset(cost.name for cost in self.additional_costs)) != len(self.additional_costs):
|
||||
raise ValueError("Duplicate additional cost names")
|
||||
if len(frozenset(multiplier.name for multiplier in self.multipliers)) != len(self.multipliers):
|
||||
raise ValueError("Duplicate multiplier names")
|
||||
_check_subsets({quantity.measure: quantity.value for quantity in self.usage if quantity.unit == "token"})
|
||||
return self
|
||||
|
||||
|
||||
class CostEvent(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
event_id: Identifier
|
||||
usage_event_id: Identifier
|
||||
occurred_at: Instant
|
||||
dimensions: Dimensions
|
||||
charges: Metrics
|
||||
snapshot: PricingSnapshot
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_charge(self) -> Self:
|
||||
if len(self.charges) != 1 or self.charges[0].name is not MetricName.SPEND:
|
||||
raise ValueError("A cost event requires one authoritative USD spend metric")
|
||||
if self.event_id == self.usage_event_id:
|
||||
raise ValueError("Cost and usage events require distinct identities")
|
||||
if self.snapshot.total_cost != self.charges[0].value:
|
||||
raise ValueError("Snapshot total must equal the authoritative spend")
|
||||
return self
|
||||
|
||||
|
||||
class Unpriced(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
usage_event_id: Identifier
|
||||
reason: Identifier
|
||||
|
||||
|
||||
class Measurement(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
source_event_id: Identifier
|
||||
occurred_at: Instant
|
||||
dimensions: Dimensions
|
||||
metric: Metric
|
||||
|
||||
|
||||
class DimensionFilter(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
name: DimensionName
|
||||
values: Annotated[tuple[Identifier, ...], Field(min_length=1)]
|
||||
|
||||
|
||||
class AggregateQuery(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
metric: MetricName
|
||||
filters: tuple[DimensionFilter, ...]
|
||||
group_by: tuple[DimensionName, ...]
|
||||
start: Instant
|
||||
end: Instant
|
||||
granularity: Granularity | None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_query(self) -> Self:
|
||||
_interval(self.start, self.end)
|
||||
if len(frozenset(item.name for item in self.filters)) != len(self.filters):
|
||||
raise ValueError("Use one IN filter per dimension; filters combine with AND")
|
||||
if len(frozenset(self.group_by)) != len(self.group_by):
|
||||
raise ValueError("Duplicate group-by dimensions")
|
||||
if any(
|
||||
(instant.hour, instant.minute, instant.second, instant.microsecond) != (0, 0, 0, 0)
|
||||
for instant in (self.start, self.end)
|
||||
):
|
||||
raise ValueError("Initial reporting queries require UTC day boundaries")
|
||||
return self
|
||||
|
||||
|
||||
class AggregateRow(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
start: Instant
|
||||
end: Instant
|
||||
dimensions: Dimensions
|
||||
value: Quantity
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_bounds(self) -> Self:
|
||||
_interval(self.start, self.end)
|
||||
return self
|
||||
|
||||
|
||||
def validate_aggregate_rows(rows: tuple[AggregateRow, ...]) -> tuple[AggregateRow, ...]:
|
||||
"""Validate [start, end) buckets per group; overlapping tag groups are not additive."""
|
||||
groups: Final = frozenset(frozenset(row.dimensions) for row in rows)
|
||||
for group in groups:
|
||||
ordered = sorted((row for row in rows if frozenset(row.dimensions) == group), key=lambda row: row.start)
|
||||
if any(left.end > right.start for left, right in zip(ordered, ordered[1:])):
|
||||
raise ValueError("Aggregate buckets overlap within a dimension group")
|
||||
return rows
|
||||
|
||||
|
||||
class AdmissionDecision(BaseModel):
|
||||
model_config = _CONFIG
|
||||
|
||||
allowed: bool
|
||||
reason: Identifier | None = None
|
||||
|
|
@ -338,6 +338,8 @@ _PRISMA_MODELS: Final[frozenset[str]] = frozenset(
|
|||
"LiteLLM_ToolTable",
|
||||
"LiteLLM_TelemetryReport",
|
||||
"LiteLLM_AccessGroupTable",
|
||||
"LiteLLM_AccountingDelivery",
|
||||
"LiteLLM_AccountingEvent",
|
||||
"LiteLLM_ClaudeCodePluginTable",
|
||||
"LiteLLM_MemoryTable",
|
||||
"LiteLLM_AdaptiveRouterState",
|
||||
|
|
|
|||
|
|
@ -266,6 +266,7 @@ _FINISH_REASON_MAP: Final[dict[str, OpenAIChatCompletionFinishReason]] = {
|
|||
"tool_use": "tool_calls",
|
||||
"refusal": "content_filter",
|
||||
"compaction": "length",
|
||||
"model_context_window_exceeded": "length",
|
||||
# Cohere
|
||||
"COMPLETE": "stop",
|
||||
"ERROR_TOXIC": "content_filter",
|
||||
|
|
|
|||
|
|
@ -305,6 +305,11 @@ def iter_message_text(data: Mapping[str, object]) -> Iterator[str]:
|
|||
yield from _iter_text_parts_in_content(message.get("content"))
|
||||
|
||||
|
||||
def _is_tool_message(message: object) -> bool:
|
||||
fields: Final = _text_object(message)
|
||||
return fields is not None and str(fields.get("role") or "").lower() == "tool"
|
||||
|
||||
|
||||
def _call_item(item: object) -> tuple[str, object] | None:
|
||||
"""Return ``(field, value)`` for a Responses tool-call input item, else None."""
|
||||
fields: Final = _text_object(item)
|
||||
|
|
@ -316,11 +321,14 @@ def walk_user_text(
|
|||
data: dict[str, object],
|
||||
visit: Callable[[str], str],
|
||||
visit_tool: Callable[[JsonValue], JsonValue] | None = None,
|
||||
*,
|
||||
skip_tool_message: bool = False,
|
||||
) -> int:
|
||||
"""Rewrite every text fragment in place via ``visit``.
|
||||
|
||||
Mutates ``data["messages"]`` and ``data["input"]`` (including tool payloads), ``data["system"]`` and
|
||||
``data["instructions"]``. Returns the number
|
||||
``data["instructions"]``. ``skip_tool_message`` leaves ``role: tool`` messages and ``tool_result``
|
||||
blocks as sent. Returns the number
|
||||
of fragments visited so callers can short-circuit when nothing was
|
||||
inspected.
|
||||
"""
|
||||
|
|
@ -353,6 +361,8 @@ def walk_user_text(
|
|||
|
||||
def _rewrite_part(part: Mapping[str, object], native: bool) -> Mapping[str, object]:
|
||||
nonlocal visited
|
||||
if skip_tool_message and part.get("type") == "tool_result":
|
||||
return part
|
||||
text: Final = _part_text(part)
|
||||
rewritten: Final = {**part, **({"text": visit(text)} if text is not None else {})}
|
||||
if text is not None:
|
||||
|
|
@ -384,18 +394,17 @@ def walk_user_text(
|
|||
return content
|
||||
|
||||
messages: Final = data.get("messages")
|
||||
if isinstance(messages, list):
|
||||
for message in messages:
|
||||
if not isinstance(message, dict):
|
||||
continue
|
||||
if "content" in message:
|
||||
message["content"] = _rewrite_content(message["content"])
|
||||
calls = message.get("tool_calls")
|
||||
if isinstance(calls, list):
|
||||
for call in calls:
|
||||
function = call.get("function") if isinstance(call, dict) else None
|
||||
if isinstance(function, dict) and "arguments" in function:
|
||||
function["arguments"] = _rewrite_arguments(function["arguments"])
|
||||
for message in _object_list(messages) or ():
|
||||
if (skip_tool_message and _is_tool_message(message)) or not isinstance(message, dict):
|
||||
continue
|
||||
if "content" in message:
|
||||
message["content"] = _rewrite_content(message["content"])
|
||||
calls = message.get("tool_calls")
|
||||
if isinstance(calls, list):
|
||||
for call in calls:
|
||||
function = call.get("function") if isinstance(call, dict) else None
|
||||
if isinstance(function, dict) and "arguments" in function:
|
||||
function["arguments"] = _rewrite_arguments(function["arguments"])
|
||||
|
||||
if "system" in data:
|
||||
data["system"] = _rewrite_content(data["system"]) # rebind-ok: preserve the native hook's in-place contract
|
||||
|
|
|
|||
|
|
@ -2061,3 +2061,73 @@ model LiteLLM_TelemetryReport {
|
|||
@@index([window_end])
|
||||
@@index([created_at])
|
||||
}
|
||||
|
||||
enum AccountingEventKind {
|
||||
inbound_usage
|
||||
outbound_usage
|
||||
cost
|
||||
}
|
||||
|
||||
enum AccountingDeliveryState {
|
||||
pending
|
||||
applied
|
||||
parked
|
||||
}
|
||||
|
||||
enum AccountingDeliveryTarget {
|
||||
LiteLLM_VerificationToken
|
||||
LiteLLM_UserTable
|
||||
LiteLLM_TeamTable
|
||||
LiteLLM_TeamMembership
|
||||
LiteLLM_OrganizationTable
|
||||
LiteLLM_OrganizationMembership
|
||||
LiteLLM_EndUserTable
|
||||
LiteLLM_AgentsTable
|
||||
LiteLLM_TagTable
|
||||
LiteLLM_ProjectTable
|
||||
LiteLLM_ModelAccessGroupBudgetTable
|
||||
LiteLLM_BudgetTable
|
||||
LiteLLM_BudgetWindowSpend
|
||||
LiteLLM_DailyUserSpend
|
||||
LiteLLM_DailyTeamSpend
|
||||
LiteLLM_DailyOrganizationSpend
|
||||
LiteLLM_DailyEndUserSpend
|
||||
LiteLLM_DailyAgentSpend
|
||||
LiteLLM_DailyTagSpend
|
||||
LiteLLM_DailyModelUsage
|
||||
LiteLLM_DailyToolSpend
|
||||
LiteLLM_AutoRouterDailySpend
|
||||
}
|
||||
|
||||
// Durable accounting ledger. One row per usage or cost event; payload holds the
|
||||
// private event envelope versioned by schema_version. Usage rows retain request-scoped
|
||||
// pricing context; cost rows include resolved rates and the finalized pricing snapshot.
|
||||
model LiteLLM_AccountingEvent {
|
||||
event_id String @id
|
||||
kind AccountingEventKind
|
||||
source_event_id String?
|
||||
schema_version Int
|
||||
payload_hash String
|
||||
occurred_at DateTime
|
||||
recorded_at DateTime @default(now())
|
||||
payload Json
|
||||
|
||||
@@index([recorded_at])
|
||||
@@index([kind, source_event_id])
|
||||
}
|
||||
|
||||
// One row per (event, target table, version): pending work for the projector and,
|
||||
// once applied in the same transaction as the target write, the receipt that stops a retry.
|
||||
model LiteLLM_AccountingDelivery {
|
||||
event_id String
|
||||
target AccountingDeliveryTarget
|
||||
version Int
|
||||
state AccountingDeliveryState @default(pending)
|
||||
attempts Int @default(0)
|
||||
next_attempt_at DateTime @default(now())
|
||||
applied_at DateTime?
|
||||
last_error String?
|
||||
|
||||
@@id([event_id, target, version])
|
||||
@@index([target, state, next_attempt_at])
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2061,3 +2061,73 @@ model LiteLLM_TelemetryReport {
|
|||
@@index([window_end])
|
||||
@@index([created_at])
|
||||
}
|
||||
|
||||
enum AccountingEventKind {
|
||||
inbound_usage
|
||||
outbound_usage
|
||||
cost
|
||||
}
|
||||
|
||||
enum AccountingDeliveryState {
|
||||
pending
|
||||
applied
|
||||
parked
|
||||
}
|
||||
|
||||
enum AccountingDeliveryTarget {
|
||||
LiteLLM_VerificationToken
|
||||
LiteLLM_UserTable
|
||||
LiteLLM_TeamTable
|
||||
LiteLLM_TeamMembership
|
||||
LiteLLM_OrganizationTable
|
||||
LiteLLM_OrganizationMembership
|
||||
LiteLLM_EndUserTable
|
||||
LiteLLM_AgentsTable
|
||||
LiteLLM_TagTable
|
||||
LiteLLM_ProjectTable
|
||||
LiteLLM_ModelAccessGroupBudgetTable
|
||||
LiteLLM_BudgetTable
|
||||
LiteLLM_BudgetWindowSpend
|
||||
LiteLLM_DailyUserSpend
|
||||
LiteLLM_DailyTeamSpend
|
||||
LiteLLM_DailyOrganizationSpend
|
||||
LiteLLM_DailyEndUserSpend
|
||||
LiteLLM_DailyAgentSpend
|
||||
LiteLLM_DailyTagSpend
|
||||
LiteLLM_DailyModelUsage
|
||||
LiteLLM_DailyToolSpend
|
||||
LiteLLM_AutoRouterDailySpend
|
||||
}
|
||||
|
||||
// Durable accounting ledger. One row per usage or cost event; payload holds the
|
||||
// private event envelope versioned by schema_version. Usage rows retain request-scoped
|
||||
// pricing context; cost rows include resolved rates and the finalized pricing snapshot.
|
||||
model LiteLLM_AccountingEvent {
|
||||
event_id String @id
|
||||
kind AccountingEventKind
|
||||
source_event_id String?
|
||||
schema_version Int
|
||||
payload_hash String
|
||||
occurred_at DateTime
|
||||
recorded_at DateTime @default(now())
|
||||
payload Json
|
||||
|
||||
@@index([recorded_at])
|
||||
@@index([kind, source_event_id])
|
||||
}
|
||||
|
||||
// One row per (event, target table, version): pending work for the projector and,
|
||||
// once applied in the same transaction as the target write, the receipt that stops a retry.
|
||||
model LiteLLM_AccountingDelivery {
|
||||
event_id String
|
||||
target AccountingDeliveryTarget
|
||||
version Int
|
||||
state AccountingDeliveryState @default(pending)
|
||||
attempts Int @default(0)
|
||||
next_attempt_at DateTime @default(now())
|
||||
applied_at DateTime?
|
||||
last_error String?
|
||||
|
||||
@@id([event_id, target, version])
|
||||
@@index([target, state, next_attempt_at])
|
||||
}
|
||||
|
|
|
|||
128
tests/integration/database/test_accounting_ledger.py
Normal file
128
tests/integration/database/test_accounting_ledger.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
from psycopg import sql
|
||||
|
||||
from litellm.accounting import DeliveryState, DeliveryTarget, EventKind
|
||||
from tests.integration._support.database import scratch_database
|
||||
|
||||
_SCHEMA: Final = Path(__file__).resolve().parents[3] / "litellm-proxy-extras/litellm_proxy_extras/schema.prisma"
|
||||
_EVENT_INSERT: Final = (
|
||||
'INSERT INTO "LiteLLM_AccountingEvent" '
|
||||
"(event_id, kind, schema_version, payload_hash, occurred_at, payload) "
|
||||
"VALUES ('event', %s, %s, 'hash', '2026-01-01', '{}') RETURNING kind::text, schema_version"
|
||||
)
|
||||
_DELIVERY_INSERT: Final = (
|
||||
'INSERT INTO "LiteLLM_AccountingDelivery" (event_id, target, version, state, attempts) '
|
||||
"VALUES ('event', %s, %s, %s, %s) RETURNING target::text, version, state::text, attempts"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ledger_database() -> Iterator[str]:
|
||||
with scratch_database("accounting") as database_url:
|
||||
result: Final = subprocess.run(
|
||||
[sys.executable, "-I", "-m", "prisma", "migrate", "deploy", "--schema", str(_SCHEMA)],
|
||||
env={**os.environ, "DATABASE_URL": database_url},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=300,
|
||||
)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
yield database_url
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ledger(ledger_database: str) -> Iterator[psycopg.Connection[tuple[object, ...]]]:
|
||||
with psycopg.connect(ledger_database) as connection:
|
||||
try:
|
||||
yield connection
|
||||
finally:
|
||||
connection.rollback()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind", tuple(EventKind))
|
||||
def test_ledger_accepts_every_public_event_kind(
|
||||
ledger: psycopg.Connection[tuple[object, ...]], kind: EventKind
|
||||
) -> None:
|
||||
row: Final = ledger.execute(_EVENT_INSERT, (kind.value, 1)).fetchone()
|
||||
assert row == (kind.value, 1)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("target", tuple(DeliveryTarget))
|
||||
@pytest.mark.parametrize("state", tuple(DeliveryState))
|
||||
def test_delivery_accepts_every_public_target_and_state(
|
||||
ledger: psycopg.Connection[tuple[object, ...]], target: DeliveryTarget, state: DeliveryState
|
||||
) -> None:
|
||||
row: Final = ledger.execute(_DELIVERY_INSERT, (target.value, 1, state.value, 0)).fetchone()
|
||||
assert row == (target.value, 1, state.value, 0)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("enum_name", "values"),
|
||||
[
|
||||
("AccountingEventKind", tuple(value.value for value in EventKind)),
|
||||
("AccountingDeliveryState", tuple(value.value for value in DeliveryState)),
|
||||
("AccountingDeliveryTarget", tuple(value.value for value in DeliveryTarget)),
|
||||
],
|
||||
)
|
||||
def test_migrated_enums_match_the_public_contract(
|
||||
ledger: psycopg.Connection[tuple[object, ...]], enum_name: str, values: tuple[str, ...]
|
||||
) -> None:
|
||||
rows: Final = ledger.execute(
|
||||
"SELECT e.enumlabel FROM pg_enum e JOIN pg_type t ON t.oid = e.enumtypid WHERE t.typname = %s",
|
||||
(enum_name,),
|
||||
).fetchall()
|
||||
assert frozenset(row[0] for row in rows) == frozenset(values)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("version", [0, -1])
|
||||
def test_event_rejects_nonpositive_schema_versions(
|
||||
ledger: psycopg.Connection[tuple[object, ...]], version: int
|
||||
) -> None:
|
||||
with pytest.raises(psycopg.errors.CheckViolation, match="AccountingEvent_schema_version_check"):
|
||||
ledger.execute(_EVENT_INSERT, (EventKind.OUTBOUND_USAGE.value, version))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("version", "attempts", "constraint"),
|
||||
[
|
||||
(0, 0, "AccountingDelivery_version_check"),
|
||||
(-1, 0, "AccountingDelivery_version_check"),
|
||||
(1, -1, "AccountingDelivery_attempts_check"),
|
||||
],
|
||||
)
|
||||
def test_delivery_rejects_invalid_versions_and_attempts(
|
||||
ledger: psycopg.Connection[tuple[object, ...]], version: int, attempts: int, constraint: str
|
||||
) -> None:
|
||||
with pytest.raises(psycopg.errors.CheckViolation, match=constraint):
|
||||
ledger.execute(_DELIVERY_INSERT, (DeliveryTarget.KEY_SPEND.value, version, "pending", attempts))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("column", ["state", "target"])
|
||||
def test_delivery_rejects_unknown_enum_values(ledger: psycopg.Connection[tuple[object, ...]], column: str) -> None:
|
||||
ledger.execute(_DELIVERY_INSERT, (DeliveryTarget.KEY_SPEND.value, 1, "pending", 0))
|
||||
with pytest.raises(psycopg.errors.InvalidTextRepresentation, match="invalid input value for enum"):
|
||||
ledger.execute(
|
||||
sql.SQL('UPDATE "LiteLLM_AccountingDelivery" SET {} = %s').format(sql.Identifier(column)), ("invalid",)
|
||||
)
|
||||
|
||||
|
||||
def test_event_rejects_unknown_kind(ledger: psycopg.Connection[tuple[object, ...]]) -> None:
|
||||
with pytest.raises(psycopg.errors.InvalidTextRepresentation, match="invalid input value for enum"):
|
||||
ledger.execute(_EVENT_INSERT, ("invalid", 1))
|
||||
|
||||
|
||||
def test_new_delivery_defaults_to_pending_with_no_attempts(ledger: psycopg.Connection[tuple[object, ...]]) -> None:
|
||||
row: Final = ledger.execute(
|
||||
'INSERT INTO "LiteLLM_AccountingDelivery" (event_id, target, version) '
|
||||
"VALUES ('event', %s, 1) RETURNING state::text, attempts",
|
||||
(DeliveryTarget.KEY_SPEND.value,),
|
||||
).fetchone()
|
||||
assert row == ("pending", 0)
|
||||
|
|
@ -0,0 +1,122 @@
|
|||
import json
|
||||
import uuid
|
||||
from collections.abc import Mapping
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from pydantic import JsonValue
|
||||
|
||||
MODEL: Final = "claude-haiku-4-5"
|
||||
|
||||
|
||||
def _sse(event: Mapping[str, JsonValue]) -> bytes:
|
||||
return f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode()
|
||||
|
||||
|
||||
def _message(identity: str, stop_reason: str | None) -> dict[str, JsonValue]:
|
||||
return {
|
||||
"id": "msg_" + identity,
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": MODEL,
|
||||
"content": [{"type": "text", "text": "alpha bravo"}],
|
||||
"stop_reason": stop_reason,
|
||||
"stop_sequence": None,
|
||||
"usage": {"input_tokens": 199878, "output_tokens": 12799},
|
||||
}
|
||||
|
||||
|
||||
def _stream(identity: str, stop_reason: str) -> tuple[bytes, ...]:
|
||||
return (
|
||||
_sse({"type": "message_start", "message": {**_message(identity, None), "content": []}}),
|
||||
_sse({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}),
|
||||
_sse({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "alpha bravo"}}),
|
||||
_sse({"type": "content_block_stop", "index": 0}),
|
||||
_sse(
|
||||
{
|
||||
"type": "message_delta",
|
||||
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
|
||||
"usage": {"output_tokens": 12799},
|
||||
}
|
||||
),
|
||||
_sse({"type": "message_stop"}),
|
||||
)
|
||||
|
||||
|
||||
def _anthropic(identity: str, stop_reason: str):
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST" and request.target == "/v1/messages", request.target
|
||||
if json.loads(request.body).get("stream"):
|
||||
return Reply(content_type="text/event-stream", chunks=_stream(identity, stop_reason))
|
||||
return Reply(body=json.dumps(_message(identity, stop_reason)).encode())
|
||||
|
||||
return respond
|
||||
|
||||
|
||||
def _events(gateway: Gateway, path: str, body: Mapping[str, JsonValue]) -> tuple[dict[str, JsonValue], ...]:
|
||||
with gateway.client.stream(
|
||||
"POST", path, json={**body, "stream": True}, headers={"Authorization": f"Bearer {gateway.key}"}
|
||||
) as response:
|
||||
assert response.status_code == 200, response.read().decode()
|
||||
return tuple(
|
||||
json.loads(line.removeprefix("data: "))
|
||||
for line in response.iter_lines()
|
||||
if line.startswith("data: ") and line != "data: [DONE]"
|
||||
)
|
||||
|
||||
|
||||
def _chat_finish_reason(gateway: Gateway, model: str, *, stream: bool) -> JsonValue:
|
||||
body: Final = {"model": model, "max_tokens": 64000, "messages": [{"role": "user", "content": "copy the list"}]}
|
||||
if stream:
|
||||
reasons: Final = tuple(
|
||||
choice["finish_reason"]
|
||||
for chunk in _events(gateway, "/v1/chat/completions", body)
|
||||
for choice in chunk["choices"] # comprehension-ok: chunk -> choices flatten
|
||||
if choice.get("finish_reason")
|
||||
)
|
||||
assert len(reasons) == 1, reasons
|
||||
return reasons[0]
|
||||
response: Final = gateway.request("POST", "/v1/chat/completions", body)
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json()["choices"][0]["finish_reason"]
|
||||
|
||||
|
||||
def _responses_status(gateway: Gateway, model: str, *, stream: bool) -> JsonValue:
|
||||
body: Final = {"model": model, "max_output_tokens": 64000, "input": "copy the list"}
|
||||
if stream:
|
||||
terminal: Final = tuple(
|
||||
event["response"]
|
||||
for event in _events(gateway, "/v1/responses", body)
|
||||
if event["type"] in ("response.completed", "response.incomplete", "response.failed")
|
||||
)
|
||||
assert len(terminal) == 1, terminal
|
||||
return terminal[0]["status"]
|
||||
response: Final = gateway.request("POST", "/v1/responses", body)
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json()["status"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
|
||||
@pytest.mark.parametrize(
|
||||
("stop_reason", "finish_reason"),
|
||||
[("model_context_window_exceeded", "length"), ("max_tokens", "length"), ("end_turn", "stop")],
|
||||
)
|
||||
def test_chat_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, finish_reason: str, stream: bool) -> None:
|
||||
identity: Final = uuid.uuid4().hex
|
||||
with wire_server(_anthropic(identity, stop_reason)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
|
||||
assert _chat_finish_reason(gateway, model, stream=stream) == finish_reason
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
|
||||
@pytest.mark.parametrize(
|
||||
("stop_reason", "status"),
|
||||
[("model_context_window_exceeded", "incomplete"), ("max_tokens", "incomplete"), ("end_turn", "completed")],
|
||||
)
|
||||
def test_responses_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, status: str, stream: bool) -> None:
|
||||
identity: Final = uuid.uuid4().hex
|
||||
with wire_server(_anthropic(identity, stop_reason)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
|
||||
assert _responses_status(gateway, model, stream=stream) == status
|
||||
|
|
@ -413,3 +413,71 @@ def test_resent_tool_history_is_scanned_once(single_worker: Rig, shape: str) ->
|
|||
second: Final = timed(lambda: single_worker.proxy.request("POST", path, following))
|
||||
assert first.value.status_code == 200 and second.value.status_code == 200, (first.value.text, second.value.text)
|
||||
assert_within_factor_of_control(second.seconds, control=first.seconds, factor=0.5)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("placement", ("guardrail", "global"))
|
||||
def test_skip_tool_message_in_guardrail_forwards_tool_results(tmp_path: Path, placement: str) -> None:
|
||||
def configure(configuration: dict[str, JsonValue]) -> None:
|
||||
guardrails: Final = configuration["guardrails"]
|
||||
settings: Final = configuration["litellm_settings"]
|
||||
assert isinstance(guardrails, list) and isinstance(guardrails[0], dict) and isinstance(settings, dict)
|
||||
params: Final = guardrails[0]["litellm_params"]
|
||||
assert isinstance(params, dict)
|
||||
(params if placement == "guardrail" else settings)["skip_tool_message_in_guardrail"] = True
|
||||
|
||||
with tool_rig(tmp_path, configure=configure) as rig:
|
||||
rig.wire.drain()
|
||||
replies: Final = (
|
||||
rig.proxy.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{
|
||||
"model": "claude",
|
||||
"max_tokens": 32,
|
||||
"cache": {"no-cache": True},
|
||||
"messages": [
|
||||
{"role": "user", "content": TEXT},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{"type": "tool_use", "id": "tool_1", "name": "read_file", "input": {"env": TEXT}}
|
||||
],
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "tool_result", "tool_use_id": "tool_1", "content": TEXT}],
|
||||
},
|
||||
],
|
||||
},
|
||||
),
|
||||
rig.proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": "gpt",
|
||||
"cache": {"no-cache": True},
|
||||
"messages": [
|
||||
{"role": "user", "content": TEXT},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "read_file", "arguments": json.dumps({"env": TEXT})},
|
||||
}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": TEXT},
|
||||
],
|
||||
},
|
||||
),
|
||||
)
|
||||
assert all(reply.status_code == 200 for reply in replies), [reply.text for reply in replies]
|
||||
sent: Final = rig.wire.drain()
|
||||
assert len(sent) == 2
|
||||
for request in sent:
|
||||
messages = json.loads(request.body)["messages"]
|
||||
assert SECRET not in json.dumps(messages[:2])
|
||||
assert SECRET in json.dumps(messages[2])
|
||||
|
|
|
|||
0
tests/unit/accounting/__init__.py
Normal file
0
tests/unit/accounting/__init__.py
Normal file
235
tests/unit/accounting/test_measurements.py
Normal file
235
tests/unit/accounting/test_measurements.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
from datetime import datetime, timezone
|
||||
from decimal import Decimal
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.accounting import (
|
||||
CostEvent,
|
||||
Dimension,
|
||||
DimensionName,
|
||||
InboundUsageEvent,
|
||||
Measurement,
|
||||
Metric,
|
||||
MetricName,
|
||||
Outcome,
|
||||
OutboundUsageEvent,
|
||||
PricingSnapshot,
|
||||
measurements_for,
|
||||
)
|
||||
from litellm.types.utils import CallTypes
|
||||
|
||||
_START: Final = datetime(2026, 1, 1, 23, 59, 59, tzinfo=timezone.utc)
|
||||
_END: Final = datetime(2026, 1, 2, tzinfo=timezone.utc)
|
||||
_DIMENSIONS: Final = (
|
||||
Dimension(name=DimensionName.TEAM, value="team"),
|
||||
Dimension(name=DimensionName.TAG, value="alpha"),
|
||||
Dimension(name=DimensionName.TAG, value="beta"),
|
||||
)
|
||||
|
||||
|
||||
def _snapshot(amount: Decimal) -> PricingSnapshot:
|
||||
return PricingSnapshot(
|
||||
calculator_version="test",
|
||||
model="test-model",
|
||||
provider=None,
|
||||
pricing_source="upstream_header",
|
||||
priced_model_key="test-model",
|
||||
upstream_reported_cost=amount,
|
||||
total_cost=amount,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("outcome", "outcome_metric"),
|
||||
[
|
||||
(Outcome.SUCCEEDED, MetricName.SUCCESSFUL_REQUESTS),
|
||||
(Outcome.FAILED, MetricName.FAILED_REQUESTS),
|
||||
(Outcome.CANCELLED, MetricName.FAILED_REQUESTS),
|
||||
],
|
||||
)
|
||||
def test_caller_measurements_keep_both_request_and_outcome_counts(outcome, outcome_metric):
|
||||
event = InboundUsageEvent(
|
||||
event_id="caller-event",
|
||||
request_id="caller",
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_END,
|
||||
outcome=outcome,
|
||||
dimensions=_DIMENSIONS,
|
||||
metrics=(Metric(name=MetricName.REQUESTS, value=Decimal(1)), Metric(name=outcome_metric, value=Decimal(1))),
|
||||
)
|
||||
assert measurements_for(event) == (
|
||||
Measurement(
|
||||
source_event_id="caller-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.REQUESTS, value=Decimal(1)),
|
||||
),
|
||||
Measurement(
|
||||
source_event_id="caller-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=outcome_metric, value=Decimal(1)),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_usage_measurements_keep_every_metric_in_the_start_time_bucket():
|
||||
event = OutboundUsageEvent(
|
||||
event_id="usage-event",
|
||||
attempt_id="attempt",
|
||||
request_id="caller",
|
||||
retry_of_attempt_id=None,
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_END,
|
||||
outcome=Outcome.SUCCEEDED,
|
||||
dimensions=_DIMENSIONS,
|
||||
metrics=(
|
||||
Metric(name=MetricName.PROMPT_TOKENS, value=Decimal(10)),
|
||||
Metric(name=MetricName.COMPLETION_TOKENS, value=Decimal(2)),
|
||||
Metric(name=MetricName.TOTAL_TOKENS, value=Decimal(12)),
|
||||
Metric(name=MetricName.CACHE_READ_INPUT_TOKENS, value=Decimal(3)),
|
||||
Metric(name=MetricName.CACHE_CREATION_INPUT_TOKENS, value=Decimal(1)),
|
||||
),
|
||||
)
|
||||
assert measurements_for(event) == (
|
||||
Measurement(
|
||||
source_event_id="usage-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.PROMPT_TOKENS, value=Decimal(10)),
|
||||
),
|
||||
Measurement(
|
||||
source_event_id="usage-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.COMPLETION_TOKENS, value=Decimal(2)),
|
||||
),
|
||||
Measurement(
|
||||
source_event_id="usage-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.TOTAL_TOKENS, value=Decimal(12)),
|
||||
),
|
||||
Measurement(
|
||||
source_event_id="usage-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.CACHE_READ_INPUT_TOKENS, value=Decimal(3)),
|
||||
),
|
||||
Measurement(
|
||||
source_event_id="usage-event",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.CACHE_CREATION_INPUT_TOKENS, value=Decimal(1)),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("metrics", [None, ()])
|
||||
def test_unknown_or_empty_usage_does_not_fabricate_zero_measurements(metrics):
|
||||
event = OutboundUsageEvent(
|
||||
event_id="usage-event",
|
||||
attempt_id="attempt",
|
||||
request_id=None,
|
||||
retry_of_attempt_id=None,
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_END,
|
||||
outcome=Outcome.SUCCEEDED,
|
||||
dimensions=(),
|
||||
metrics=metrics,
|
||||
)
|
||||
assert measurements_for(event) == ()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("amount", [Decimal(0), Decimal("0.1234567890123456789")])
|
||||
def test_cost_measurements_preserve_authoritative_amount_identity_and_occurrence(amount):
|
||||
event = CostEvent(
|
||||
event_id="cost-event",
|
||||
usage_event_id="usage-event",
|
||||
occurred_at=_END,
|
||||
dimensions=_DIMENSIONS,
|
||||
charges=(Metric(name=MetricName.SPEND, value=amount, unit="USD"),),
|
||||
snapshot=_snapshot(amount),
|
||||
)
|
||||
assert measurements_for(event) == (
|
||||
Measurement(
|
||||
source_event_id="cost-event",
|
||||
occurred_at=_END,
|
||||
dimensions=_DIMENSIONS,
|
||||
metric=Metric(name=MetricName.SPEND, value=amount, unit="USD"),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_billed_retry_and_success_contribute_one_caller_request_and_both_charges():
|
||||
caller = InboundUsageEvent(
|
||||
event_id="caller-event",
|
||||
request_id="caller",
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_END,
|
||||
outcome=Outcome.SUCCEEDED,
|
||||
dimensions=_DIMENSIONS,
|
||||
metrics=(Metric(name=MetricName.REQUESTS, value=Decimal(1)),),
|
||||
)
|
||||
retry = OutboundUsageEvent(
|
||||
event_id="retry-usage",
|
||||
attempt_id="retry",
|
||||
request_id="caller",
|
||||
retry_of_attempt_id=None,
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_END,
|
||||
outcome=Outcome.FAILED,
|
||||
dimensions=_DIMENSIONS,
|
||||
metrics=(Metric(name=MetricName.TOTAL_TOKENS, value=Decimal(4)),),
|
||||
)
|
||||
success = OutboundUsageEvent(
|
||||
event_id="success-usage",
|
||||
attempt_id="success",
|
||||
request_id="caller",
|
||||
retry_of_attempt_id="retry",
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_END,
|
||||
outcome=Outcome.SUCCEEDED,
|
||||
dimensions=_DIMENSIONS,
|
||||
metrics=(Metric(name=MetricName.TOTAL_TOKENS, value=Decimal(12)),),
|
||||
)
|
||||
retry_cost = CostEvent(
|
||||
event_id="retry-cost",
|
||||
usage_event_id="retry-usage",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
charges=(Metric(name=MetricName.SPEND, value=Decimal("0.025"), unit="USD"),),
|
||||
snapshot=_snapshot(Decimal("0.025")),
|
||||
)
|
||||
success_cost = CostEvent(
|
||||
event_id="success-cost",
|
||||
usage_event_id="success-usage",
|
||||
occurred_at=_START,
|
||||
dimensions=_DIMENSIONS,
|
||||
charges=(Metric(name=MetricName.SPEND, value=Decimal("0.125"), unit="USD"),),
|
||||
snapshot=_snapshot(Decimal("0.125")),
|
||||
)
|
||||
measurements = (
|
||||
*measurements_for(caller),
|
||||
*measurements_for(retry),
|
||||
*measurements_for(success),
|
||||
*measurements_for(retry_cost),
|
||||
*measurements_for(success_cost),
|
||||
)
|
||||
assert sum(m.metric.value for m in measurements if m.metric.name is MetricName.REQUESTS) == Decimal(1)
|
||||
assert sum(m.metric.value for m in measurements if m.metric.name is MetricName.TOTAL_TOKENS) == Decimal(16)
|
||||
assert sum(m.metric.value for m in measurements if m.metric.name is MetricName.SPEND) == Decimal("0.150")
|
||||
assert {m.source_event_id for m in measurements} == {
|
||||
"caller-event",
|
||||
"retry-usage",
|
||||
"success-usage",
|
||||
"retry-cost",
|
||||
"success-cost",
|
||||
}
|
||||
484
tests/unit/accounting/test_types.py
Normal file
484
tests/unit/accounting/test_types.py
Normal file
|
|
@ -0,0 +1,484 @@
|
|||
import re
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from decimal import Decimal
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from litellm.accounting import (
|
||||
AdmissionDecision,
|
||||
AggregateQuery,
|
||||
AggregateRow,
|
||||
CostEvent,
|
||||
DeliveryState,
|
||||
DeliveryTarget,
|
||||
Dimension,
|
||||
DimensionFilter,
|
||||
DimensionName,
|
||||
EventKind,
|
||||
Granularity,
|
||||
InboundUsageEvent,
|
||||
Metric,
|
||||
MetricName,
|
||||
Multiplier,
|
||||
Outcome,
|
||||
OutboundUsageEvent,
|
||||
PricedQuantity,
|
||||
PricingSnapshot,
|
||||
Rate,
|
||||
validate_aggregate_rows,
|
||||
)
|
||||
from litellm.accounting.types import PricingUnit
|
||||
from litellm.types.utils import CallTypes
|
||||
|
||||
_START: Final = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
_END: Final = datetime(2026, 1, 2, tzinfo=timezone.utc)
|
||||
_REQUEST: Final = Metric(name=MetricName.REQUESTS, value=Decimal(1))
|
||||
_TOKEN: Final = Metric(name=MetricName.PROMPT_TOKENS, value=Decimal(10))
|
||||
_SPEND: Final = Metric(name=MetricName.SPEND, value=Decimal("0.1234567890123456789"), unit="USD")
|
||||
_TEAM: Final = Dimension(name=DimensionName.TEAM, value="team")
|
||||
|
||||
|
||||
def _inbound(metrics=(_REQUEST,), outcome=Outcome.SUCCEEDED):
|
||||
return InboundUsageEvent(
|
||||
event_id="inbound:request",
|
||||
request_id="request",
|
||||
call_type=CallTypes.acompletion,
|
||||
started_at=_START,
|
||||
ended_at=_START,
|
||||
outcome=outcome,
|
||||
dimensions=(_TEAM,),
|
||||
metrics=metrics,
|
||||
)
|
||||
|
||||
|
||||
def _outbound(metrics=(_TOKEN,), **changes):
|
||||
return OutboundUsageEvent.model_validate(
|
||||
{
|
||||
"event_id": "outbound:attempt",
|
||||
"attempt_id": "attempt",
|
||||
"request_id": "request",
|
||||
"retry_of_attempt_id": None,
|
||||
"call_type": CallTypes.acompletion,
|
||||
"started_at": _START,
|
||||
"ended_at": _END,
|
||||
"outcome": Outcome.SUCCEEDED,
|
||||
"dimensions": (_TEAM,),
|
||||
"metrics": metrics,
|
||||
**changes,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _query(**changes):
|
||||
return AggregateQuery.model_validate(
|
||||
{
|
||||
"metric": MetricName.SPEND,
|
||||
"filters": (DimensionFilter(name=DimensionName.TEAM, values=("one", "two")),),
|
||||
"group_by": (DimensionName.MODEL,),
|
||||
"start": _START,
|
||||
"end": _END,
|
||||
"granularity": Granularity.DAY,
|
||||
**changes,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_wire_values_preserve_decimal_precision_and_normalize_aware_times():
|
||||
offset = timezone(timedelta(hours=-8))
|
||||
event = CostEvent(
|
||||
event_id="cost:attempt",
|
||||
usage_event_id="outbound:attempt",
|
||||
occurred_at=datetime(2025, 12, 31, 16, tzinfo=offset),
|
||||
dimensions=(_TEAM,),
|
||||
charges=(_SPEND,),
|
||||
snapshot=_snapshot(),
|
||||
)
|
||||
wire = event.model_dump(mode="json")
|
||||
assert wire["occurred_at"] == "2026-01-01T00:00:00Z"
|
||||
assert wire["charges"] == [{"name": "spend", "value": "0.1234567890123456789", "unit": "USD"}]
|
||||
assert event.occurred_at == _START
|
||||
assert CostEvent.model_validate_json(event.model_dump_json()).charges[0].value == _SPEND.value
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("name", "value", "unit", "message"),
|
||||
[
|
||||
(MetricName.SPEND, Decimal(1), "count", "SPEND requires USD"),
|
||||
(MetricName.REQUESTS, Decimal(1), "USD", "SPEND requires USD"),
|
||||
(MetricName.PROMPT_TOKENS, Decimal("1.5"), "count", "integral"),
|
||||
(MetricName.SPEND, Decimal("NaN"), "USD", "finite"),
|
||||
(MetricName.SPEND, Decimal("Infinity"), "USD", "finite"),
|
||||
(MetricName.SPEND, Decimal(-1), "USD", "greater than or equal"),
|
||||
(MetricName.SPEND, 0.1, "USD", "instance of Decimal"),
|
||||
],
|
||||
)
|
||||
def test_metrics_reject_wrong_units_nonfinite_values_and_binary_float_inputs(name, value, unit, message):
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
Metric(name=name, value=value, unit=unit)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("metrics", [(_TOKEN,), (_SPEND,), ()])
|
||||
def test_inbound_activity_cannot_own_usage_or_cost_or_omit_request_count(metrics):
|
||||
with pytest.raises(ValidationError, match="Inbound events"):
|
||||
_inbound(metrics)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("outcome", "metric"),
|
||||
[
|
||||
(Outcome.FAILED, Metric(name=MetricName.SUCCESSFUL_REQUESTS, value=Decimal(1))),
|
||||
(Outcome.SUCCEEDED, Metric(name=MetricName.FAILED_REQUESTS, value=Decimal(1))),
|
||||
],
|
||||
)
|
||||
def test_inbound_outcome_counts_cannot_contradict_outcome(outcome, metric):
|
||||
with pytest.raises(ValidationError, match="count must match outcome"):
|
||||
_inbound((_REQUEST, metric), outcome)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("metrics", [(_REQUEST,), (_SPEND,)])
|
||||
def test_outbound_attempts_cannot_own_caller_counts_or_charges(metrics):
|
||||
with pytest.raises(ValidationError, match="Outbound events own only"):
|
||||
_outbound(metrics)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("changes", "message"),
|
||||
[
|
||||
({"started_at": datetime(2026, 1, 1)}, "timezone"),
|
||||
({"ended_at": _START - timedelta(seconds=1)}, "End must follow start"),
|
||||
({"retry_of_attempt_id": "attempt"}, "cannot retry itself"),
|
||||
({"dimensions": (_TEAM, _TEAM)}, "Duplicate dimension values"),
|
||||
({"dimensions": (_TEAM, Dimension(name=DimensionName.TEAM, value="other"))}, "Only tag and model access group"),
|
||||
(
|
||||
{
|
||||
"metrics": (
|
||||
Metric(name=MetricName.COMPLETION_TOKENS, value=Decimal(5)),
|
||||
Metric(name=MetricName.REASONING_TOKENS, value=Decimal(6)),
|
||||
)
|
||||
},
|
||||
"reasoning_tokens cannot exceed completion_tokens",
|
||||
),
|
||||
(
|
||||
{
|
||||
"metrics": (
|
||||
Metric(name=MetricName.CACHE_CREATION_INPUT_TOKENS, value=Decimal(5)),
|
||||
Metric(name=MetricName.CACHE_CREATION_INPUT_TOKENS_ABOVE_1HR, value=Decimal(6)),
|
||||
)
|
||||
},
|
||||
"cache_creation_input_tokens_above_1hr cannot exceed cache_creation_input_tokens",
|
||||
),
|
||||
({"event_id": " "}, "pattern"),
|
||||
({"metrics": (_TOKEN, _TOKEN)}, "Duplicate metric names"),
|
||||
],
|
||||
)
|
||||
def test_outbound_events_reject_invalid_identity_time_and_duplicate_attribution(changes, message):
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
_outbound(**changes)
|
||||
|
||||
|
||||
def test_unknown_and_known_zero_usage_are_distinct_and_values_are_immutable():
|
||||
unknown = _outbound(None)
|
||||
zero = _outbound((Metric(name=MetricName.PROMPT_TOKENS, value=Decimal(0)),))
|
||||
assert unknown.metrics is None
|
||||
assert zero.metrics is not None and zero.metrics[0].value == 0
|
||||
assert _inbound().started_at == _inbound().ended_at
|
||||
with pytest.raises(ValidationError, match="frozen"):
|
||||
zero.attempt_id = "different"
|
||||
with pytest.raises(ValidationError, match="frozen"):
|
||||
zero.metrics[0].value = Decimal(10)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("charges", [(), (_TOKEN,), (_SPEND, _SPEND)])
|
||||
def test_cost_events_require_one_authoritative_spend(charges):
|
||||
with pytest.raises(ValidationError, match=r"authoritative USD spend|Duplicate metric"):
|
||||
CostEvent(
|
||||
event_id="cost",
|
||||
usage_event_id="usage",
|
||||
occurred_at=_START,
|
||||
dimensions=(),
|
||||
charges=charges,
|
||||
snapshot=_snapshot(),
|
||||
)
|
||||
|
||||
|
||||
def test_cost_event_cannot_reuse_its_usage_identity():
|
||||
with pytest.raises(ValidationError, match="distinct identities"):
|
||||
CostEvent(
|
||||
event_id="usage",
|
||||
usage_event_id="usage",
|
||||
occurred_at=_START,
|
||||
dimensions=(),
|
||||
charges=(_SPEND,),
|
||||
snapshot=_snapshot(),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("granularity", [Granularity.DAY, None])
|
||||
def test_daily_and_whole_interval_queries_preserve_filter_and_group_semantics(granularity):
|
||||
query = _query(granularity=granularity)
|
||||
assert query.granularity is granularity
|
||||
assert query.filters[0].values == ("one", "two")
|
||||
assert query.group_by == (DimensionName.MODEL,)
|
||||
assert query.start == _START and query.end == _END
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("changes", "message"),
|
||||
[
|
||||
({"end": _START}, "End must follow start"),
|
||||
({"start": _START + timedelta(hours=1)}, "UTC day boundaries"),
|
||||
({"end": _END + timedelta(microseconds=1)}, "UTC day boundaries"),
|
||||
({"granularity": "hour"}, "day"),
|
||||
({"group_by": (DimensionName.TEAM, DimensionName.TEAM)}, "Duplicate group-by"),
|
||||
({"filters": (DimensionFilter(name=DimensionName.TEAM, values=("one",)),) * 2}, "one IN filter"),
|
||||
({"filters": ({"name": "team", "values": ()},)}, "at least 1"),
|
||||
],
|
||||
)
|
||||
def test_queries_reject_unsupported_or_ambiguous_ranges_and_filters(changes, message):
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
_query(**changes)
|
||||
|
||||
|
||||
def test_rows_allow_adjacent_buckets_and_overlapping_distinct_groups():
|
||||
first = AggregateRow(start=_START, end=_END, dimensions=(_TEAM,), value=Decimal(1))
|
||||
second = AggregateRow(start=_END, end=_END + timedelta(days=1), dimensions=(_TEAM,), value=Decimal(2))
|
||||
other = AggregateRow(start=_START, end=_END, dimensions=(), value=Decimal(3))
|
||||
rows = (second, other, first)
|
||||
assert validate_aggregate_rows(rows) is rows
|
||||
assert validate_aggregate_rows(()) == ()
|
||||
|
||||
|
||||
def test_rows_reject_overlap_within_a_group_regardless_of_dimension_order():
|
||||
model = Dimension(name=DimensionName.MODEL, value="model")
|
||||
first = AggregateRow(start=_START, end=_END, dimensions=(_TEAM, model), value=Decimal(1))
|
||||
overlapping = AggregateRow(start=_START, end=_END, dimensions=(model, _TEAM), value=Decimal(2))
|
||||
with pytest.raises(ValueError, match="overlap within a dimension group"):
|
||||
validate_aggregate_rows((first, overlapping))
|
||||
|
||||
|
||||
def test_rows_reject_empty_intervals():
|
||||
with pytest.raises(ValidationError, match="End must follow start"):
|
||||
AggregateRow(start=_START, end=_START, dimensions=(), value=Decimal(0))
|
||||
|
||||
|
||||
def test_admission_decision_preserves_a_denial_reason():
|
||||
decision = AdmissionDecision(allowed=False, reason="Team budget exhausted")
|
||||
assert not decision.allowed
|
||||
assert decision.reason == "Team budget exhausted"
|
||||
|
||||
|
||||
def _snapshot(**changes: object) -> PricingSnapshot:
|
||||
return PricingSnapshot.model_validate(
|
||||
{
|
||||
"calculator_version": "1.80.0",
|
||||
"model": "gpt-4o",
|
||||
"provider": "openai",
|
||||
"pricing_source": "cost_map",
|
||||
"priced_model_key": "openai/gpt-4o",
|
||||
"multipliers": (Multiplier(name="data_residency", value="eu", factor=Decimal("1.1")),),
|
||||
"rates": (Rate(measure="input", unit="token", usd_per_unit=Decimal("0.0000025")),),
|
||||
"usage": (PricedQuantity(measure="input", unit="token", value=_TOKEN.value),),
|
||||
"input_cost": Decimal("0.000025"),
|
||||
"total_cost": _SPEND.value,
|
||||
**changes,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_cost_event_carries_a_snapshot_whose_total_matches_the_spend():
|
||||
event = CostEvent(
|
||||
event_id="cost:attempt",
|
||||
usage_event_id="outbound:attempt",
|
||||
occurred_at=_START,
|
||||
dimensions=(_TEAM,),
|
||||
charges=(_SPEND,),
|
||||
snapshot=_snapshot(),
|
||||
)
|
||||
assert event.snapshot is not None
|
||||
assert event.snapshot.total_cost == _SPEND.value
|
||||
assert event.snapshot.multipliers[0].factor == Decimal("1.1")
|
||||
with pytest.raises(ValidationError, match="Snapshot total must equal the authoritative spend"):
|
||||
CostEvent(
|
||||
event_id="cost:attempt",
|
||||
usage_event_id="outbound:attempt",
|
||||
occurred_at=_START,
|
||||
dimensions=(_TEAM,),
|
||||
charges=(_SPEND,),
|
||||
snapshot=_snapshot(total_cost=Decimal("0.5")),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("changes", "message"),
|
||||
(
|
||||
(
|
||||
{
|
||||
"rates": (
|
||||
Rate(measure="input", unit="token", usd_per_unit=Decimal(1)),
|
||||
Rate(measure="input", unit="token", usd_per_unit=Decimal(2)),
|
||||
)
|
||||
},
|
||||
"Duplicate rate measures and units",
|
||||
),
|
||||
({"usage": (_SPEND,)}, "PricedQuantity"),
|
||||
({"usage": (_REQUEST,)}, "PricedQuantity"),
|
||||
({"pricing_source": "guess"}, "pricing_source"),
|
||||
({"multipliers": ({"name": "x", "value": "y", "factor": Decimal(0)},)}, "factor"),
|
||||
({"schema_version": 2}, "schema_version"),
|
||||
(
|
||||
{"additional_costs": ({"name": "fee", "amount": Decimal(1)}, {"name": "fee", "amount": Decimal(2)})},
|
||||
"Duplicate additional cost names",
|
||||
),
|
||||
),
|
||||
)
|
||||
def test_snapshots_reject_ambiguous_rates_wrong_usage_and_unknown_sources(changes: Mapping[str, object], message: str):
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
_snapshot(**changes)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("unit", "value"),
|
||||
[("token", "10"), ("second", "1.25"), ("character", "25"), ("image", "2"), ("request", "1")],
|
||||
)
|
||||
def test_snapshot_preserves_matching_rates_and_quantities_for_each_billing_unit(unit: PricingUnit, value: str) -> None:
|
||||
quantity: Final = PricedQuantity(measure="input", unit=unit, value=Decimal(value))
|
||||
rate: Final = Rate(measure="input", unit=unit, usd_per_unit=Decimal("0.003"))
|
||||
amount: Final = quantity.value * rate.usd_per_unit
|
||||
snapshot: Final = _snapshot(rates=(rate,), usage=(quantity,), total_cost=amount)
|
||||
restored: Final = PricingSnapshot.model_validate_json(snapshot.model_dump_json())
|
||||
assert restored == snapshot
|
||||
assert restored.usage[0].value * restored.rates[0].usd_per_unit == amount
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("usage", "message"),
|
||||
[
|
||||
((), "Every rate requires"),
|
||||
(({"measure": "output", "unit": "token", "value": Decimal(10)},), "Every rate requires"),
|
||||
(({"measure": "input", "unit": "second", "value": Decimal(10)},), "Every rate requires"),
|
||||
(
|
||||
({"measure": "input", "unit": "token", "value": Decimal(10)},) * 2,
|
||||
"Duplicate priced quantity",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_snapshot_rejects_missing_mismatched_or_duplicate_priced_quantities(
|
||||
usage: tuple[Mapping[str, object], ...], message: str
|
||||
) -> None:
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
_snapshot(usage=usage)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("unit", ["token", "character", "image", "request"])
|
||||
def test_priced_counts_remain_integral(unit: PricingUnit) -> None:
|
||||
with pytest.raises(ValidationError, match="integral"):
|
||||
PricedQuantity(measure="input", unit=unit, value=Decimal("1.25"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "message"),
|
||||
[
|
||||
(Decimal(-1), "greater than or equal"),
|
||||
(Decimal("NaN"), "finite"),
|
||||
(Decimal("Infinity"), "finite"),
|
||||
(1.25, "instance of Decimal"),
|
||||
],
|
||||
)
|
||||
def test_pricing_values_reject_invalid_amounts(value: Decimal | float, message: str) -> None:
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
PricedQuantity.model_validate({"measure": "input", "unit": "second", "value": value})
|
||||
with pytest.raises(ValidationError, match=message):
|
||||
Rate.model_validate({"measure": "input", "unit": "second", "usd_per_unit": value})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("snapshot_fields", [{}, {"snapshot": None}])
|
||||
def test_cost_event_cannot_omit_its_derivation(snapshot_fields: Mapping[str, object]) -> None:
|
||||
with pytest.raises(ValidationError, match="snapshot"):
|
||||
CostEvent.model_validate(
|
||||
{
|
||||
"event_id": "cost",
|
||||
"usage_event_id": "usage",
|
||||
"occurred_at": _START,
|
||||
"dimensions": (),
|
||||
"charges": (_SPEND,),
|
||||
**snapshot_fields,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_upstream_charge_snapshot_preserves_reported_cost_without_inventing_rates() -> None:
|
||||
snapshot: Final = _snapshot(
|
||||
pricing_source="upstream_header", rates=(), usage=(), upstream_reported_cost=_SPEND.value
|
||||
)
|
||||
restored: Final = PricingSnapshot.model_validate_json(snapshot.model_dump_json())
|
||||
assert restored.upstream_reported_cost == restored.total_cost == _SPEND.value
|
||||
assert restored.rates == restored.usage == ()
|
||||
|
||||
|
||||
def test_subset_token_counts_round_trip_without_changing_totals() -> None:
|
||||
metrics: Final = (
|
||||
Metric(name=MetricName.COMPLETION_TOKENS, value=Decimal(10)),
|
||||
Metric(name=MetricName.REASONING_TOKENS, value=Decimal(3)),
|
||||
Metric(name=MetricName.CACHE_CREATION_INPUT_TOKENS, value=Decimal(8)),
|
||||
Metric(name=MetricName.CACHE_CREATION_INPUT_TOKENS_ABOVE_1HR, value=Decimal(2)),
|
||||
)
|
||||
event: Final = _outbound(metrics=metrics)
|
||||
assert OutboundUsageEvent.model_validate_json(event.model_dump_json()).metrics == metrics
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"schema_path",
|
||||
[
|
||||
"schema.prisma",
|
||||
"litellm/proxy/schema.prisma",
|
||||
"litellm-proxy-extras/litellm_proxy_extras/schema.prisma",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("enum_name", "python_enum"),
|
||||
[
|
||||
("AccountingEventKind", EventKind),
|
||||
("AccountingDeliveryState", DeliveryState),
|
||||
("AccountingDeliveryTarget", DeliveryTarget),
|
||||
],
|
||||
)
|
||||
def test_prisma_accepts_exactly_the_public_accounting_enum_values(
|
||||
schema_path: str, enum_name: str, python_enum: type[Enum]
|
||||
) -> None:
|
||||
schema: Final = (Path(__file__).resolve().parents[3] / schema_path).read_text()
|
||||
declaration: Final = re.search(r"enum " + enum_name + r"\s*\{([^}]+)\}", schema)
|
||||
assert declaration is not None, f"Missing persisted enum {enum_name}"
|
||||
assert frozenset(declaration[1].split()) == frozenset(member.value for member in python_enum)
|
||||
|
||||
|
||||
def test_one_event_carries_every_matched_model_access_group():
|
||||
groups = (
|
||||
Dimension(name=DimensionName.MODEL_ACCESS_GROUP, value="gpt-group"),
|
||||
Dimension(name=DimensionName.MODEL_ACCESS_GROUP, value="claude-group"),
|
||||
)
|
||||
assert _outbound(dimensions=(_TEAM, *groups)).dimensions == (_TEAM, *groups)
|
||||
with pytest.raises(ValidationError, match="Duplicate dimension values"):
|
||||
_outbound(dimensions=(groups[0], groups[0]))
|
||||
|
||||
|
||||
def test_snapshots_reject_duplicate_multipliers_and_contradictory_token_counts():
|
||||
multiplier = Multiplier(name="data_residency", value="eu", factor=Decimal("1.1"))
|
||||
with pytest.raises(ValidationError, match="Duplicate multiplier names"):
|
||||
_snapshot(multipliers=(multiplier, multiplier))
|
||||
with pytest.raises(ValidationError, match="reasoning_tokens cannot exceed completion_tokens"):
|
||||
_snapshot(
|
||||
usage=(
|
||||
PricedQuantity(measure="input", unit="token", value=_TOKEN.value),
|
||||
PricedQuantity(measure="completion_tokens", unit="token", value=Decimal(5)),
|
||||
PricedQuantity(measure="reasoning_tokens", unit="token", value=Decimal(6)),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_agent_live_spend_is_a_delivery_target():
|
||||
assert DeliveryTarget.AGENT_SPEND.value == "LiteLLM_AgentsTable"
|
||||
|
|
@ -256,6 +256,7 @@ class TestMapFinishReasonAnthropic:
|
|||
("max_tokens", "length"),
|
||||
("tool_use", "tool_calls"),
|
||||
("compaction", "length"),
|
||||
("model_context_window_exceeded", "length"),
|
||||
("content_filtered", "content_filter"),
|
||||
],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
import json
|
||||
from functools import reduce
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from pydantic import JsonValue
|
||||
|
|
@ -924,6 +925,34 @@ def test_native_writer_visits_each_tool_and_system_slot(data):
|
|||
assert "payload" not in json.dumps(data)
|
||||
|
||||
|
||||
def test_skip_tool_message_leaves_tool_results_and_masks_the_rest() -> None:
|
||||
data: Final[dict[str, object]] = {
|
||||
"messages": [
|
||||
{"role": "user", "content": "secret"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [{"type": "tool_use", "id": "tool", "name": "read", "input": {"env": "secret"}}],
|
||||
"tool_calls": [{"id": "call", "function": {"name": "read", "arguments": '{"env": "secret"}'}}],
|
||||
},
|
||||
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "tool", "content": "secret"}]},
|
||||
{"role": "tool", "tool_call_id": "call", "content": "secret"},
|
||||
]
|
||||
}
|
||||
assert walk_user_text(data, lambda text: text.replace("secret", "[MASKED]"), skip_tool_message=True) == 3
|
||||
assert data == {
|
||||
"messages": [
|
||||
{"role": "user", "content": "[MASKED]"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [{"type": "tool_use", "id": "tool", "name": "read", "input": {"env": "[MASKED]"}}],
|
||||
"tool_calls": [{"id": "call", "function": {"name": "read", "arguments": '{"env": "[MASKED]"}'}}],
|
||||
},
|
||||
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "tool", "content": "secret"}]},
|
||||
{"role": "tool", "tool_call_id": "call", "content": "secret"},
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"part",
|
||||
(
|
||||
|
|
|
|||
|
|
@ -42,6 +42,10 @@ class TestSpendLogHits:
|
|||
assert guard.spend_log_hits(diff) == ("litellm/proxy/schema.prisma: a LiteLLM_SpendLogs",)
|
||||
|
||||
|
||||
def _review(login: str, state: str, commit_id: str | None = "head") -> "guard.Review":
|
||||
return guard.Review(state=state, user=guard.Reviewer(login=login), commit_id=commit_id)
|
||||
|
||||
|
||||
class TestApprovedBy:
|
||||
def test_the_latest_review_of_the_approver_decides(self):
|
||||
reviews: Final = tuple(
|
||||
|
|
@ -61,6 +65,42 @@ class TestApprovedBy:
|
|||
assert guard.approved_by((), "kerry-berri") is False
|
||||
|
||||
|
||||
class TestLatestApproval:
|
||||
def test_the_approvers_last_approved_review_is_returned_with_its_commit(self):
|
||||
reviews: Final = (
|
||||
_review("kerry-berri", "APPROVED", "old"),
|
||||
_review("someone-else", "APPROVED", "head"),
|
||||
_review("kerry-berri", "APPROVED", "new"),
|
||||
_review("kerry-berri", "COMMENTED", "head"),
|
||||
)
|
||||
assert guard.latest_approval(reviews, "kerry-berri") is reviews[2]
|
||||
assert guard.latest_approval(reviews[:2], "kerry-berri") is reviews[0]
|
||||
assert guard.latest_approval(reviews[1:2], "kerry-berri") is None
|
||||
assert guard.latest_approval((), "kerry-berri") is None
|
||||
|
||||
|
||||
class TestApprovalCovers:
|
||||
_HITS: Final = ("schema.prisma: a LiteLLM_SpendLogs", "schema.prisma: b LiteLLM_SpendLogs")
|
||||
|
||||
def test_an_approval_on_the_head_covers_it_without_reading_the_approved_commit(self):
|
||||
def unreadable(commit: str) -> None:
|
||||
raise AssertionError(f"read {commit}")
|
||||
|
||||
assert guard.approval_covers(_review("kerry-berri", "APPROVED", "head"), "head", self._HITS, unreadable) is True
|
||||
|
||||
def test_an_older_approval_covers_the_head_only_when_it_saw_every_spend_log_line(self):
|
||||
approval: Final = _review("kerry-berri", "APPROVED", "old")
|
||||
assert guard.approval_covers(approval, "head", self._HITS, lambda commit: self._HITS) is True
|
||||
assert guard.approval_covers(approval, "head", self._HITS[:1], lambda commit: self._HITS) is True
|
||||
assert guard.approval_covers(approval, "head", self._HITS, lambda commit: self._HITS[:1]) is False
|
||||
|
||||
def test_an_approval_whose_commit_cannot_be_read_does_not_cover_the_head(self):
|
||||
gone: Final = _review("kerry-berri", "APPROVED", "gone")
|
||||
assert guard.approval_covers(gone, "head", self._HITS, lambda commit: None) is False
|
||||
without_commit: Final = _review("kerry-berri", "APPROVED", None)
|
||||
assert guard.approval_covers(without_commit, "head", self._HITS, lambda commit: self._HITS) is False
|
||||
|
||||
|
||||
class TestExitCode:
|
||||
def test_a_pull_request_without_spend_log_lines_passes_without_reviews(self, tmp_path, monkeypatch, capsys):
|
||||
monkeypatch.setattr(guard, "added_diff", lambda base, head: _diff("schema.prisma", " foo String", 1))
|
||||
|
|
@ -75,5 +115,15 @@ class TestExitCode:
|
|||
reviews.write_text('[{"user": {"login": "someone-else"}, "state": "APPROVED"}]')
|
||||
assert guard.main(["--base", "a", "--head", "b", "--approver", "kerry-berri", "--reviews", str(reviews)]) == 1
|
||||
assert "schema.prisma: x LiteLLM_SpendLogs" in capsys.readouterr().out
|
||||
reviews.write_text('[{"user": {"login": "kerry-berri"}, "state": "APPROVED"}]')
|
||||
reviews.write_text('[{"user": {"login": "kerry-berri"}, "state": "APPROVED", "commit_id": "b"}]')
|
||||
assert guard.main(["--base", "a", "--head", "b", "--approver", "kerry-berri", "--reviews", str(reviews)]) == 0
|
||||
|
||||
def test_an_approval_before_a_new_spend_log_line_was_pushed_fails(self, tmp_path, monkeypatch, capsys):
|
||||
monkeypatch.setattr(guard, "added_diff", lambda base, head: _diff("schema.prisma", " x LiteLLM_SpendLogs", 7))
|
||||
monkeypatch.setattr(guard, "reviewed_hits", lambda base, commit: () if commit == "old" else None, raising=False)
|
||||
reviews: Final = tmp_path / "r.json"
|
||||
reviews.write_text('[{"user": {"login": "kerry-berri"}, "state": "APPROVED", "commit_id": "old"}]')
|
||||
assert guard.main(["--base", "a", "--head", "b", "--approver", "kerry-berri", "--reviews", str(reviews)]) == 1
|
||||
assert "changed after @kerry-berri approved old" in capsys.readouterr().out
|
||||
reviews.write_text('[{"user": {"login": "kerry-berri"}, "state": "APPROVED", "commit_id": "gone"}]')
|
||||
assert guard.main(["--base", "a", "--head", "b", "--approver", "kerry-berri", "--reviews", str(reviews)]) == 1
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue