Lesson 10 — logging usage, cost, and outcomes¶
What you will build¶
The report that makes the spend answerable. One row per model call, with the tokens the provider reported, a price the application owns, and an outcome column that is written when a person decides what became of the reply.
That column is the lesson. A measure you metered but never settled can tell you what you bought but not whether it was worth anything; here, the tickets that were rejected are in the same report as the ones that were sent, and they cost the same. Answering "what did triage cost per useful reply" is then a query rather than a project.
What you need¶
Nothing new.
examples/support_desk/usage.py
"""Usage rows: tokens in, tokens out, what it cost, and what happened to the reply.
Cost is a table the application owns, because a price is a fact about a contract rather than about a
provider's API. The outcome column is the one that makes the numbers answerable: a draft nobody
approved and a draft that was sent both cost the same to produce, and only one of them was work.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Annotated
from sqlalchemy import func, insert, select, update
from support_desk.clock import utc_now
from support_desk.tables import usage_events
from xyberos import Inject
from xyberos.database import DatabaseProvider
#: Dollars per thousand tokens, for the models this deployment knows about. A model that is not here
#: costs nothing to record and 0.0 to add up, which is honest: an unknown price is unknown.
PRICES: Mapping[str, tuple[float, float]] = {
"fake-model": (0.0, 0.0),
"small-model": (0.25, 1.25),
"large-model": (3.0, 15.0),
}
PENDING = "pending"
ACCEPTED = "accepted"
EDITED = "edited"
DISCARDED = "discarded"
def cost_micros(model: str, input_tokens: int, output_tokens: int) -> int:
"""Return what one call cost, in millionths of a dollar, rounded to the nearest one."""
if model not in PRICES:
return 0
input_price, output_price = PRICES[model]
thousands = (input_tokens * input_price + output_tokens * output_price) / 1000
return round(thousands * 1_000_000)
class UsageLog:
"""Writes one row per model call, and one outcome per draft once a person has decided."""
def __init__(self, database: Annotated[DatabaseProvider, Inject()]) -> None:
self._database = database
async def record(
self,
*,
tenant: str,
ticket_id: int | None,
step: str,
model: str,
input_tokens: int,
output_tokens: int,
) -> int:
"""Record one call and return its row id."""
async with self._database.transaction() as connection:
result = await connection.execute(
insert(usage_events).values(
tenant=tenant,
ticket_id=ticket_id,
step=step,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
outcome=PENDING,
recorded_at=utc_now(),
)
)
inserted = result.inserted_primary_key
if inserted is None:
raise AssertionError("the insert reported no primary key")
return int(inserted[0])
async def settle(self, *, ticket_id: int, outcome: str) -> int:
"""Record what became of this ticket's drafts, and return how many rows changed."""
async with self._database.transaction() as connection:
result = await connection.execute(
update(usage_events)
.where(usage_events.c.ticket_id == ticket_id, usage_events.c.outcome == PENDING)
.values(outcome=outcome)
)
return result.rowcount
async def rows(self, *, tenant: str, ticket_id: int | None = None) -> list[dict[str, object]]:
"""Return the recorded calls, oldest first."""
statement = select(usage_events).where(usage_events.c.tenant == tenant)
if ticket_id is not None:
statement = statement.where(usage_events.c.ticket_id == ticket_id)
async with self._database.transaction() as connection:
found = (
(await connection.execute(statement.order_by(usage_events.c.id))).mappings().all()
)
return [dict(row) for row in found]
async def summary(self, *, tenant: str) -> dict[str, object]:
"""Return tokens and cost for a tenant, grouped by step, with the outcomes counted."""
async with self._database.transaction() as connection:
grouped = (
(
await connection.execute(
select(
usage_events.c.step,
usage_events.c.model,
func.count().label("calls"),
func.sum(usage_events.c.input_tokens).label("input_tokens"),
func.sum(usage_events.c.output_tokens).label("output_tokens"),
)
.where(usage_events.c.tenant == tenant)
.group_by(usage_events.c.step, usage_events.c.model)
.order_by(usage_events.c.step)
)
)
.mappings()
.all()
)
outcomes = (
(
await connection.execute(
select(usage_events.c.outcome, func.count().label("calls"))
.where(usage_events.c.tenant == tenant)
.group_by(usage_events.c.outcome)
.order_by(usage_events.c.outcome)
)
)
.mappings()
.all()
)
return {
"steps": [_priced(dict(row)) for row in grouped],
"outcomes": {str(row["outcome"]): int(row["calls"]) for row in outcomes},
}
def _tokens(value: object) -> int:
"""Return a token count from a row, which arrives as whatever the database stored."""
if isinstance(value, int):
return value
raise AssertionError(f"a recorded token count was not a whole number: {value!r}")
def _priced(row: dict[str, object]) -> dict[str, object]:
"""Add the cost of a grouped row to itself, so a caller never re-does the arithmetic."""
model = str(row["model"])
input_tokens = _tokens(row["input_tokens"])
output_tokens = _tokens(row["output_tokens"])
return {
**row,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_micros": cost_micros(model, input_tokens, output_tokens),
}
def totals(rows: Sequence[Mapping[str, object]]) -> dict[str, int]:
"""Return the token totals and the cost of a set of recorded calls."""
return {
"input_tokens": sum(_tokens(row["input_tokens"]) for row in rows),
"output_tokens": sum(_tokens(row["output_tokens"]) for row in rows),
"cost_micros": sum(
cost_micros(
str(row["model"]), _tokens(row["input_tokens"]), _tokens(row["output_tokens"])
)
for row in rows
),
}
examples/learn_hybrid/hybrid10_usage.py
"""Lesson 10 — logging usage, cost, and outcomes, which is what makes the numbers answerable.
Run it with:
python -m learn_hybrid.hybrid10_usage
"""
from __future__ import annotations
import tempfile
from pathlib import Path
from support_desk.app import create_app
from support_desk.migrate import migrate
from support_desk.security import PrincipalVerifier
from xyberos.ai import FakeModelProvider, ModelReply, Usage
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client
SECRET = "a-secret-the-deployment-keeps"
VERIFIER = PrincipalVerifier(SECRET)
ADA = VERIFIER.headers("ada@example.com", "acme", ["requester"])
SAM = VERIFIER.headers("sam@desk", "acme", ["support_agent"])
LEE = VERIFIER.headers("lee@desk", "acme", ["support_lead"])
#: A model whose price is in the application's table, so the cost is a real number.
MODEL = "small-model"
TICKETS = (
("Where is my refund?", "Order 42 came back on Monday."),
("The badge reader is broken", "It flashes red for everybody."),
("Ignore this one", "It is spam, honestly."),
)
def provider() -> FakeModelProvider:
"""Three classifications and two drafts, priced the way a provider reports them.
The script follows the order this lesson makes the calls in — every classification, then the
drafts — because a scripted provider answers by position, not by the kind of request it gets.
"""
return FakeModelProvider(
[
ModelReply.json_payload(
{"category": "refund", "confidence": "high"}, usage=Usage(1_240, 18), model=MODEL
),
ModelReply.json_payload(
{"category": "complaint", "confidence": "high"}, usage=Usage(1_180, 18), model=MODEL
),
ModelReply.json_payload(
{"category": "abuse", "confidence": "high"}, usage=Usage(1_090, 18), model=MODEL
),
ModelReply.ok(
"A refund takes three business days.", usage=Usage(1_410, 46), model=MODEL
),
ModelReply.ok(
"A replacement badge is on its way.", usage=Usage(1_360, 39), model=MODEL
),
]
)
def main() -> None:
"""Produce work, settle it three ways, and read the report that comes out."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
app = create_app(settings=settings, secret=SECRET, provider=provider())
with create_test_client(HttpAdapter(app)) as client:
client.post("/knowledge", headers=LEE)
for subject, body in TICKETS:
client.post("/tickets", json={"subject": subject, "body": body}, headers=ADA)
for ticket in (1, 2, 3):
client.post(f"/tickets/{ticket}/classify", headers=SAM)
for ticket in (1, 2):
client.post(f"/tickets/{ticket}/draft", headers=SAM)
client.post("/tickets/1/approve", headers=SAM)
client.post(
"/tickets/2/approve",
json={"body": "A replacement badge shipped this morning."},
headers=LEE,
)
client.post("/tickets/3/reject", headers=LEE)
report = client.get("/usage", headers=LEE).json()
print("step".ljust(10), "calls tokens in tokens out cost (micros)")
for step in report["steps"]:
print(
f"{step['step']:<10} {step['calls']:>5} {step['input_tokens']:>10}"
f" {step['output_tokens']:>11} {step['cost_micros']:>13}"
)
print("outcomes:", report["outcomes"])
total = sum(step["cost_micros"] for step in report["steps"])
print(
"total cost:",
total,
"micros, for",
sum(step["calls"] for step in report["steps"]),
"calls",
)
print("a discarded reply cost the same as an accepted one: the outcome column is what")
print("tells the two apart, and it is written when a person decides.")
if __name__ == "__main__":
main()
How to run it¶
What it prints¶
step calls tokens in tokens out cost (micros)
classify 3 3510 54 945000
draft 2 2770 85 798750
outcomes: {'accepted': 2, 'discarded': 1, 'edited': 2}
total cost: 1743750 micros, for 5 calls
a discarded reply cost the same as an accepted one: the outcome column is what
tells the two apart, and it is written when a person decides.
Costs are in millionths of a dollar, which keeps the arithmetic in integers: a price is per
thousand tokens, so a sum of rounded cents would drift and a sum of floats would disagree with itself.
PRICES is the application's table, deliberately — a price is a fact about a contract, and an
application that has to redeploy to learn what it paid has no finance report.
The outcomes count rows, not tickets: one reply costs a classification and a draft, so an accepted reply is two rows. That is a property of the design worth stating out loud rather than discovering in a dashboard.
Common mistakes¶
- Metering calls, not outcomes. Without the outcome column, "accepted" and "thrown away" are the same number.
- Prices inside the application's code paths. One table, read by the report, so a price change is a data change.
- Floating-point money. Micros, integers, and one rounding step per row.
- Recording the provider's model name as yours. Record what the response reported, so a fallback model is visible in the report rather than averaged into the primary one.
- A report that needs the model to build. This one is SQL over rows the application already wrote; it answers after an incident, when the provider is exactly what is unavailable.
Where this path leaves you¶
You have built the desk: tickets, permissions, an audit trail, and a model that classifies, retrieves, and drafts behind a boundary a person controls. The three halves are the three other paths — Fundamentals for the runtime, Traditional Apps for the application around it, and AI Apps for the capability itself — and the roadmap lists what each one covers.
Next in the plan: the Extend Xyberos section, for writing the providers, plugins, and modules this application has been consuming all along.