Houmeres/LLeMbas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Show what a reply cost, live and afterwards

Browse Source 
Tokens, how full the context is, and tokens per second -- as chips under
each assistant bubble, updating while the reply streams and still there when
it finishes.

The numbers come from one Metrics object built either from the generation
still being written or from the row it left behind. That is the point rather
than tidiness: the finished bubble is re-rendered from the database the
instant the stream ends, so two code paths would make the figures visibly
jump at exactly the moment someone is watching them. Here the only thing
that changes is that an estimate may become exact.

Message.usage_json has existed and been dead since the schema was written.
It is the store.

Two counts that look like one. prompt and completion are summed across tool
rounds -- what the reply cost. context_tokens is overwritten each round with
that round's prompt plus completion -- what the window actually holds. A
three-round reply pays for its prompt three times and only ever occupies the
window once, so a single number would be wrong for one of the two questions.

Generation gains started_at as a field rather than a local in _run, because
_follow is a different function that sees only the Generation and otherwise
has nothing to compute a live speed against. It also carries a prompt
estimate taken before the first chunk, since real usage arrives in one chunk
at the very end and a percentage that appears only after the reply is
useless.

Everything is marked with a tilde when the endpoint reported nothing, and
the percentage is simply absent when no context length is set: unknown has
to stay tellable from small, and a percentage of an unknown total is a
made-up number in a place people trust numbers.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-08-01 00:40:44 +02:00
parent ff58ada6bf
commit b8618b0c91
8 changed files with 485 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lembas/api/chats.py
+14
View File
@@ -21,6 +21,7 @@ from lembas.services import audio as audio_service
from lembas.services import chat as chat_service
from lembas.services import files as files_service
from lembas.services import generation as generation_service
from lembas.services import metrics as metrics_service
from lembas.services import sse
from lembas.services.markdown import escape_text, render_markdown
from lembas.web.templating import render, templates
@@ -257,6 +258,7 @@ async def _follow(chat_id: str, message_id: str) -> AsyncIterator[str]:
yield sse.event("tools", _tool_activity(generation.tool_events))
if generation.content:
yield sse.event("render", render_markdown(generation.text))
yield sse.event("metrics", _metrics_html(generation))
last_frame = time.monotonic()
if generation.done:
@@ -314,6 +316,18 @@ async def _follow(chat_id: str, message_id: str) -> AsyncIterator[str]:
yield sse.event("close", "")
def _metrics_html(generation) -> str:
"""The metric chips for a reply still being written.
Built from the same Metrics object the finished bubble uses, so the numbers
do not jump when the stream ends -- the only thing that changes is that an
estimate may have become exact.
"""
return templates.get_template("chat/_metrics.html").render(
{"metrics": metrics_service.from_generation(generation)}
)
def _thread_context(db: DBSession, chat: Chat, user: User) -> dict:
"""Everything chat/_thread.html needs to render the conversation."""
messages = list(
src/lembas/services/generation.py
+55
View File
@@ -25,10 +25,13 @@ from datetime import UTC, datetime, timedelta
from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Message, User
from lembas.db.session import session_scope
from lembas.services import chat as chat_service
from lembas.services import metrics as metrics_service
from lembas.services import prompts as prompts_service
from lembas.services import tokens
from lembas.services import tools as tools_service
from lembas.services.llm.openai_client import (
LLMError,
chunk_usage,
delta_reasoning,
delta_text,
delta_tool_calls,
@@ -64,6 +67,26 @@ class Generation:
# live as the model works and kept on the message afterwards.
tool_events: list[dict] = field(default_factory=list)
# --- What it cost --------------------------------------------------------
# Prompt and completion are summed across tool rounds: what the reply cost.
# context_tokens is overwritten each round with that round's prompt plus
# completion, because a three-round reply pays for its prompt three times
# but only ever occupies the window once.
prompt_tokens: int = 0
completion_tokens: int = 0
context_tokens: int = 0
context_limit: int = 0
# Filled from the assembled request before the first chunk, so a follower
# has a percentage to show while the reply is still being written -- real
# usage only arrives in a single chunk at the very end.
prompt_estimate: int = 0
rounds: int = 0
# time.monotonic() at the start. A field rather than a local in `_run`
# because `_follow` is a different function that sees only this object, and
# without it there is nothing to compute a live tokens/second against.
started_at: float = 0.0
elapsed_ms: int = 0
error: str = ""
stopped: bool = False
done: bool = False
@@ -183,6 +206,7 @@ async def _run(generation: Generation) -> None:
"""
splitter = ReasoningSplitter()
started = time.monotonic()
generation.started_at = started
reasoning_started: float | None = None
question = ""
endpoint = model_id = None
@@ -212,13 +236,30 @@ async def _run(generation: Generation) -> None:
title_prompt = prompts_service.resolve(db, "task.title")
tool_context = tools_service.context_for(db, owner, chat)
model = chat_service.model_for(db, chat)
generation.context_limit = model.context_length if model is not None else 0
generation.prompt_estimate = tokens.estimate_request(payload)
for round_number in range(tools_service.MAX_ROUNDS + 1):
generation.rounds = round_number + 1
accumulator = tools_service.ToolCallAccumulator()
# Text the model produced in *this* round, needed separately from
# generation.content when echoing the assistant turn back.
round_text: list[str] = []
async for chunk in stream_chat(endpoint, payload):
counts = chunk_usage(chunk)
if counts is not None:
generation.prompt_tokens += counts.get("prompt_tokens", 0)
generation.completion_tokens += counts.get("completion_tokens", 0)
# Overwritten, not summed: this round's prompt already
# contains every earlier round.
generation.context_tokens = counts.get("prompt_tokens", 0) + counts.get(
"completion_tokens", 0
)
generation.touch()
thought = delta_reasoning(chunk)
if thought:
if reasoning_started is None:
@@ -307,6 +348,17 @@ async def _run(generation: Generation) -> None:
if reasoning_started is not None and not generation.reasoning_ms:
generation.reasoning_ms = int((time.monotonic() - reasoning_started) * 1000)
generation.elapsed_ms = int((time.monotonic() - started) * 1000)
if not generation.completion_tokens:
# The endpoint reported nothing, so fall back to the estimate. Marked
# as such everywhere it is shown -- four characters to a token is
# wrong enough on code and CJK to be worth saying out loud.
generation.completion_tokens = tokens.estimate(
generation.text + generation.thinking
)
generation.prompt_tokens = generation.prompt_estimate
generation.context_tokens = generation.prompt_tokens + generation.completion_tokens
# Naming the chat is a second, short completion, so it has to happen
# here rather than in the synchronous persist step below. Best-effort:
# a chat title is never worth surfacing an error for.
@@ -379,6 +431,9 @@ def _persist(generation: Generation, title: str, elapsed: float) -> None:
message.reasoning = generation.thinking
message.reasoning_ms = generation.reasoning_ms
message.tool_calls_json = generation.tool_events
message.usage_json = metrics_service.to_json(
metrics_service.from_generation(generation)
)
message.error = generation.error
message.stopped = generation.stopped
message.complete = True
src/lembas/services/metrics.py
+136
View File
@@ -0,0 +1,136 @@
"""What a reply cost, how fast it arrived, and how full the window is.
One shape, built either from a generation still being written or from the row
it left behind. That matters more than it looks: the finished bubble is
re-rendered from the database the instant the stream ends, so if the live
numbers and the stored ones came from different code they would visibly jump at
exactly the moment the reader is looking at them. Here the only thing that
changes when a reply finishes is that an estimate may become exact.
Nothing here is authoritative about tokens. `estimated` says which kind of
number this is, and every surface that shows one has to say so too.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from lembas.services import tokens
# Where the context bar changes colour. Not thresholds anyone tunes: they mark
# "worth noticing" and "about to be a problem", and the second is deliberately
# below the default compaction threshold so the warning arrives first.
WARNING_AT = 80
DANGER_AT = 95
@dataclass(frozen=True)
class Metrics:
"""Token counts and timing for one reply."""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
# What the window holds after this turn: the last round's prompt plus its
# completion. Distinct from prompt+completion summed over tool rounds, which
# is what the reply *cost* -- a three-round reply pays for its prompt three
# times but only ever occupies the window once.
context_tokens: int = 0
context_limit: int = 0
estimated: bool = False
elapsed_ms: int = 0
rounds: int = 1
@property
def percent(self) -> int:
"""How full the window is, or 0 when nobody has said how big it is."""
if self.context_limit <= 0 or self.context_tokens <= 0:
return 0
return min(100, round(self.context_tokens * 100 / self.context_limit))
@property
def tokens_per_second(self) -> float:
if self.elapsed_ms <= 0 or self.completion_tokens <= 0:
return 0.0
return self.completion_tokens / (self.elapsed_ms / 1000)
@property
def pressure(self) -> str:
""""", "warning" or "danger" -- the class the context chip takes."""
percent = self.percent
if not percent:
return ""
if percent >= DANGER_AT:
return "danger"
if percent >= WARNING_AT:
return "warning"
return ""
@property
def has_anything(self) -> bool:
return bool(self.total_tokens or self.completion_tokens or self.elapsed_ms)
def from_generation(generation: Any) -> Metrics:
"""Metrics for a reply still being written.
Usage arrives in a single chunk at the very end, so mid-stream there is
nothing to report and everything is estimated. The counts stop being
estimates the moment that chunk lands, which is usually a beat before the
bubble is replaced.
"""
import time
completion = generation.completion_tokens or tokens.estimate(
generation.text + generation.thinking
)
prompt = generation.prompt_tokens or generation.prompt_estimate
elapsed = generation.elapsed_ms or (
int((time.monotonic() - generation.started_at) * 1000) if generation.started_at else 0
)
return Metrics(
prompt_tokens=prompt,
completion_tokens=completion,
total_tokens=prompt + completion,
context_tokens=generation.context_tokens or (prompt + completion),
context_limit=generation.context_limit,
estimated=not (generation.prompt_tokens and generation.completion_tokens),
elapsed_ms=elapsed,
rounds=max(1, generation.rounds),
)
def from_message(usage_json: dict[str, Any] | None) -> Metrics:
"""Metrics for a finished reply, read back off the row."""
stored = usage_json or {}
def _int(key: str) -> int:
value = stored.get(key)
return int(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else 0
return Metrics(
prompt_tokens=_int("prompt_tokens"),
completion_tokens=_int("completion_tokens"),
total_tokens=_int("total_tokens"),
context_tokens=_int("context_tokens"),
context_limit=_int("context_limit"),
estimated=bool(stored.get("estimated")),
elapsed_ms=_int("elapsed_ms"),
rounds=max(1, _int("rounds")),
)
def to_json(metrics: Metrics) -> dict[str, Any]:
"""The shape stored in Message.usage_json."""
return {
"prompt_tokens": metrics.prompt_tokens,
"completion_tokens": metrics.completion_tokens,
"total_tokens": metrics.total_tokens,
"context_tokens": metrics.context_tokens,
"context_limit": metrics.context_limit,
"estimated": metrics.estimated,
"elapsed_ms": metrics.elapsed_ms,
"rounds": metrics.rounds,
}
src/lembas/web/static/css/chat.css
+37
View File
@@ -167,6 +167,43 @@
.reasoning__summary::-webkit-details-marker { display: none; }
.reasoning__summary:hover { color: var(--ink); background: var(--surface-hover); }
/* --- Metrics ---------------------------------------------------------------
What a reply cost, under the bubble. Quiet by default: it is reference, not
something to read every time.
*/
.msg__metrics {
display: flex;
align-items: center;
flex-wrap: wrap;
gap: var(--sp-3);
margin-top: var(--sp-2);
font-size: var(--text-xs);
color: var(--ink-faint);
font-variant-numeric: tabular-nums;
}
.msg__metrics:empty { display: none; }
.metric { display: inline-flex; align-items: center; gap: var(--sp-1); cursor: default; }
.metric__bar {
display: inline-block;
width: 3rem;
height: 0.3rem;
border-radius: var(--radius-full);
background: var(--surface-active);
overflow: hidden;
}
.metric__fill {
display: block;
height: 100%;
background: var(--ink-faint);
transition: width var(--transition);
}
.metric--context.is-warning { color: var(--warning); }
.metric--context.is-warning .metric__fill { background: var(--warning); }
.metric--context.is-danger { color: var(--danger); }
.metric--context.is-danger .metric__fill { background: var(--danger); }
.reasoning__icon { color: var(--leaf); flex: none; }
.reasoning__label { flex: 1; font-style: italic; }
src/lembas/web/templates/chat/_message.html
+13
View File
@@ -120,6 +120,10 @@
<div class="msg__waiting">
<span class="dots"><i></i><i></i><i></i></span>
</div>
{# Counts as the reply is written. Everything is an estimate until the
usage chunk lands at the very end, and the chips say so. #}
<div class="msg__metrics" id="metrics-{{ message.id }}"
sse-swap="metrics" hx-swap="innerHTML"></div>
{% else %}
{# Finished. Same order as the live view above -- thinking, then what it
looked up, then the answer -- so a reply does not rearrange itself the
@@ -177,6 +181,15 @@
leave an empty box under the file. #}
{% endif %}
{% if not streaming and message.role == "assistant" and message.usage_json %}
{# Above the buttons, not among them: the actions row is things you press. #}
<div class="msg__metrics" id="metrics-{{ message.id }}">
{% with metrics = message.usage_json | metrics %}
{% include "chat/_metrics.html" %}
{% endwith %}
</div>
{% endif %}
{% if not streaming %}
<footer class="msg__actions">
<button class="btn btn--icon btn--sm" type="button"
src/lembas/web/templates/chat/_metrics.html
+32
View File
@@ -0,0 +1,32 @@
{#
What a reply cost. Chips only, no wrapper: the same markup is swapped into
the live bubble with innerHTML and rendered into the finished one, so the
numbers cannot change shape when the stream ends.
A tilde means the endpoint reported no token counts and these were worked out
at about four characters per token. Nothing here is ever shown as exact when
it is not.
#}
{% if metrics.has_anything %}
<span class="metric" title="{% if metrics.estimated %}Estimated: this endpoint reports no token counts.
{% endif %}{{ metrics.prompt_tokens }} in, {{ metrics.completion_tokens }} out{% if metrics.rounds > 1 %}, over {{ metrics.rounds }} rounds of tool calls{% endif %}">
{% if metrics.estimated %}~{% endif %}{{ metrics.total_tokens }} tokens
</span>
{% if metrics.context_limit %}
<span class="metric metric--context{{ ' is-' ~ metrics.pressure if metrics.pressure }}"
title="{% if metrics.estimated %}Estimated. {% endif %}{{ metrics.context_tokens }} of {{ metrics.context_limit }} tokens of context used">
<span class="metric__bar">
{# A width is data, not a design value: it is the measurement itself. #}
<span class="metric__fill" style="width: {{ metrics.percent }}%"></span>
</span>
{% if metrics.estimated %}~{% endif %}{{ metrics.percent }}%
</span>
{% endif %}
{% if metrics.tokens_per_second %}
<span class="metric" title="{% if metrics.estimated %}Estimated. {% endif %}{{ metrics.completion_tokens }} tokens in {{ metrics.elapsed_ms }} ms">
{% if metrics.estimated %}~{% endif %}{{ '%.1f' | format(metrics.tokens_per_second) }} tok/s
</span>
{% endif %}
{% endif %}
src/lembas/web/templating.py
+5
View File
@@ -11,6 +11,7 @@ from fastapi.templating import Jinja2Templates
from lembas import __version__
from lembas.config import settings
from lembas.db.models import User
from lembas.services import metrics as metrics_service
from lembas.services.reasoning import format_duration
TEMPLATE_DIR = Path(__file__).parent / "templates"
@@ -23,6 +24,10 @@ templates.env.lstrip_blocks = True
# {{ message.reasoning_ms | duration }} -> "8 seconds"
templates.env.filters["duration"] = format_duration
# {{ message.usage_json | metrics }} -> a Metrics, so the finished bubble reads
# its numbers through the same object the live frames are built from.
templates.env.filters["metrics"] = metrics_service.from_message
def stable_hue(value: str) -> int:
"""A deterministic 0-359 hue for a string.
tests/test_metrics.py
+193 -1
View File
@@ -6,7 +6,7 @@ from fastapi.testclient import TestClient
from sqlalchemy import select
from lembas.db.models import Connection, Model
from lembas.services import tokens
from lembas.services import metrics, tokens
from lembas.services.crypto import encrypt
from lembas.services.llm.openai_client import chunk_usage, context_from
@@ -258,3 +258,195 @@ def test_a_request_estimate_includes_the_tools_array():
assert tokens.estimate_request(payload) > tokens.estimate_request(
{"messages": payload["messages"]}
)
# --- The Metrics object -------------------------------------------------------
def test_the_percentage_is_zero_when_nobody_said_how_big_the_window_is():
"""Unknown must stay tellable from small. A percentage of an unknown total
is a made-up number in a place people trust numbers."""
assert metrics.Metrics(context_tokens=5000, context_limit=0).percent == 0
assert metrics.Metrics(context_tokens=5000, context_limit=10000).percent == 50
def test_pressure_marks_the_bar_only_when_it_is_worth_noticing():
assert metrics.Metrics(context_tokens=50, context_limit=100).pressure == ""
assert metrics.Metrics(context_tokens=85, context_limit=100).pressure == "warning"
assert metrics.Metrics(context_tokens=96, context_limit=100).pressure == "danger"
assert metrics.Metrics(context_tokens=0, context_limit=0).pressure == ""
def test_speed_needs_both_a_count_and_a_clock():
assert metrics.Metrics(completion_tokens=100, elapsed_ms=2000).tokens_per_second == 50.0
assert metrics.Metrics(completion_tokens=100, elapsed_ms=0).tokens_per_second == 0.0
assert metrics.Metrics(completion_tokens=0, elapsed_ms=2000).tokens_per_second == 0.0
def test_metrics_survive_a_round_trip_through_the_row():
original = metrics.Metrics(
prompt_tokens=100,
completion_tokens=20,
total_tokens=120,
context_tokens=120,
context_limit=8192,
estimated=True,
elapsed_ms=1500,
rounds=2,
)
assert metrics.from_message(metrics.to_json(original)) == original
def test_a_row_with_no_usage_reads_as_nothing_rather_than_failing():
assert metrics.from_message(None).has_anything is False
assert metrics.from_message({}).has_anything is False
assert metrics.from_message({"prompt_tokens": "lots"}).prompt_tokens == 0
# --- Through a generation -----------------------------------------------------
def _chunks(*frames: str) -> str:
return "".join(f"data: {frame}\n\n" for frame in frames) + "data: [DONE]\n\n"
async def test_usage_is_summed_across_tool_rounds(db, registered, make_chat, monkeypatch):
"""Prompt and completion are what the reply cost; context_tokens is what the
window holds. A three-round reply pays for its prompt three times and only
ever occupies the window once."""
from lembas.services import generation as generation_service
generation = generation_service.Generation(chat_id="c", message_id="m")
for prompt_tokens, completion_tokens in ((100, 10), (250, 20)):
counts = {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens}
generation.prompt_tokens += counts["prompt_tokens"]
generation.completion_tokens += counts["completion_tokens"]
generation.context_tokens = prompt_tokens + completion_tokens
assert generation.prompt_tokens == 350
assert generation.completion_tokens == 30
assert generation.context_tokens == 270
async def test_a_reply_records_what_it_cost(client: TestClient, db, registered, mock_http):
import httpx
from lembas.db.models import Message
def handler(_request: httpx.Request) -> httpx.Response:
if _request.url.path.endswith("/models"):
return httpx.Response(200, json={"data": [{"id": "m", "context_length": 1000}]})
return httpx.Response(
200,
text=_chunks(
'{"choices":[{"delta":{"content":"Waybread."}}]}',
'{"choices":[],"usage":{"prompt_tokens":120,"completion_tokens":8}}',
),
)
mock_http(handler)
client.post(
"/admin/connections",
data={"name": "Local", "base_url": "http://x.test", "api_key": ""},
follow_redirects=False,
)
client.post("/api/chats/start", data={"content": "what is lembas?"})
reply = db.scalar(select(Message).where(Message.role == "assistant"))
# Following the stream is what runs the generation to completion.
with client.stream(
"GET", f"/api/chats/{reply.chat_id}/messages/{reply.id}/stream"
) as response:
list(response.iter_lines())
db.refresh(reply)
assert reply.usage_json["prompt_tokens"] == 120
assert reply.usage_json["completion_tokens"] == 8
assert reply.usage_json["context_limit"] == 1000
assert reply.usage_json["estimated"] is False
assert reply.usage_json["elapsed_ms"] >= 0
def test_the_chips_render_from_a_stored_row(client: TestClient, db, registered, make_chat):
from lembas.db.models import Message
_model(db)
chat_id = make_chat()
db.add(
Message(
chat_id=chat_id,
role="assistant",
content="Waybread.",
usage_json={
"prompt_tokens": 100,
"completion_tokens": 20,
"total_tokens": 120,
"context_tokens": 120,
"context_limit": 1000,
"estimated": False,
"elapsed_ms": 2000,
"rounds": 1,
},
)
)
db.commit()
page = client.get(f"/chat/{chat_id}").text
assert "120 tokens" in page
assert "12%" in page
assert "10.0 tok/s" in page
assert "~" not in page.split("msg__metrics")[1][:400]
def test_an_estimated_row_says_so(client: TestClient, db, registered, make_chat):
from lembas.db.models import Message
_model(db)
chat_id = make_chat()
db.add(
Message(
chat_id=chat_id,
role="assistant",
content="Waybread.",
usage_json={
"prompt_tokens": 100,
"completion_tokens": 20,
"total_tokens": 120,
"context_tokens": 120,
"context_limit": 1000,
"estimated": True,
"elapsed_ms": 2000,
"rounds": 1,
},
)
)
db.commit()
page = client.get(f"/chat/{chat_id}").text
assert "~120 tokens" in page
assert "reports no token counts" in page
def test_no_context_length_means_no_percentage(client: TestClient, db, registered, make_chat):
from lembas.db.models import Message
_model(db)
chat_id = make_chat()
db.add(
Message(
chat_id=chat_id,
role="assistant",
content="Waybread.",
usage_json={
"prompt_tokens": 100,
"completion_tokens": 20,
"total_tokens": 120,
"context_tokens": 120,
"context_limit": 0,
"estimated": False,
"elapsed_ms": 2000,
"rounds": 1,
},
)
)
db.commit()
page = client.get(f"/chat/{chat_id}").text
assert "120 tokens" in page
assert "metric--context" not in page