mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-25 23:48:00 +00:00
feat(gateway): cache-aware cost accounting (#3920)
* feat(gateway): cache-aware cost accounting + /api/console observability endpoints - Capture prompt-cache hits (usage_metadata.input_token_details.cache_read) in RunJournal and SubagentTokenCollector as a sparse cache_read_tokens key in token_usage_by_model (JSON field — no schema migration; legacy bucket shapes unchanged) - New read-only /api/console router: GET /stats (headline counters), GET /runs (cross-thread paginated history joined with thread titles), GET /usage (zero-filled daily token series + per-model breakdown); user-scoped, 503 on the memory database backend - Optional models[*].pricing (currency, input_per_million, output_per_million, input_cache_hit_per_million) powers real spend estimation; cache-hit input tokens are billed at the hit price (omitted hit price falls back to the miss price as a conservative upper bound); unpriced models yield cost: null - create_chat_model strips the presentation-only pricing block so it never reaches the provider client (unknown kwargs are forwarded into the completion payload and break live calls) - Tests: console router SQLite round-trips, journal/collector cache capture incl. a DeepSeek raw-usage pin test, factory strip regression Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor: address review feedback on cost sum and sparse cache_read_tokens - console.py: replace the walrus-in-generator total-cost sum with an explicit loop (review noted the multi-line form reads ambiguously) - token_collector.py: omit cache_read_tokens from usage records when the provider reported no cache hits, matching the journal's sparse per-model bucket shape; absent is treated as 0 downstream - add a regression test pinning the sparse record shape Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: coffeeFish <codeingforcoffee@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
codeingforcoffee
GitHub
parent
15454b6fec
commit
4669d3c089
@ -298,6 +298,7 @@ CORS is same-origin by default when requests enter through nginx on port 2026. S
|
||||
|--------|-----------|
|
||||
| **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details |
|
||||
| **Features** (`/api/features`) | `GET /` - report config-gated feature availability (currently `agents_api.enabled`) for frontend UI gating |
|
||||
| **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured |
|
||||
| **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - update config (saves to extensions_config.json) |
|
||||
| **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) |
|
||||
| **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |
|
||||
|
||||
@ -18,6 +18,7 @@ from app.gateway.routers import (
|
||||
auth,
|
||||
channel_connections,
|
||||
channels,
|
||||
console,
|
||||
features,
|
||||
feedback,
|
||||
github_webhooks,
|
||||
@ -414,6 +415,9 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for
|
||||
# Features API is mounted at /api/features
|
||||
app.include_router(features.router)
|
||||
|
||||
# Console API (cross-thread observability) is mounted at /api/console
|
||||
app.include_router(console.router)
|
||||
|
||||
# MCP API is mounted at /api/mcp
|
||||
app.include_router(mcp.router)
|
||||
|
||||
|
||||
492
backend/app/gateway/routers/console.py
Normal file
492
backend/app/gateway/routers/console.py
Normal file
@ -0,0 +1,492 @@
|
||||
"""Read-only operations-console endpoints.
|
||||
|
||||
Aggregates observability data across all of the current user's threads: run
|
||||
history, token spend over time, and asset counts — the data layer for an
|
||||
operations dashboard or any external monitoring consumer.
|
||||
|
||||
This is a reporting layer, not a runtime path: it issues short-lived read-only
|
||||
queries against the harness-owned ``runs`` / ``threads_meta`` tables instead of
|
||||
widening the runtime ``RunStore`` surface. Requires a SQL database backend
|
||||
(``database.backend: sqlite | postgres``); returns 503 on the memory backend,
|
||||
which persists no run history to report on.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import UTC, datetime, time, timedelta
|
||||
from typing import NamedTuple
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, Request
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from app.gateway.authz import require_permission
|
||||
from app.gateway.deps import get_current_user
|
||||
from deerflow.config import get_app_config
|
||||
from deerflow.config.agents_config import list_custom_agents
|
||||
from deerflow.persistence.engine import get_session_factory
|
||||
from deerflow.persistence.run.model import RunRow
|
||||
from deerflow.persistence.thread_meta.model import ThreadMetaRow
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/console", tags=["console"])
|
||||
|
||||
_ACTIVE_STATUSES = ("pending", "running")
|
||||
_FAILED_STATUSES = ("error", "timeout")
|
||||
|
||||
# Cap the error excerpt in list responses; the full text stays on the run row.
|
||||
_ERROR_EXCERPT_CHARS = 300
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Response models
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ConsoleStatsResponse(BaseModel):
|
||||
"""Headline counters for the console dashboard."""
|
||||
|
||||
total_runs: int = Field(..., description="All recorded runs for the current user")
|
||||
active_runs: int = Field(..., description="Runs currently pending or running")
|
||||
failed_runs: int = Field(..., description="Runs that ended in error or timeout")
|
||||
total_threads: int = Field(..., description="Conversation threads owned by the current user")
|
||||
total_agents: int = Field(..., description="Custom agents owned by the current user")
|
||||
total_tokens: int = Field(..., description="Tokens consumed across all recorded runs")
|
||||
total_cost: float | None = Field(default=None, description="Estimated spend across priced runs; null when no models[*].pricing is configured")
|
||||
currency: str | None = Field(default=None, description="Display currency taken from the first configured pricing entry")
|
||||
|
||||
|
||||
class ConsoleRunItem(BaseModel):
|
||||
"""One run in the cross-thread run listing."""
|
||||
|
||||
run_id: str
|
||||
thread_id: str
|
||||
thread_title: str | None = Field(default=None, description="Display name from threads_meta, if tracked")
|
||||
assistant_id: str | None = None
|
||||
status: str
|
||||
model_name: str | None = None
|
||||
created_at: datetime | None = None
|
||||
updated_at: datetime | None = None
|
||||
duration_seconds: float | None = Field(default=None, description="Wall-clock duration; live elapsed time for active runs")
|
||||
total_tokens: int = 0
|
||||
message_count: int = 0
|
||||
cost: float | None = Field(default=None, description="Estimated spend for this run; null when its models are unpriced")
|
||||
error: str | None = Field(default=None, description="Error excerpt for failed runs")
|
||||
|
||||
|
||||
class ConsoleRunsResponse(BaseModel):
|
||||
"""Paginated cross-thread run listing, newest first."""
|
||||
|
||||
runs: list[ConsoleRunItem]
|
||||
has_more: bool
|
||||
|
||||
|
||||
class ConsoleUsageDay(BaseModel):
|
||||
"""Token usage aggregated over one local-time day."""
|
||||
|
||||
date: str = Field(..., description="Local date (YYYY-MM-DD) per the requested tz offset")
|
||||
total_tokens: int = 0
|
||||
input_tokens: int = 0
|
||||
output_tokens: int = 0
|
||||
runs: int = 0
|
||||
cost: float = Field(default=0.0, description="Estimated spend for the day across priced runs")
|
||||
|
||||
|
||||
class ConsoleUsageModelBreakdown(BaseModel):
|
||||
"""Token usage attributed to one model."""
|
||||
|
||||
tokens: int = 0
|
||||
runs: int = Field(default=0, description="Runs that used this model (non-exclusive)")
|
||||
cost: float | None = Field(default=None, description="Estimated spend for this model; null when unpriced")
|
||||
input_tokens: int = Field(default=0, description="Input tokens attributed to this model")
|
||||
cache_read_tokens: int = Field(default=0, description="Prompt-cache-hit input tokens attributed to this model")
|
||||
|
||||
|
||||
class ConsoleUsageResponse(BaseModel):
|
||||
"""Daily token-usage series plus per-model breakdown for the window."""
|
||||
|
||||
days: list[ConsoleUsageDay]
|
||||
by_model: dict[str, ConsoleUsageModelBreakdown]
|
||||
total_tokens: int
|
||||
total_runs: int
|
||||
total_cost: float | None = Field(default=None, description="Estimated spend for the window; null when no pricing is configured")
|
||||
currency: str | None = Field(default=None, description="Display currency taken from the first configured pricing entry")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _session_factory_or_503():
|
||||
sf = get_session_factory()
|
||||
if sf is None:
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="Console requires a SQL database backend; set database.backend to sqlite or postgres in config.yaml.",
|
||||
)
|
||||
return sf
|
||||
|
||||
|
||||
def _as_utc(dt: datetime | None) -> datetime | None:
|
||||
"""Normalize DB timestamps: SQLite round-trips them naive, Postgres aware."""
|
||||
if dt is None:
|
||||
return None
|
||||
return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pricing — real spend estimation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _ModelPricing(NamedTuple):
|
||||
input_per_million: float
|
||||
output_per_million: float
|
||||
currency: str
|
||||
# Price for prompt-cache-hit input tokens. None → hits are billed at the
|
||||
# full input price (conservative upper bound for providers that don't
|
||||
# discount, or when the operator hasn't configured the hit price).
|
||||
input_cache_hit_per_million: float | None = None
|
||||
|
||||
|
||||
def _build_pricing_map() -> dict[str, _ModelPricing]:
|
||||
"""Collect per-model prices from ``models[*].pricing`` in config.yaml.
|
||||
|
||||
``ModelConfig`` allows extra fields, so operators can annotate each model
|
||||
with e.g. ``pricing: {currency: CNY, input_per_million: 8,
|
||||
output_per_million: 32, input_cache_hit_per_million: 0.8}`` without any
|
||||
schema change. Entries are keyed by both the config ``name`` and the
|
||||
provider ``model`` id (plus lowercase variants), because
|
||||
``token_usage_by_model`` buckets carry the provider-reported model name.
|
||||
"""
|
||||
try:
|
||||
models = get_app_config().models
|
||||
except Exception: # pragma: no cover - defensive: cost display must not break the console
|
||||
logger.warning("console: failed to load model pricing from config", exc_info=True)
|
||||
return {}
|
||||
|
||||
pricing: dict[str, _ModelPricing] = {}
|
||||
for model_cfg in models or []:
|
||||
raw = getattr(model_cfg, "pricing", None)
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
try:
|
||||
input_price = float(raw.get("input_per_million") or 0)
|
||||
output_price = float(raw.get("output_per_million") or 0)
|
||||
raw_hit_price = raw.get("input_cache_hit_per_million")
|
||||
cache_hit_price = float(raw_hit_price) if raw_hit_price is not None else None
|
||||
except (TypeError, ValueError):
|
||||
logger.warning("console: ignoring malformed pricing on model %s", model_cfg.name)
|
||||
continue
|
||||
if input_price <= 0 and output_price <= 0:
|
||||
continue
|
||||
currency = str(raw.get("currency") or "USD").upper()
|
||||
entry = _ModelPricing(input_price, output_price, currency, cache_hit_price)
|
||||
for key in (model_cfg.name, getattr(model_cfg, "model", None)):
|
||||
if key:
|
||||
pricing.setdefault(key, entry)
|
||||
pricing.setdefault(key.lower(), entry)
|
||||
return pricing
|
||||
|
||||
|
||||
def _pricing_currency(pricing: dict[str, _ModelPricing]) -> str | None:
|
||||
"""Display currency: the first configured entry's (one currency per deployment)."""
|
||||
return next(iter(pricing.values())).currency if pricing else None
|
||||
|
||||
|
||||
def _lookup_pricing(pricing: dict[str, _ModelPricing], model: str | None) -> _ModelPricing | None:
|
||||
if not model:
|
||||
return None
|
||||
return pricing.get(model) or pricing.get(model.lower())
|
||||
|
||||
|
||||
def _token_cost(input_tokens: int, output_tokens: int, price: _ModelPricing, cache_read_tokens: int = 0) -> float:
|
||||
"""Cache-aware spend: cache-hit input tokens are billed at the hit price.
|
||||
|
||||
``cache_read_tokens`` is clamped into ``[0, input_tokens]``; the remainder
|
||||
is billed at the full (cache-miss) input price. Without a configured hit
|
||||
price all input is billed at the miss price.
|
||||
"""
|
||||
cache_read = min(max(int(cache_read_tokens or 0), 0), max(int(input_tokens or 0), 0))
|
||||
uncached = max(int(input_tokens or 0), 0) - cache_read
|
||||
hit_price = price.input_cache_hit_per_million if price.input_cache_hit_per_million is not None else price.input_per_million
|
||||
return (uncached / 1_000_000) * price.input_per_million + (cache_read / 1_000_000) * hit_price + (output_tokens / 1_000_000) * price.output_per_million
|
||||
|
||||
|
||||
def _run_cost(
|
||||
pricing: dict[str, _ModelPricing],
|
||||
*,
|
||||
model_name: str | None,
|
||||
total_input_tokens: int | None,
|
||||
total_output_tokens: int | None,
|
||||
token_usage_by_model: dict | None,
|
||||
) -> float | None:
|
||||
"""Estimate one run's spend, or None when none of its models are priced.
|
||||
|
||||
Prefers the per-model breakdown (accurate for multi-model runs, e.g.
|
||||
subagents on a different model); falls back to run-level totals priced at
|
||||
``model_name`` for legacy rows. Buckets without an input/output split are
|
||||
skipped rather than guessed.
|
||||
"""
|
||||
cost = 0.0
|
||||
priced = False
|
||||
if isinstance(token_usage_by_model, dict):
|
||||
for model, usage in token_usage_by_model.items():
|
||||
if not isinstance(usage, dict):
|
||||
continue
|
||||
price = _lookup_pricing(pricing, model)
|
||||
if price is None:
|
||||
continue
|
||||
input_tokens = int(usage.get("input_tokens") or 0)
|
||||
output_tokens = int(usage.get("output_tokens") or 0)
|
||||
if input_tokens == 0 and output_tokens == 0:
|
||||
continue
|
||||
cost += _token_cost(input_tokens, output_tokens, price, int(usage.get("cache_read_tokens") or 0))
|
||||
priced = True
|
||||
if priced:
|
||||
return cost
|
||||
price = _lookup_pricing(pricing, model_name)
|
||||
if price is None:
|
||||
return None
|
||||
input_tokens = int(total_input_tokens or 0)
|
||||
output_tokens = int(total_output_tokens or 0)
|
||||
if input_tokens == 0 and output_tokens == 0:
|
||||
return None
|
||||
return _token_cost(input_tokens, output_tokens, price)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoints
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@router.get(
|
||||
"/stats",
|
||||
response_model=ConsoleStatsResponse,
|
||||
summary="Console Stats",
|
||||
description="Headline counters (runs, threads, agents, tokens) scoped to the current user.",
|
||||
)
|
||||
@require_permission("runs", "read")
|
||||
async def console_stats(request: Request) -> ConsoleStatsResponse:
|
||||
"""Return the dashboard's headline counters."""
|
||||
sf = _session_factory_or_503()
|
||||
user_id = await get_current_user(request)
|
||||
run_where = (RunRow.user_id == user_id,) if user_id else ()
|
||||
thread_where = (ThreadMetaRow.user_id == user_id,) if user_id else ()
|
||||
|
||||
pricing = _build_pricing_map()
|
||||
|
||||
async with sf() as session:
|
||||
total_runs = await session.scalar(select(func.count()).select_from(RunRow).where(*run_where)) or 0
|
||||
active_runs = await session.scalar(select(func.count()).select_from(RunRow).where(RunRow.status.in_(_ACTIVE_STATUSES), *run_where)) or 0
|
||||
failed_runs = await session.scalar(select(func.count()).select_from(RunRow).where(RunRow.status.in_(_FAILED_STATUSES), *run_where)) or 0
|
||||
total_tokens = await session.scalar(select(func.coalesce(func.sum(RunRow.total_tokens), 0)).where(*run_where)) or 0
|
||||
total_threads = await session.scalar(select(func.count()).select_from(ThreadMetaRow).where(*thread_where)) or 0
|
||||
|
||||
total_cost: float | None = None
|
||||
if pricing:
|
||||
cost_rows = (
|
||||
await session.execute(
|
||||
select(
|
||||
RunRow.model_name,
|
||||
RunRow.total_input_tokens,
|
||||
RunRow.total_output_tokens,
|
||||
RunRow.token_usage_by_model,
|
||||
).where(*run_where)
|
||||
)
|
||||
).all()
|
||||
cost_sum = 0.0
|
||||
for model_name, input_tokens, output_tokens, usage_map in cost_rows:
|
||||
cost = _run_cost(
|
||||
pricing,
|
||||
model_name=model_name,
|
||||
total_input_tokens=input_tokens,
|
||||
total_output_tokens=output_tokens,
|
||||
token_usage_by_model=usage_map,
|
||||
)
|
||||
if cost is not None:
|
||||
cost_sum += cost
|
||||
total_cost = round(cost_sum, 6)
|
||||
|
||||
try:
|
||||
# Filesystem scan; resolves the effective user internally (AuthMiddleware
|
||||
# sets the context for real requests, "default" in no-auth mode).
|
||||
agents = await asyncio.to_thread(list_custom_agents)
|
||||
total_agents = len(agents)
|
||||
except Exception: # pragma: no cover - defensive: stats must not 500 on a bad agents dir
|
||||
logger.warning("console_stats: failed to list custom agents", exc_info=True)
|
||||
total_agents = 0
|
||||
|
||||
return ConsoleStatsResponse(
|
||||
total_runs=total_runs,
|
||||
active_runs=active_runs,
|
||||
failed_runs=failed_runs,
|
||||
total_threads=total_threads,
|
||||
total_agents=total_agents,
|
||||
total_tokens=total_tokens,
|
||||
total_cost=total_cost,
|
||||
currency=_pricing_currency(pricing),
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/runs",
|
||||
response_model=ConsoleRunsResponse,
|
||||
summary="List Runs Across Threads",
|
||||
description="Cross-thread run history for the current user, newest first, joined with thread titles.",
|
||||
)
|
||||
@require_permission("runs", "read")
|
||||
async def console_runs(
|
||||
request: Request,
|
||||
limit: int = Query(default=20, ge=1, le=100),
|
||||
offset: int = Query(default=0, ge=0),
|
||||
status: str | None = Query(default=None, description="Filter by run status (e.g. running, success, error)"),
|
||||
) -> ConsoleRunsResponse:
|
||||
"""Return a page of the user's runs across all threads."""
|
||||
sf = _session_factory_or_503()
|
||||
user_id = await get_current_user(request)
|
||||
|
||||
stmt = select(RunRow, ThreadMetaRow.display_name).join(ThreadMetaRow, ThreadMetaRow.thread_id == RunRow.thread_id, isouter=True).order_by(RunRow.created_at.desc(), RunRow.run_id.desc()).limit(limit + 1).offset(offset)
|
||||
if user_id:
|
||||
stmt = stmt.where(RunRow.user_id == user_id)
|
||||
if status:
|
||||
stmt = stmt.where(RunRow.status == status)
|
||||
|
||||
async with sf() as session:
|
||||
rows = (await session.execute(stmt)).all()
|
||||
|
||||
pricing = _build_pricing_map()
|
||||
has_more = len(rows) > limit
|
||||
now = datetime.now(UTC)
|
||||
items: list[ConsoleRunItem] = []
|
||||
for row, display_name in rows[:limit]:
|
||||
created = _as_utc(row.created_at)
|
||||
updated = _as_utc(row.updated_at)
|
||||
if row.status in _ACTIVE_STATUSES:
|
||||
duration = (now - created).total_seconds() if created else None
|
||||
else:
|
||||
duration = (updated - created).total_seconds() if created and updated else None
|
||||
cost = _run_cost(
|
||||
pricing,
|
||||
model_name=row.model_name,
|
||||
total_input_tokens=row.total_input_tokens,
|
||||
total_output_tokens=row.total_output_tokens,
|
||||
token_usage_by_model=row.token_usage_by_model,
|
||||
)
|
||||
items.append(
|
||||
ConsoleRunItem(
|
||||
run_id=row.run_id,
|
||||
thread_id=row.thread_id,
|
||||
thread_title=display_name,
|
||||
assistant_id=row.assistant_id,
|
||||
status=row.status,
|
||||
model_name=row.model_name,
|
||||
created_at=created,
|
||||
updated_at=updated,
|
||||
duration_seconds=max(duration, 0.0) if duration is not None else None,
|
||||
total_tokens=row.total_tokens or 0,
|
||||
message_count=row.message_count or 0,
|
||||
cost=round(cost, 6) if cost is not None else None,
|
||||
error=row.error[:_ERROR_EXCERPT_CHARS] if row.error else None,
|
||||
)
|
||||
)
|
||||
return ConsoleRunsResponse(runs=items, has_more=has_more)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/usage",
|
||||
response_model=ConsoleUsageResponse,
|
||||
summary="Token Usage Over Time",
|
||||
description="Daily token-usage series (zero-filled) plus per-model breakdown over the requested window.",
|
||||
)
|
||||
@require_permission("runs", "read")
|
||||
async def console_usage(
|
||||
request: Request,
|
||||
days: int = Query(default=14, ge=1, le=90),
|
||||
tz_offset_minutes: int = Query(default=0, ge=-840, le=840, description="Local-time offset from UTC for day bucketing"),
|
||||
) -> ConsoleUsageResponse:
|
||||
"""Aggregate token usage by local day and by model."""
|
||||
sf = _session_factory_or_503()
|
||||
user_id = await get_current_user(request)
|
||||
|
||||
tz_delta = timedelta(minutes=tz_offset_minutes)
|
||||
today_local = (datetime.now(UTC) + tz_delta).date()
|
||||
start_local = today_local - timedelta(days=days - 1)
|
||||
window_start_utc = datetime.combine(start_local, time.min, tzinfo=UTC) - tz_delta
|
||||
|
||||
stmt = select(RunRow).where(RunRow.created_at >= window_start_utc)
|
||||
if user_id:
|
||||
stmt = stmt.where(RunRow.user_id == user_id)
|
||||
|
||||
async with sf() as session:
|
||||
rows = (await session.execute(stmt)).scalars().all()
|
||||
|
||||
day_buckets: dict[str, ConsoleUsageDay] = {}
|
||||
for i in range(days):
|
||||
d = (start_local + timedelta(days=i)).isoformat()
|
||||
day_buckets[d] = ConsoleUsageDay(date=d)
|
||||
|
||||
pricing = _build_pricing_map()
|
||||
by_model: dict[str, ConsoleUsageModelBreakdown] = {}
|
||||
total_tokens = 0
|
||||
total_runs = 0
|
||||
total_cost = 0.0 if pricing else None
|
||||
for row in rows:
|
||||
created = _as_utc(row.created_at)
|
||||
if created is None:
|
||||
continue
|
||||
local_date = ((created + tz_delta).date()).isoformat()
|
||||
bucket = day_buckets.get(local_date)
|
||||
if bucket is None:
|
||||
# Row sits just outside the local window (UTC-window over-fetch); skip.
|
||||
continue
|
||||
run_tokens = row.total_tokens or 0
|
||||
bucket.total_tokens += run_tokens
|
||||
bucket.input_tokens += row.total_input_tokens or 0
|
||||
bucket.output_tokens += row.total_output_tokens or 0
|
||||
bucket.runs += 1
|
||||
total_tokens += run_tokens
|
||||
total_runs += 1
|
||||
|
||||
run_cost = _run_cost(
|
||||
pricing,
|
||||
model_name=row.model_name,
|
||||
total_input_tokens=row.total_input_tokens,
|
||||
total_output_tokens=row.total_output_tokens,
|
||||
token_usage_by_model=row.token_usage_by_model,
|
||||
)
|
||||
if run_cost is not None and total_cost is not None:
|
||||
bucket.cost = round(bucket.cost + run_cost, 6)
|
||||
total_cost = round(total_cost + run_cost, 6)
|
||||
|
||||
usage_map = row.token_usage_by_model or {}
|
||||
if isinstance(usage_map, dict) and usage_map:
|
||||
for model, usage in usage_map.items():
|
||||
entry = by_model.setdefault(model, ConsoleUsageModelBreakdown())
|
||||
entry.runs += 1
|
||||
if not isinstance(usage, dict):
|
||||
continue
|
||||
entry.tokens += int(usage.get("total_tokens", 0) or 0)
|
||||
entry.input_tokens += int(usage.get("input_tokens") or 0)
|
||||
entry.cache_read_tokens += int(usage.get("cache_read_tokens") or 0)
|
||||
price = _lookup_pricing(pricing, model)
|
||||
if price is not None:
|
||||
model_cost = _token_cost(int(usage.get("input_tokens") or 0), int(usage.get("output_tokens") or 0), price, int(usage.get("cache_read_tokens") or 0))
|
||||
entry.cost = round((entry.cost or 0.0) + model_cost, 6)
|
||||
elif row.model_name and run_tokens > 0:
|
||||
# Legacy rows predating token_usage_by_model: fall back to the run's model.
|
||||
entry = by_model.setdefault(row.model_name, ConsoleUsageModelBreakdown())
|
||||
entry.tokens += run_tokens
|
||||
entry.runs += 1
|
||||
if run_cost is not None:
|
||||
entry.cost = round((entry.cost or 0.0) + run_cost, 6)
|
||||
|
||||
return ConsoleUsageResponse(
|
||||
days=list(day_buckets.values()),
|
||||
by_model=by_model,
|
||||
total_tokens=total_tokens,
|
||||
total_runs=total_runs,
|
||||
total_cost=total_cost,
|
||||
currency=_pricing_currency(pricing),
|
||||
)
|
||||
@ -121,6 +121,10 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, *
|
||||
"when_thinking_disabled",
|
||||
"thinking",
|
||||
"supports_vision",
|
||||
# Presentation-only metadata (consumed by the console's cost
|
||||
# display) — must never reach the provider client, which would
|
||||
# forward unknown kwargs into the completion request payload.
|
||||
"pricing",
|
||||
},
|
||||
)
|
||||
# Compute effective when_thinking_enabled by merging in the `thinking` shortcut field.
|
||||
|
||||
@ -317,7 +317,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
per_call_model: str | None = None
|
||||
if isinstance(response_metadata, Mapping):
|
||||
per_call_model = response_metadata.get("model_name") or response_metadata.get("model")
|
||||
self._record_model_usage(per_call_model, input_tk, output_tk, total_tk)
|
||||
self._record_model_usage(per_call_model, input_tk, output_tk, total_tk, self._extract_cache_read(usage_dict))
|
||||
|
||||
self._schedule_progress_flush()
|
||||
|
||||
@ -433,12 +433,18 @@ class RunJournal(BaseCallbackHandler):
|
||||
input_tokens: int,
|
||||
output_tokens: int,
|
||||
total_tokens: int,
|
||||
cache_read_tokens: int = 0,
|
||||
) -> None:
|
||||
"""Add a single LLM call's token usage to the per-model accumulator.
|
||||
|
||||
Missing / empty ``model_name`` collapses into a shared ``"unknown"``
|
||||
bucket so the breakdown stays usable when a provider doesn't surface
|
||||
``response_metadata.model_name``.
|
||||
|
||||
``cache_read_tokens`` (prompt-cache hits, from
|
||||
``usage_metadata.input_token_details.cache_read``) is stored as a
|
||||
sparse bucket key — only written when non-zero — so buckets from
|
||||
providers without cache reporting keep their historical shape.
|
||||
"""
|
||||
if total_tokens <= 0:
|
||||
return
|
||||
@ -449,6 +455,19 @@ class RunJournal(BaseCallbackHandler):
|
||||
bucket["input_tokens"] += int(input_tokens or 0)
|
||||
bucket["output_tokens"] += int(output_tokens or 0)
|
||||
bucket["total_tokens"] += int(total_tokens)
|
||||
if cache_read_tokens > 0:
|
||||
bucket["cache_read_tokens"] = bucket.get("cache_read_tokens", 0) + int(cache_read_tokens)
|
||||
|
||||
@staticmethod
|
||||
def _extract_cache_read(usage_dict: dict) -> int:
|
||||
"""Prompt-cache-hit input tokens from LangChain's normalized usage."""
|
||||
details = usage_dict.get("input_token_details") or {}
|
||||
if not isinstance(details, Mapping):
|
||||
return 0
|
||||
try:
|
||||
return max(int(details.get("cache_read") or 0), 0)
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
# -- Public methods (called by worker) --
|
||||
|
||||
@ -466,6 +485,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
input_tokens: Input token count
|
||||
output_tokens: Output token count
|
||||
total_tokens: Total token count (computed from input+output if 0/missing)
|
||||
cache_read_tokens: Optional prompt-cache-hit input tokens
|
||||
"""
|
||||
if not self._track_tokens:
|
||||
return
|
||||
@ -500,7 +520,8 @@ class RunJournal(BaseCallbackHandler):
|
||||
else:
|
||||
self._lead_agent_tokens += total_tk
|
||||
|
||||
self._record_model_usage(record.get("model_name"), input_tk, output_tk, total_tk)
|
||||
cache_read_tk = record.get("cache_read_tokens", 0) or 0
|
||||
self._record_model_usage(record.get("model_name"), input_tk, output_tk, total_tk, int(cache_read_tk))
|
||||
|
||||
self._schedule_progress_flush()
|
||||
|
||||
|
||||
@ -47,6 +47,14 @@ class SubagentTokenCollector(BaseCallbackHandler):
|
||||
total_tk = input_tk + output_tk
|
||||
if total_tk <= 0:
|
||||
continue
|
||||
# Prompt-cache hits (needed for cache-aware cost accounting)
|
||||
details = usage_dict.get("input_token_details") or {}
|
||||
cache_read_tk = 0
|
||||
if isinstance(details, Mapping):
|
||||
try:
|
||||
cache_read_tk = max(int(details.get("cache_read") or 0), 0)
|
||||
except (TypeError, ValueError):
|
||||
cache_read_tk = 0
|
||||
# Capture the model that actually produced this response so the
|
||||
# parent journal can bucket tokens by real model rather than the
|
||||
# lead agent's resolved model
|
||||
@ -55,16 +63,19 @@ class SubagentTokenCollector(BaseCallbackHandler):
|
||||
if isinstance(response_metadata, Mapping):
|
||||
model_name = response_metadata.get("model_name") or response_metadata.get("model")
|
||||
self._counted_run_ids.add(rid)
|
||||
self._records.append(
|
||||
{
|
||||
"source_run_id": rid,
|
||||
"caller": self.caller,
|
||||
"model_name": model_name,
|
||||
"input_tokens": input_tk,
|
||||
"output_tokens": output_tk,
|
||||
"total_tokens": total_tk,
|
||||
}
|
||||
)
|
||||
record: dict[str, int | str | None] = {
|
||||
"source_run_id": rid,
|
||||
"caller": self.caller,
|
||||
"model_name": model_name,
|
||||
"input_tokens": input_tk,
|
||||
"output_tokens": output_tk,
|
||||
"total_tokens": total_tk,
|
||||
}
|
||||
# Sparse, matching the journal's per-model buckets: the key is
|
||||
# only present when the provider actually reported cache hits.
|
||||
if cache_read_tk > 0:
|
||||
record["cache_read_tokens"] = cache_read_tk
|
||||
self._records.append(record)
|
||||
return
|
||||
|
||||
def snapshot_records(self) -> list[dict[str, int | str | None]]:
|
||||
|
||||
289
backend/tests/test_console_router.py
Normal file
289
backend/tests/test_console_router.py
Normal file
@ -0,0 +1,289 @@
|
||||
"""Tests for the console router (cross-thread observability endpoints).
|
||||
|
||||
Covers:
|
||||
1. /api/console/stats — headline counters
|
||||
2. /api/console/runs — cross-thread listing, thread-title join, pagination, status filter
|
||||
3. /api/console/usage — daily zero-filled buckets + per-model breakdown (incl. legacy fallback)
|
||||
4. user scoping — rows filtered when the request resolves to a user
|
||||
5. 503 when no SQL session factory is available (memory backend)
|
||||
|
||||
Uses a real temp-file SQLite database (NullPool, so seeding in one event loop
|
||||
and serving TestClient requests in another never share a connection).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from _router_auth_helpers import make_authed_test_app
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||
from sqlalchemy.pool import NullPool
|
||||
|
||||
from app.gateway.routers import console
|
||||
from deerflow.persistence.base import Base
|
||||
from deerflow.persistence.run.model import RunRow
|
||||
from deerflow.persistence.thread_meta.model import ThreadMetaRow
|
||||
|
||||
NOW = datetime.now(UTC)
|
||||
|
||||
|
||||
def _seed_rows() -> tuple[list[ThreadMetaRow], list[RunRow]]:
|
||||
threads = [
|
||||
ThreadMetaRow(thread_id="t1", user_id="user-a", display_name="调研鹿角再生"),
|
||||
ThreadMetaRow(thread_id="t2", user_id="user-a", display_name="Card assistant chat"),
|
||||
]
|
||||
runs = [
|
||||
RunRow(
|
||||
run_id="r1",
|
||||
thread_id="t1",
|
||||
user_id="user-a",
|
||||
status="success",
|
||||
model_name="minimax-m2",
|
||||
total_tokens=1200,
|
||||
total_input_tokens=800,
|
||||
total_output_tokens=400,
|
||||
token_usage_by_model={"minimax-m2": {"input_tokens": 800, "output_tokens": 400, "total_tokens": 1200, "cache_read_tokens": 500}},
|
||||
message_count=4,
|
||||
created_at=NOW - timedelta(hours=1),
|
||||
updated_at=NOW - timedelta(hours=1) + timedelta(seconds=30),
|
||||
),
|
||||
RunRow(
|
||||
run_id="r2",
|
||||
thread_id="t1",
|
||||
user_id="user-a",
|
||||
status="running",
|
||||
model_name="minimax-m2",
|
||||
total_tokens=300,
|
||||
total_input_tokens=200,
|
||||
total_output_tokens=100,
|
||||
token_usage_by_model={"minimax-m2": {"input_tokens": 200, "output_tokens": 100, "total_tokens": 300}},
|
||||
message_count=1,
|
||||
created_at=NOW - timedelta(seconds=60),
|
||||
updated_at=NOW - timedelta(seconds=1),
|
||||
),
|
||||
RunRow(
|
||||
run_id="r3",
|
||||
thread_id="t2",
|
||||
user_id="user-a",
|
||||
status="error",
|
||||
model_name="gpt-x",
|
||||
error="Boom: provider exploded",
|
||||
total_tokens=50,
|
||||
total_input_tokens=50,
|
||||
total_output_tokens=0,
|
||||
token_usage_by_model={}, # legacy row → model_name fallback path
|
||||
message_count=2,
|
||||
created_at=NOW - timedelta(days=1),
|
||||
updated_at=NOW - timedelta(days=1) + timedelta(seconds=5),
|
||||
),
|
||||
RunRow(
|
||||
run_id="r4",
|
||||
thread_id="t2",
|
||||
user_id="user-a",
|
||||
status="success",
|
||||
model_name="minimax-m2",
|
||||
total_tokens=999,
|
||||
token_usage_by_model={"minimax-m2": {"input_tokens": 600, "output_tokens": 399, "total_tokens": 999}},
|
||||
created_at=NOW - timedelta(days=40), # outside the default usage window
|
||||
updated_at=NOW - timedelta(days=40) + timedelta(seconds=10),
|
||||
),
|
||||
RunRow(
|
||||
run_id="r5",
|
||||
thread_id="t3", # no threads_meta row → exercises the outer join
|
||||
user_id="user-b",
|
||||
status="success",
|
||||
model_name="qwen",
|
||||
total_tokens=70,
|
||||
total_input_tokens=40,
|
||||
total_output_tokens=30,
|
||||
token_usage_by_model={"qwen": {"input_tokens": 40, "output_tokens": 30, "total_tokens": 70}},
|
||||
created_at=NOW - timedelta(hours=2),
|
||||
updated_at=NOW - timedelta(hours=2) + timedelta(seconds=8),
|
||||
),
|
||||
]
|
||||
return threads, runs
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def session_factory(tmp_path):
|
||||
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'console.db'}", poolclass=NullPool)
|
||||
sf = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
async def _setup() -> None:
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
threads, runs = _seed_rows()
|
||||
async with sf() as session:
|
||||
session.add_all(threads)
|
||||
session.add_all(runs)
|
||||
await session.commit()
|
||||
|
||||
asyncio.run(_setup())
|
||||
yield sf
|
||||
asyncio.run(engine.dispose())
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client(session_factory, monkeypatch):
|
||||
monkeypatch.setattr(console, "get_session_factory", lambda: session_factory)
|
||||
monkeypatch.setattr(console, "get_current_user", AsyncMock(return_value=None))
|
||||
monkeypatch.setattr(console, "list_custom_agents", lambda: [object(), object()])
|
||||
# No pricing configured by default; TestPricing patches its own config.
|
||||
monkeypatch.setattr(console, "get_app_config", lambda: SimpleNamespace(models=[]))
|
||||
app = make_authed_test_app()
|
||||
app.include_router(console.router)
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
class TestConsoleStats:
|
||||
def test_headline_counters(self, client):
|
||||
resp = client.get("/api/console/stats")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total_runs"] == 5
|
||||
assert data["active_runs"] == 1 # r2 running
|
||||
assert data["failed_runs"] == 1 # r3 error
|
||||
assert data["total_threads"] == 2
|
||||
assert data["total_agents"] == 2
|
||||
assert data["total_tokens"] == 1200 + 300 + 50 + 999 + 70
|
||||
|
||||
|
||||
class TestConsoleRuns:
|
||||
def test_listing_orders_paginates_and_joins_titles(self, client):
|
||||
resp = client.get("/api/console/runs", params={"limit": 3})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["has_more"] is True
|
||||
ids = [r["run_id"] for r in data["runs"]]
|
||||
assert ids == ["r2", "r1", "r5"] # newest first
|
||||
by_id = {r["run_id"]: r for r in data["runs"]}
|
||||
assert by_id["r2"]["thread_title"] == "调研鹿角再生"
|
||||
assert by_id["r5"]["thread_title"] is None # t3 has no threads_meta row
|
||||
# Terminal run: duration from created→updated; active run: live elapsed > 0.
|
||||
assert by_id["r1"]["duration_seconds"] == pytest.approx(30.0, abs=1.0)
|
||||
assert by_id["r2"]["duration_seconds"] > 0
|
||||
|
||||
def test_offset_pagination(self, client):
|
||||
resp = client.get("/api/console/runs", params={"limit": 3, "offset": 3})
|
||||
data = resp.json()
|
||||
assert [r["run_id"] for r in data["runs"]] == ["r3", "r4"]
|
||||
assert data["has_more"] is False
|
||||
|
||||
def test_status_filter(self, client):
|
||||
resp = client.get("/api/console/runs", params={"status": "error"})
|
||||
data = resp.json()
|
||||
assert [r["run_id"] for r in data["runs"]] == ["r3"]
|
||||
assert data["runs"][0]["error"].startswith("Boom")
|
||||
|
||||
|
||||
class TestConsoleUsage:
|
||||
def test_daily_buckets_and_model_breakdown(self, client):
|
||||
resp = client.get("/api/console/usage", params={"days": 14})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["days"]) == 14
|
||||
# r4 (40 days old) excluded from the window
|
||||
assert data["total_runs"] == 4
|
||||
assert data["total_tokens"] == 1200 + 300 + 50 + 70
|
||||
# Zero-filled series sums to the window total
|
||||
assert sum(d["total_tokens"] for d in data["days"]) == data["total_tokens"]
|
||||
yesterday = (NOW - timedelta(days=1)).date().isoformat()
|
||||
day_map = {d["date"]: d for d in data["days"]}
|
||||
assert day_map[yesterday]["total_tokens"] == 50
|
||||
assert data["by_model"]["minimax-m2"]["tokens"] == 1500
|
||||
assert data["by_model"]["qwen"]["tokens"] == 70
|
||||
# Legacy row (empty token_usage_by_model) falls back to model_name
|
||||
assert data["by_model"]["gpt-x"]["tokens"] == 50
|
||||
|
||||
def test_window_excludes_old_rows_but_stats_include_them(self, client):
|
||||
usage = client.get("/api/console/usage", params={"days": 7}).json()
|
||||
assert all(r != 999 for d in usage["days"] for r in [d["total_tokens"]])
|
||||
stats = client.get("/api/console/stats").json()
|
||||
assert stats["total_tokens"] >= 999
|
||||
|
||||
|
||||
def _priced_config(*, cache_hit_price: float | None = 0.8):
|
||||
pricing = {"currency": "CNY", "input_per_million": 8, "output_per_million": 32}
|
||||
if cache_hit_price is not None:
|
||||
pricing["input_cache_hit_per_million"] = cache_hit_price
|
||||
return SimpleNamespace(
|
||||
models=[
|
||||
SimpleNamespace(name="minimax-m2", model="MiniMax-M2", pricing=pricing),
|
||||
SimpleNamespace(name="gpt-x", model="gpt-x-1", pricing=None), # explicitly unpriced
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
# Expected per-run costs at 8 (miss) / 0.8 (hit) / 32 (output) per million:
|
||||
# r1: 500 of 800 input tokens were cache hits → 300*8 + 500*0.8 + 400*32 (µ¥)
|
||||
_R1_COST_CACHED = 300 * 8e-6 + 500 * 0.8e-6 + 400 * 32e-6 # 0.0156
|
||||
_R1_COST_UNCACHED = 800 * 8e-6 + 400 * 32e-6 # 0.0192 (no hit price configured)
|
||||
_R2_COST = 200 * 8e-6 + 100 * 32e-6 # 0.0048 (no cache hits recorded)
|
||||
_R4_COST = 600 * 8e-6 + 399 * 32e-6 # 0.017568
|
||||
|
||||
|
||||
class TestPricing:
|
||||
def test_costs_use_cache_hit_price(self, client, monkeypatch):
|
||||
monkeypatch.setattr(console, "get_app_config", lambda: _priced_config())
|
||||
stats = client.get("/api/console/stats").json()
|
||||
assert stats["currency"] == "CNY"
|
||||
# r3 (gpt-x) and r5 (qwen) are unpriced and excluded.
|
||||
assert stats["total_cost"] == pytest.approx(_R1_COST_CACHED + _R2_COST + _R4_COST)
|
||||
|
||||
usage = client.get("/api/console/usage").json()
|
||||
assert usage["currency"] == "CNY"
|
||||
assert usage["total_cost"] == pytest.approx(_R1_COST_CACHED + _R2_COST) # r4 outside window
|
||||
assert usage["by_model"]["minimax-m2"]["cost"] == pytest.approx(_R1_COST_CACHED + _R2_COST)
|
||||
assert usage["by_model"]["minimax-m2"]["input_tokens"] == 1000
|
||||
assert usage["by_model"]["minimax-m2"]["cache_read_tokens"] == 500
|
||||
assert usage["by_model"]["gpt-x"]["cost"] is None
|
||||
assert sum(d["cost"] for d in usage["days"]) == pytest.approx(_R1_COST_CACHED + _R2_COST)
|
||||
|
||||
runs = client.get("/api/console/runs", params={"limit": 50}).json()
|
||||
by_id = {r["run_id"]: r for r in runs["runs"]}
|
||||
assert by_id["r1"]["cost"] == pytest.approx(_R1_COST_CACHED)
|
||||
assert by_id["r3"]["cost"] is None # unpriced model
|
||||
|
||||
def test_cache_hits_billed_at_miss_price_without_hit_price(self, client, monkeypatch):
|
||||
"""No input_cache_hit_per_million configured → conservative upper bound."""
|
||||
monkeypatch.setattr(console, "get_app_config", lambda: _priced_config(cache_hit_price=None))
|
||||
runs = client.get("/api/console/runs", params={"limit": 50}).json()
|
||||
by_id = {r["run_id"]: r for r in runs["runs"]}
|
||||
assert by_id["r1"]["cost"] == pytest.approx(_R1_COST_UNCACHED)
|
||||
|
||||
def test_costs_null_without_pricing(self, client):
|
||||
stats = client.get("/api/console/stats").json()
|
||||
assert stats["total_cost"] is None
|
||||
assert stats["currency"] is None
|
||||
usage = client.get("/api/console/usage").json()
|
||||
assert usage["total_cost"] is None
|
||||
runs = client.get("/api/console/runs").json()
|
||||
assert all(r["cost"] is None for r in runs["runs"])
|
||||
|
||||
|
||||
class TestUserScoping:
|
||||
def test_rows_filtered_by_resolved_user(self, client, monkeypatch):
|
||||
monkeypatch.setattr(console, "get_current_user", AsyncMock(return_value="user-a"))
|
||||
stats = client.get("/api/console/stats").json()
|
||||
assert stats["total_runs"] == 4 # r5 (user-b) excluded
|
||||
assert stats["total_tokens"] == 1200 + 300 + 50 + 999
|
||||
runs = client.get("/api/console/runs", params={"limit": 50}).json()
|
||||
assert all(r["run_id"] != "r5" for r in runs["runs"])
|
||||
usage = client.get("/api/console/usage").json()
|
||||
assert "qwen" not in usage["by_model"]
|
||||
|
||||
|
||||
class TestNoSqlBackend:
|
||||
def test_503_when_memory_backend(self, session_factory, monkeypatch):
|
||||
monkeypatch.setattr(console, "get_session_factory", lambda: None)
|
||||
monkeypatch.setattr(console, "get_current_user", AsyncMock(return_value=None))
|
||||
app = make_authed_test_app()
|
||||
app.include_router(console.router)
|
||||
c = TestClient(app)
|
||||
for path in ("/api/console/stats", "/api/console/runs", "/api/console/usage"):
|
||||
resp = c.get(path)
|
||||
assert resp.status_code == 503
|
||||
assert "SQL database backend" in resp.json()["detail"]
|
||||
@ -103,6 +103,23 @@ def test_raises_when_model_not_found(monkeypatch):
|
||||
factory_module.create_chat_model(name="ghost-model")
|
||||
|
||||
|
||||
def test_pricing_metadata_never_reaches_the_provider_client(monkeypatch):
|
||||
"""`models[*].pricing` is console-only metadata (issue: ChatOpenAI forwards
|
||||
unknown kwargs into the completion request payload, so an un-stripped
|
||||
`pricing` block breaks every live LLM call with
|
||||
``Completions.create() got an unexpected keyword argument 'pricing'``)."""
|
||||
model = _make_model("priced")
|
||||
# ModelConfig is extra="allow" — pricing rides along as an extra field.
|
||||
model.pricing = {"currency": "CNY", "input_per_million": 8, "output_per_million": 32, "input_cache_hit_per_million": 0.8}
|
||||
cfg = _make_app_config([model])
|
||||
_patch_factory(monkeypatch, cfg)
|
||||
|
||||
FakeChatModel.captured_kwargs = {}
|
||||
factory_module.create_chat_model(name="priced")
|
||||
|
||||
assert "pricing" not in FakeChatModel.captured_kwargs
|
||||
|
||||
|
||||
def test_appends_all_tracing_callbacks(monkeypatch):
|
||||
cfg = _make_app_config([_make_model("alpha")])
|
||||
_patch_factory(monkeypatch, cfg)
|
||||
|
||||
@ -429,3 +429,139 @@ async def test_include_active_picks_up_running_progress_snapshot(tmp_path):
|
||||
assert active["total_tokens"] == 70
|
||||
finally:
|
||||
await _close_sql_engine()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompt-cache-hit accounting (powers cache-aware cost estimation in
|
||||
# /api/console): cache_read_tokens is a *sparse* bucket key — present only
|
||||
# when a provider reported cache hits — so pre-existing bucket shapes and
|
||||
# exact-equality assertions above stay valid.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestJournalCacheRead:
|
||||
def test_cache_read_accumulates_as_sparse_key(self) -> None:
|
||||
j = _journal()
|
||||
j.on_llm_end(
|
||||
_make_llm_response(
|
||||
usage={"input_tokens": 100, "output_tokens": 10, "total_tokens": 110, "input_token_details": {"cache_read": 80}},
|
||||
model_name="m",
|
||||
),
|
||||
run_id=uuid4(),
|
||||
parent_run_id=None,
|
||||
tags=["lead_agent"],
|
||||
)
|
||||
# Second call without cache hits still accumulates into the same bucket.
|
||||
j.on_llm_end(
|
||||
_make_llm_response(usage={"input_tokens": 50, "output_tokens": 5, "total_tokens": 55}, model_name="m"),
|
||||
run_id=uuid4(),
|
||||
parent_run_id=None,
|
||||
tags=["lead_agent"],
|
||||
)
|
||||
data = j.get_completion_data()
|
||||
assert data["token_usage_by_model"]["m"] == {
|
||||
"input_tokens": 150,
|
||||
"output_tokens": 15,
|
||||
"total_tokens": 165,
|
||||
"cache_read_tokens": 80,
|
||||
}
|
||||
|
||||
def test_bucket_without_cache_hits_keeps_legacy_shape(self) -> None:
|
||||
j = _journal()
|
||||
j.on_llm_end(
|
||||
_make_llm_response(usage={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, model_name="m"),
|
||||
run_id=uuid4(),
|
||||
parent_run_id=None,
|
||||
tags=["lead_agent"],
|
||||
)
|
||||
assert j.get_completion_data()["token_usage_by_model"]["m"] == {
|
||||
"input_tokens": 10,
|
||||
"output_tokens": 5,
|
||||
"total_tokens": 15,
|
||||
}
|
||||
|
||||
def test_external_records_carry_cache_read(self) -> None:
|
||||
j = _journal()
|
||||
j.record_external_llm_usage_records(
|
||||
[
|
||||
{
|
||||
"source_run_id": "sub-1",
|
||||
"caller": "subagent:general-purpose",
|
||||
"model_name": "sub-m",
|
||||
"input_tokens": 40,
|
||||
"output_tokens": 10,
|
||||
"total_tokens": 50,
|
||||
"cache_read_tokens": 25,
|
||||
},
|
||||
],
|
||||
)
|
||||
assert j.get_completion_data()["token_usage_by_model"]["sub-m"]["cache_read_tokens"] == 25
|
||||
|
||||
def test_deepseek_raw_usage_normalizes_to_cache_read(self) -> None:
|
||||
"""Pin the DeepSeek chat-completions shape end-to-end: the raw
|
||||
``prompt_tokens_details.cached_tokens`` field is what langchain-openai's
|
||||
``_create_usage_metadata`` normalizes into
|
||||
``input_token_details.cache_read`` (DeepSeek's top-level
|
||||
``prompt_cache_hit/miss_tokens`` are redundant aliases LangChain does
|
||||
not read), and the journal captures it. The derived cache-miss count
|
||||
(input − cache_read) must equal DeepSeek's own
|
||||
``prompt_cache_miss_tokens``, which is what cache-aware pricing bills
|
||||
at the full input price."""
|
||||
from langchain_openai.chat_models.base import _create_usage_metadata
|
||||
|
||||
raw = {
|
||||
"prompt_tokens": 106,
|
||||
"completion_tokens": 112,
|
||||
"total_tokens": 218,
|
||||
"prompt_tokens_details": {"cached_tokens": 64},
|
||||
"prompt_cache_hit_tokens": 64,
|
||||
"prompt_cache_miss_tokens": 42,
|
||||
}
|
||||
usage = _create_usage_metadata(raw)
|
||||
j = _journal()
|
||||
j.on_llm_end(
|
||||
_make_llm_response(usage=dict(usage), model_name="deepseek-chat"),
|
||||
run_id=uuid4(),
|
||||
parent_run_id=None,
|
||||
tags=["lead_agent"],
|
||||
)
|
||||
bucket = j.get_completion_data()["token_usage_by_model"]["deepseek-chat"]
|
||||
assert bucket == {
|
||||
"input_tokens": 106,
|
||||
"output_tokens": 112,
|
||||
"total_tokens": 218,
|
||||
"cache_read_tokens": 64,
|
||||
}
|
||||
assert bucket["input_tokens"] - bucket["cache_read_tokens"] == raw["prompt_cache_miss_tokens"]
|
||||
|
||||
def test_collector_extracts_cache_read_from_usage_metadata(self) -> None:
|
||||
from deerflow.subagents.token_collector import SubagentTokenCollector
|
||||
|
||||
collector = SubagentTokenCollector("subagent:general-purpose")
|
||||
collector.on_llm_end(
|
||||
_make_llm_response(
|
||||
usage={"input_tokens": 30, "output_tokens": 6, "total_tokens": 36, "input_token_details": {"cache_read": 20}},
|
||||
model_name="sub-m",
|
||||
),
|
||||
run_id=uuid4(),
|
||||
)
|
||||
records = collector.snapshot_records()
|
||||
assert len(records) == 1
|
||||
assert records[0]["cache_read_tokens"] == 20
|
||||
|
||||
def test_collector_omits_cache_read_key_when_no_cache_hits(self) -> None:
|
||||
from deerflow.subagents.token_collector import SubagentTokenCollector
|
||||
|
||||
collector = SubagentTokenCollector("subagent:general-purpose")
|
||||
collector.on_llm_end(
|
||||
_make_llm_response(
|
||||
usage={"input_tokens": 30, "output_tokens": 6, "total_tokens": 36},
|
||||
model_name="sub-m",
|
||||
),
|
||||
run_id=uuid4(),
|
||||
)
|
||||
records = collector.snapshot_records()
|
||||
assert len(records) == 1
|
||||
# Sparse record shape: no explicit 0 when the provider reported no
|
||||
# cache hits (record_external_llm_usage_records treats absent as 0).
|
||||
assert "cache_read_tokens" not in records[0]
|
||||
|
||||
@ -70,6 +70,17 @@ max_recursion_limit: 1000
|
||||
# Models Configuration
|
||||
# ============================================================================
|
||||
# Configure available LLM models for the agent to use
|
||||
#
|
||||
# Optional per-model pricing (powers the real-cost display on the workspace
|
||||
# console). Add a `pricing` block to any model entry; use ONE currency across
|
||||
# all models. Prices are per one million tokens.
|
||||
#
|
||||
# pricing:
|
||||
# currency: CNY # ISO code shown in the console (CNY, USD, ...)
|
||||
# input_per_million: 8.0 # price per 1M input tokens (cache miss)
|
||||
# output_per_million: 32.0 # price per 1M output tokens
|
||||
# input_cache_hit_per_million: 0.8 # price per 1M cache-hit input tokens
|
||||
# # (optional; omit → hits billed at miss price)
|
||||
|
||||
models:
|
||||
# Example: Volcengine (Doubao) model
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user