diff --git a/app/api/chat.py b/app/api/chat.py index a83fd61..84a2d56 100644 --- a/app/api/chat.py +++ b/app/api/chat.py @@ -16,7 +16,8 @@ from app.dependencies import ( voice_for_route, ) from app.core.memory_extractor import maybe_schedule_extraction -from app.quota import enforce_quota, record_usage, QuotaExceededError +from app.core.costs import start_meter, persist_costs +from app.quota import enforce_quota, record_usage, QuotaExceededError, record_web_search from app.runtime_config import runtime_settings from app.safety.emergency import ( record_manual_emergency, @@ -97,6 +98,7 @@ async def chat( # Explizit angeforderte Stimme gewinnt; sonst folgt sie der Routensprache. voice = payload.voice or voice_for_route(route.tts_provider, route.language, route.voice_gender) + start_meter() # Kosten dieses Turns sammeln try: trace, audio = await orchestrator.chat_text( payload.text, @@ -110,6 +112,8 @@ async def chat( raise HTTPException(status_code=502, detail=str(exc)) record_usage(user, store, len(payload.text) + len(trace.semantic_response or "")) + record_web_search(user, store, trace.web_searches) + persist_costs(user, store) # Turn persistieren (User-Eingabe + semantische Antwort) fuer das Gedaechtnis. if session_id: diff --git a/app/api/ws.py b/app/api/ws.py index 20a8720..9aa3338 100644 --- a/app/api/ws.py +++ b/app/api/ws.py @@ -34,6 +34,7 @@ from app.audio.vad import EnergyVAD from app.core.memory_extractor import maybe_schedule_extraction from app.quota import (enforce_quota, record_usage, QuotaExceededError, web_search_within_budget, record_web_search) +from app.core.costs import start_meter, persist_costs from app.pipeline.search_backstop import should_force_search # Freundliche Hinweise (Senioren), wenn die Web-Suche bei einer offensichtlich @@ -106,6 +107,7 @@ async def _run_turn( effective_voice: str | None = None, ): """Faehrt einen Antwort-Turn und streamt die Events an den Client.""" + start_meter() # Kosten dieses Turns sammeln (LLM/Sonar via OpenRouter usage.cost) conversation = ( store.get_recent_messages(session_id, settings.history_max_messages) if session_id @@ -199,6 +201,7 @@ async def _run_turn( record_usage(user, store, len(text) + len(trace.semantic_response or "")) record_web_search(user, store, trace.web_searches) + persist_costs(user, store) if session_id: store.append_message(session_id, user.id, "user", text) diff --git a/app/core/costs.py b/app/core/costs.py new file mode 100644 index 0000000..b457e08 --- /dev/null +++ b/app/core/costs.py @@ -0,0 +1,80 @@ +"""Request-weiter Kosten-Zähler (USD) über ContextVars. + +Kosten entstehen in vielen unverbundenen Schichten (LLM-Provider, Tool, TTS, +Koreferenz-Vorstufe, Notruf). Statt einen Callback durch alle Signaturen zu +fädeln, hält ein contextvar-basierter `CostMeter` die Kosten eines Requests +zusammen: jede Stelle ruft `add_cost(betrag_usd, kategorie)`. Die API-Schicht +öffnet pro Turn `start_meter()` und persistiert danach `current_meter()`. + +Multitool-ready: neue Kostenarten = neue (freie) Kategorie-Strings. +Siehe Docs/kosten-erfassung.md. +""" +import contextvars +import json + +_meter: contextvars.ContextVar = contextvars.ContextVar("cost_meter", default=None) + + +class CostMeter: + __slots__ = ("total", "by_cat") + + def __init__(self): + self.total: float = 0.0 + self.by_cat: dict[str, float] = {} + + def add(self, usd, category: str) -> None: + try: + usd = float(usd or 0.0) + except (TypeError, ValueError): + return + if usd <= 0: + return + self.total += usd + self.by_cat[category] = self.by_cat.get(category, 0.0) + usd + + +def start_meter() -> CostMeter: + """Setzt einen frischen Zähler für den aktuellen (Request-)Kontext.""" + meter = CostMeter() + _meter.set(meter) + return meter + + +def current_meter() -> "CostMeter | None": + return _meter.get() + + +def add_cost(usd, category: str) -> None: + """Bucht Kosten auf den aktuellen Zähler (no-op, wenn keiner aktiv ist).""" + meter = _meter.get() + if meter is not None: + meter.add(usd, category) + + +def record_openrouter_cost(usage, category: str) -> None: + """Liest `cost` aus einem OpenRouter-`usage`-Objekt und bucht es (USD).""" + if isinstance(usage, dict): + add_cost(usage.get("cost"), category) + + +def persist_costs(user, store) -> None: + """Schreibt die im aktuellen Meter gesammelten Kosten pro Kategorie in den Store.""" + meter = _meter.get() + if meter is None or user is None: + return + for category, amount in meter.by_cat.items(): + store.add_cost_usage(user.id, category, amount) + + +def record_sse_usage(line: str, category: str) -> None: + """Bucht die Kosten aus einer OpenRouter-SSE-Zeile (finaler `usage`-Chunk).""" + if not line.startswith("data:"): + return + body = line[len("data:"):].strip() + if not body or body == "[DONE]": + return + try: + obj = json.loads(body) + except ValueError: + return + record_openrouter_cost(obj.get("usage"), category) diff --git a/app/pipeline/decontextualizer.py b/app/pipeline/decontextualizer.py index e507f1f..948e999 100644 --- a/app/pipeline/decontextualizer.py +++ b/app/pipeline/decontextualizer.py @@ -9,6 +9,8 @@ import logging import httpx +from app.core.costs import record_openrouter_cost + logger = logging.getLogger(__name__) ENDPOINT = "https://openrouter.ai/api/v1/chat/completions" @@ -66,7 +68,7 @@ class Decontextualizer: return text user = (f"Conversation:\n{self._render(history)}\n\n" f"Latest message: {text}\n\nRewritten self-contained message:") - payload = {"model": self.model, "temperature": 0.0, + payload = {"model": self.model, "temperature": 0.0, "usage": {"include": True}, "messages": [{"role": "system", "content": _SYSTEM}, {"role": "user", "content": user}]} headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"} @@ -74,7 +76,9 @@ class Decontextualizer: async with httpx.AsyncClient(timeout=httpx.Timeout(self.timeout)) as client: resp = await client.post(ENDPOINT, json=payload, headers=headers) resp.raise_for_status() - out = (resp.json()["choices"][0]["message"]["content"] or "").strip() + data = resp.json() + record_openrouter_cost(data.get("usage"), "llm") + out = (data["choices"][0]["message"]["content"] or "").strip() except Exception as exc: # noqa: BLE001 — best effort, bei Fehler Original behalten logger.warning("Decontextualizer fehlgeschlagen: %s", exc) return text diff --git a/app/providers/llm/openrouter.py b/app/providers/llm/openrouter.py index 722f788..a20eae5 100644 --- a/app/providers/llm/openrouter.py +++ b/app/providers/llm/openrouter.py @@ -5,6 +5,7 @@ from collections.abc import AsyncIterator import httpx from app.core import clock +from app.core.costs import record_openrouter_cost, record_sse_usage from app.providers.llm.base import ( LLMProvider, lang_instruction, @@ -122,6 +123,7 @@ class OpenRouterLLMProvider(LLMProvider): payload = { "model": self.model, "messages": self._build_messages(text, history, language=language), + "usage": {"include": True}, } timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0) last_error: Exception | None = None @@ -158,6 +160,7 @@ class OpenRouterLLMProvider(LLMProvider): raise RuntimeError(f"OpenRouter LLM transport error: {exc}") from exc data = response.json() + record_openrouter_cost(data.get("usage"), "llm") try: content = data["choices"][0]["message"]["content"] except (KeyError, IndexError, TypeError) as exc: @@ -182,6 +185,7 @@ class OpenRouterLLMProvider(LLMProvider): "model": self.model, "messages": self._build_messages(text, history, language=language), "stream": True, + "usage": {"include": True}, } timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0) last_error: Exception | None = None @@ -213,6 +217,7 @@ class OpenRouterLLMProvider(LLMProvider): f"{body.decode(errors='replace')}" ) async for line in response.aiter_lines(): + record_sse_usage(line, "llm") delta = sse_delta(line) if delta: yield delta diff --git a/app/providers/llm/tool_calling.py b/app/providers/llm/tool_calling.py index 97e8667..8340f24 100644 --- a/app/providers/llm/tool_calling.py +++ b/app/providers/llm/tool_calling.py @@ -17,6 +17,7 @@ from collections.abc import AsyncIterator import httpx from app.core import clock +from app.core.costs import record_openrouter_cost from app.metrics import metrics from app.pipeline.search_backstop import should_force_search from app.providers.llm.base import LLMProvider, lang_instruction, with_lang_reminder @@ -180,7 +181,8 @@ class ToolCallingLLM(LLMProvider): async def _chat(self, messages: list[dict], allow_tools: bool = True) -> dict: """Ein Modell-Aufruf; liefert die Assistant-Message (mit/ohne tool_calls).""" - payload = {"model": self.model, "messages": messages, "temperature": self.temperature} + payload = {"model": self.model, "messages": messages, "temperature": self.temperature, + "usage": {"include": True}} if allow_tools and not self._tools_unsupported: payload["tools"] = [WEB_SEARCH_TOOL] payload["tool_choice"] = "auto" @@ -215,6 +217,7 @@ class ToolCallingLLM(LLMProvider): await asyncio.sleep(min(2.0 * attempt, 30.0)) continue raise RuntimeError(f"Ungültige OpenRouter-Antwort: {resp.text[:300]}") + record_openrouter_cost(data.get("usage"), "llm") return data["choices"][0]["message"] raise last_error or RuntimeError("ToolCallingLLM: alle Versuche fehlgeschlagen") @@ -338,8 +341,8 @@ class ToolCallingLLM(LLMProvider): async def _stream_chat(self, messages: list[dict], allow_tools: bool, acc: _StreamAcc) -> AsyncIterator[str]: """Ein gestreamter Modell-Call. Yieldet Text-Deltas; füllt acc (Text + tool_calls).""" - payload = {"model": self.model, "messages": messages, - "temperature": self.temperature, "stream": True} + payload = {"model": self.model, "messages": messages, "temperature": self.temperature, + "stream": True, "usage": {"include": True}} if allow_tools and not self._tools_unsupported: payload["tools"] = [WEB_SEARCH_TOOL] payload["tool_choice"] = "auto" @@ -362,9 +365,15 @@ class ToolCallingLLM(LLMProvider): if not data or data == "[DONE]": continue try: - delta = json.loads(data)["choices"][0]["delta"] - except (ValueError, KeyError, IndexError, TypeError): + obj = json.loads(data) + except ValueError: continue + if obj.get("usage"): # finaler Chunk -> Ist-Kosten + record_openrouter_cost(obj["usage"], "llm") + choices = obj.get("choices") or [] + if not choices: + continue + delta = choices[0].get("delta") or {} content = delta.get("content") if content: acc.add_text(content) diff --git a/app/store.py b/app/store.py index 4653dab..3e00c6e 100644 --- a/app/store.py +++ b/app/store.py @@ -128,6 +128,14 @@ class Store(ABC): def add_web_search_usage(self, user_id: str, n: int = 1, day: str | None = None) -> int: """Zaehlt n Web-Suchen dazu und liefert die neue Tages-Zahl.""" + @abstractmethod + def add_cost_usage(self, user_id: str, category: str, usd: float, day: str | None = None) -> None: + """Bucht Kosten (USD) auf Nutzer/Tag/Kategorie.""" + + @abstractmethod + def cost_for_day(self, user_id: str, day: str | None = None) -> float: + """Summe der Kosten (USD) des Nutzers am Tag (Default: heute, UTC).""" + @abstractmethod def delete_user(self, user_id: str) -> bool: """Loescht einen Nutzer und alle seine Daten (Sessions, Nachrichten, Erinnerungen, @@ -243,6 +251,13 @@ class SQLiteStore(Store): count INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (user_id, day) ); + CREATE TABLE IF NOT EXISTS cost_usage ( + user_id TEXT NOT NULL, + day TEXT NOT NULL, + category TEXT NOT NULL, + amount_usd REAL NOT NULL DEFAULT 0, + PRIMARY KEY (user_id, day, category) + ); CREATE TABLE IF NOT EXISTS emergency_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, @@ -517,6 +532,29 @@ class SQLiteStore(Store): ).fetchone() return int(row["count"]) + def add_cost_usage(self, user_id: str, category: str, usd: float, + day: str | None = None) -> None: + if not usd or usd <= 0: + return + day = day or self._today() + with self._connect() as conn: + conn.execute( + "INSERT INTO cost_usage (user_id, day, category, amount_usd) VALUES (?, ?, ?, ?)" + " ON CONFLICT(user_id, day, category) DO UPDATE SET" + " amount_usd = amount_usd + excluded.amount_usd", + (user_id, day, category, float(usd)), + ) + + def cost_for_day(self, user_id: str, day: str | None = None) -> float: + day = day or self._today() + with self._connect() as conn: + row = conn.execute( + "SELECT COALESCE(SUM(amount_usd), 0) AS s FROM cost_usage" + " WHERE user_id = ? AND day = ?", + (user_id, day), + ).fetchone() + return float(row["s"]) if row else 0.0 + def delete_user(self, user_id: str) -> bool: if user_id == ANONYMOUS_USER_ID: raise ValueError("Der anonyme Nutzer kann nicht geloescht werden.") @@ -532,6 +570,7 @@ class SQLiteStore(Store): conn.execute("DELETE FROM memories WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM usage WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM web_search_usage WHERE user_id = ?", (user_id,)) + conn.execute("DELETE FROM cost_usage WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) return True @@ -639,15 +678,21 @@ class SQLiteStore(Store): return [dict(r) for r in rows] def get_all_usage(self) -> list[dict]: + month = datetime.now(timezone.utc).strftime("%Y-%m") + "-%" with self._connect() as conn: rows = conn.execute( "SELECT u.user_id, COALESCE(usr.display_name, u.user_id) AS display_name," " SUM(u.requests) AS total_requests, SUM(u.units) AS total_units," " MAX(u.day) AS last_active," " COALESCE((SELECT SUM(w.count) FROM web_search_usage w" - " WHERE w.user_id = u.user_id), 0) AS total_web_searches" + " WHERE w.user_id = u.user_id), 0) AS total_web_searches," + " COALESCE((SELECT SUM(c.amount_usd) FROM cost_usage c" + " WHERE c.user_id = u.user_id AND c.day LIKE ?), 0) AS month_cost_usd," + " COALESCE((SELECT SUM(uu.requests) FROM usage uu" + " WHERE uu.user_id = u.user_id AND uu.day LIKE ?), 0) AS month_requests" " FROM usage u LEFT JOIN users usr ON usr.id = u.user_id" " GROUP BY u.user_id ORDER BY total_requests DESC", + (month, month), ).fetchall() return [dict(r) for r in rows] diff --git a/app/tools/web_search.py b/app/tools/web_search.py index fa1ca67..8b949ba 100644 --- a/app/tools/web_search.py +++ b/app/tools/web_search.py @@ -11,6 +11,7 @@ from dataclasses import dataclass, field import httpx +from app.core.costs import record_openrouter_cost from app.metrics import metrics from app.providers.llm.base import lang_instruction @@ -70,7 +71,8 @@ class SonarTool: logger.warning("SonarTool ohne API-Key") return ToolResult(text=_NO_DATA, ok=False) - payload = {"model": self.model, "messages": self._messages(query, language)} + payload = {"model": self.model, "messages": self._messages(query, language), + "usage": {"include": True}} headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"} timeout = httpx.Timeout(self.timeout) @@ -90,6 +92,7 @@ class SonarTool: return ToolResult(text=_NO_DATA, ok=False) data = resp.json() + record_openrouter_cost(data.get("usage"), "web_search") try: content = data["choices"][0]["message"]["content"] except (KeyError, IndexError, TypeError): diff --git a/app/web/app.js b/app/web/app.js index e8d2362..3ff4cd6 100644 --- a/app/web/app.js +++ b/app/web/app.js @@ -1827,6 +1827,8 @@ async function loadMetrics() { const totalReq = data.reduce((s, u) => s + (u.total_requests || 0), 0); const totalWS = data.reduce((s, u) => s + (u.total_web_searches || 0), 0); + const totalCost = data.reduce((s, u) => s + (u.month_cost_usd || 0), 0); + const fmtUsd = (n) => "$" + (Number(n) || 0).toFixed(4); const maxReq = Math.max(...data.map((u) => u.total_requests || 0), 1); const bars = data.map((u) => { @@ -1849,6 +1851,8 @@ async function loadMetrics() { Nutzer Anfragen gesamt mit Websuche + Kosten (Monat) + Ø/Anfrage Einheiten Letzte Aktivität @@ -1859,6 +1863,8 @@ async function loadMetrics() { ${escHtml(u.display_name)} ${u.total_requests ?? 0} ${u.total_web_searches ?? 0} + ${fmtUsd(u.month_cost_usd)} + ${u.month_requests ? fmtUsd((u.month_cost_usd || 0) / u.month_requests) : "—"} ${u.total_units ?? 0} ${u.last_active || "—"} `).join("")} @@ -1868,6 +1874,8 @@ async function loadMetrics() { Gesamt ${totalReq} ${totalWS} + ${fmtUsd(totalCost)} +