diff --git a/docs/cli.md b/docs/cli.md index 138683b011..fbb9f6eb48 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -107,6 +107,15 @@ See [models.md](models.md#moe-strategies) for what each strategy does, and [perf | `--enable-cache-report` | off | Report prefix-cache hits in each response's usage block | | `--anthropic-inline-system` | auto | Placement of late Anthropic system instructions: `auto`, `preserve`, or `fold` | +Every chat completion, completion and Responses response also carries a llama.cpp-shaped +`timings` object next to `usage` (`prompt_n`, `prompt_ms`, `prompt_per_second`, +`predicted_n`, `predicted_ms`, `predicted_per_second`, `cache_n`), so frontends like +llama-swap and Open Web UI show per-request prefill/decode rates and the prefix-cache hit. +Its spans match the cumulative timing on `GET /v1/stats`: prefill runs from request start +to the first output token, decode between output tokens. `cache_n` follows llama.cpp +(always present, `prompt_n` counts only the evaluated tokens) and is not gated by +`--enable-cache-report`, which controls only `usage.prompt_tokens_details.cached_tokens`. + ### Image input Experimental. Needs a checkpoint whose family registers a vision encoder ([models.md](models.md#image-input) lists them and how each one diff --git a/python/freetoken/server/generation.py b/python/freetoken/server/generation.py index 2e342ff899..82f2595bc2 100644 --- a/python/freetoken/server/generation.py +++ b/python/freetoken/server/generation.py @@ -115,6 +115,8 @@ class GenDone: completion_tokens: int matched_stop: str | None = None cached_tokens: int = 0 + prompt_ms: float = 0.0 + predicted_ms: float = 0.0 GenEvent = ReasoningDelta | ContentDelta | ToolCallStart | ToolCallArgsDelta | ToolCallsDelta | GenDone @@ -130,6 +132,35 @@ class GenResult: completion_tokens: int matched_stop: str | None = None cached_tokens: int = 0 + prompt_ms: float = 0.0 + predicted_ms: float = 0.0 + + +def timings_block( + prompt_tokens: int, completion_tokens: int, cached_tokens: int, + prompt_ms: float, predicted_ms: float, +) -> dict[str, Any] | None: + """llama.cpp's timings shape (server_slot_stats::to_json), which frontends like + llama-swap parse off the response. prompt_n counts only the tokens evaluated after + the cached prefix and decode runs n_gen - 1 steps (the first token rides the last + prompt batch), matching llama.cpp's own arithmetic.""" + if prompt_ms <= 0.0: + return None + prompt_s = prompt_ms / 1000.0 + predicted_s = predicted_ms / 1000.0 + processed = prompt_tokens - cached_tokens + gen_steps = max(completion_tokens - 1, 0) + return { + "cache_n": cached_tokens, + "prompt_n": processed, + "prompt_ms": round(prompt_ms, 2), + "prompt_per_token_ms": round(prompt_ms / processed, 2) if processed > 0 else 0.0, + "prompt_per_second": round(processed / prompt_s, 2) if prompt_s > 0 else 0.0, + "predicted_n": completion_tokens, + "predicted_ms": round(predicted_ms, 2), + "predicted_per_token_ms": round(predicted_ms / gen_steps, 2) if gen_steps > 0 else 0.0, + "predicted_per_second": round(gen_steps / predicted_s, 2) if predicted_s > 0 else 0.0, + } @dataclass @@ -608,6 +639,7 @@ async def _generate_events_impl(uid: int, spec: GenSpec, state: Any) -> AsyncIte completion_tokens = 0 cached_tokens = 0 pending = "" + started = time.monotonic() parse_tools = spec.parse_tools reasoning_parser = _make_reasoning_parser(spec, state) specials = _leaked_special_tokens(state) @@ -696,12 +728,19 @@ def _route_tool_text(piece: str) -> list[GenEvent]: engine_finish_reason: str | None = None engine_matched_stop: str | None = None + first_output_at: float | None = None + last_output_at: float | None = None async for ack in state.wait_for_ack(uid): if getattr(ack, "error", None): raise GenerationError(ack.error, getattr(ack, "error_code", None)) prompt_tokens += ack.prompt_tokens_delta completion_tokens += ack.completion_tokens_delta cached_tokens += ack.cached_tokens + if ack.completion_tokens_delta > 0: + now = time.monotonic() + if first_output_at is None: + first_output_at = now + last_output_at = now content_delta = ack.incremental_output if reasoning_parser is not None and content_delta: reasoning_delta, content_delta = reasoning_parser.parse_stream_chunk(content_delta) @@ -781,9 +820,14 @@ def _route_tool_text(piece: str) -> list[GenEvent]: elif parse_tools and pending: yield ContentDelta(strip_special_tokens(pending, specials)) + # No output ever arrived: the whole span is prefill. + ended = time.monotonic() + prompt_ms = ((first_output_at if first_output_at is not None else ended) - started) * 1000.0 + predicted_ms = (last_output_at - first_output_at) * 1000.0 if first_output_at is not None else 0.0 yield GenDone( finish_reason, prompt_tokens, completion_tokens, matched_stop=engine_matched_stop, cached_tokens=cached_tokens, + prompt_ms=prompt_ms, predicted_ms=predicted_ms, ) @@ -794,6 +838,9 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any) -> GenResult: prompt_tokens = 0 completion_tokens = 0 cached_tokens = 0 + started = time.monotonic() + first_output_at: float | None = None + last_output_at: float | None = None engine_finish_reason: str | None = None engine_matched_stop: str | None = None async for ack in state.wait_for_ack(uid): @@ -802,12 +849,21 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any) -> GenResult: prompt_tokens += ack.prompt_tokens_delta completion_tokens += ack.completion_tokens_delta cached_tokens += ack.cached_tokens + if ack.completion_tokens_delta > 0: + now = time.monotonic() + if first_output_at is None: + first_output_at = now + last_output_at = now full_content += ack.incremental_output if ack.finished: engine_finish_reason = getattr(ack, "finish_reason", None) engine_matched_stop = getattr(ack, "matched_stop", None) break + ended = time.monotonic() + prompt_ms = ((first_output_at if first_output_at is not None else ended) - started) * 1000.0 + predicted_ms = (last_output_at - first_output_at) * 1000.0 if first_output_at is not None else 0.0 + reasoning_text, content_text = _split_reasoning(full_content, spec, state) # Engine reason ("stop"/"length"); a tool call overrides it, but a truncation (length) wins. finish_reason = engine_finish_reason or "stop" @@ -828,4 +884,6 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any) -> GenResult: completion_tokens=completion_tokens, matched_stop=engine_matched_stop, cached_tokens=cached_tokens, + prompt_ms=prompt_ms, + predicted_ms=predicted_ms, ) diff --git a/python/freetoken/server/openai_api.py b/python/freetoken/server/openai_api.py index 99a4283549..73beca9d77 100644 --- a/python/freetoken/server/openai_api.py +++ b/python/freetoken/server/openai_api.py @@ -37,6 +37,7 @@ prerender_error, render_messages, resolve_sampling, + timings_block, submit_generation, ) @@ -219,7 +220,7 @@ async def handle_chat_completion( if result.tool_calls: message["tool_calls"] = _tool_calls_to_openai(result.tool_calls) - return { + response: dict[str, Any] = { "id": f"chatcmpl-{uid}", "object": "chat.completion", "created": int(time.time()), @@ -237,6 +238,12 @@ async def handle_chat_completion( _reported_cached(state, result.cached_tokens), ), } + if (timings := timings_block( + result.prompt_tokens, result.completion_tokens, result.cached_tokens, + result.prompt_ms, result.predicted_ms, + )) is not None: + response["timings"] = timings + return response async def stream_chat_completion_chunks( @@ -259,6 +266,8 @@ async def stream_chat_completion_chunks( prompt_tokens = 0 completion_tokens = 0 cached_tokens = 0 + prompt_ms = 0.0 + predicted_ms = 0.0 tool_calls_sent = 0 open_tool: dict[str, Any] | None = None events = generate_events(uid, spec, state, source="/v1/chat/completions") @@ -368,21 +377,33 @@ async def stream_chat_completion_chunks( prompt_tokens = ev.prompt_tokens completion_tokens = ev.completion_tokens cached_tokens = ev.cached_tokens - yield _sse(_chat_chunk(req, uid, [{"delta": {}, "index": 0, "finish_reason": ev.finish_reason}])) + prompt_ms = ev.prompt_ms + predicted_ms = ev.predicted_ms + # llama.cpp puts timings on the finish chunk too, so streams without + # include_usage still get them. + done = _chat_chunk(req, uid, [{"delta": {}, "index": 0, "finish_reason": ev.finish_reason}]) + if (timings := timings_block( + prompt_tokens, completion_tokens, cached_tokens, prompt_ms, predicted_ms + )) is not None: + done["timings"] = timings + yield _sse(done) if req.stream_options and req.stream_options.include_usage: - yield _sse( - { - "id": f"chatcmpl-{uid}", - "object": "chat.completion.chunk", - "created": int(time.time()), - "model": req.model, - "choices": [], - "usage": _usage( - prompt_tokens, completion_tokens, _reported_cached(state, cached_tokens) - ), - } - ) + final: dict[str, Any] = { + "id": f"chatcmpl-{uid}", + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": req.model, + "choices": [], + "usage": _usage( + prompt_tokens, completion_tokens, _reported_cached(state, cached_tokens) + ), + } + if (timings := timings_block( + prompt_tokens, completion_tokens, cached_tokens, prompt_ms, predicted_ms + )) is not None: + final["timings"] = timings + yield _sse(final) yield b"data: [DONE]\n\n" @@ -424,6 +445,9 @@ async def handle_completion( prompt_tokens = 0 completion_tokens = 0 cached_tokens = 0 + started = time.monotonic() + first_output_at: float | None = None + last_output_at: float | None = None for index, prompt in enumerate(prompts): uid = state.new_user() await state.send_one( @@ -443,13 +467,19 @@ async def handle_completion( prompt_tokens += ack.prompt_tokens_delta completion_tokens += ack.completion_tokens_delta cached_tokens += ack.cached_tokens + if ack.completion_tokens_delta > 0: + now = time.monotonic() + if first_output_at is None: + first_output_at = now + last_output_at = now text += ack.incremental_output if ack.finished: finish_reason = getattr(ack, "finish_reason", None) or "stop" break choices.append({"index": index, "text": text, "finish_reason": finish_reason, "logprobs": None}) - return { + ended = time.monotonic() + response: dict[str, Any] = { "id": f"cmpl-{uuid.uuid4().hex}", "object": "text_completion", "created": int(time.time()), @@ -457,12 +487,22 @@ async def handle_completion( "choices": choices, "usage": _usage(prompt_tokens, completion_tokens, _reported_cached(state, cached_tokens)), } + if (timings := timings_block( + prompt_tokens, completion_tokens, cached_tokens, + ((first_output_at if first_output_at is not None else ended) - started) * 1000.0, + (last_output_at - first_output_at) * 1000.0 if first_output_at is not None else 0.0, + )) is not None: + response["timings"] = timings + return response async def stream_completion_chunks(uid: int, req: CompletionRequest, state: Any) -> AsyncIterator[bytes]: prompt_tokens = 0 completion_tokens = 0 cached_tokens = 0 + started = time.monotonic() + first_output_at: float | None = None + last_output_at: float | None = None finish_reason = "stop" async for ack in state.wait_for_ack(uid): if getattr(ack, "error", None): @@ -472,6 +512,11 @@ async def stream_completion_chunks(uid: int, req: CompletionRequest, state: Any) prompt_tokens += ack.prompt_tokens_delta completion_tokens += ack.completion_tokens_delta cached_tokens += ack.cached_tokens + if ack.completion_tokens_delta > 0: + now = time.monotonic() + if first_output_at is None: + first_output_at = now + last_output_at = now if ack.incremental_output: yield _sse( { @@ -493,28 +538,40 @@ async def stream_completion_chunks(uid: int, req: CompletionRequest, state: Any) finish_reason = getattr(ack, "finish_reason", None) or "stop" break - yield _sse( - { + ended = time.monotonic() + done_chunk: dict[str, Any] = { + "id": f"cmpl-{uid}", + "object": "text_completion.chunk", + "created": int(time.time()), + "model": req.model, + "choices": [{"text": "", "index": 0, "finish_reason": finish_reason, "logprobs": None}], + } + if (timings := timings_block( + prompt_tokens, completion_tokens, cached_tokens, + ((first_output_at if first_output_at is not None else ended) - started) * 1000.0, + (last_output_at - first_output_at) * 1000.0 if first_output_at is not None else 0.0, + )) is not None: + done_chunk["timings"] = timings + yield _sse(done_chunk) + if req.stream_options and req.stream_options.include_usage: + ended = time.monotonic() + final: dict[str, Any] = { "id": f"cmpl-{uid}", "object": "text_completion.chunk", "created": int(time.time()), "model": req.model, - "choices": [{"text": "", "index": 0, "finish_reason": finish_reason, "logprobs": None}], + "choices": [], + "usage": _usage( + prompt_tokens, completion_tokens, _reported_cached(state, cached_tokens) + ), } - ) - if req.stream_options and req.stream_options.include_usage: - yield _sse( - { - "id": f"cmpl-{uid}", - "object": "text_completion.chunk", - "created": int(time.time()), - "model": req.model, - "choices": [], - "usage": _usage( - prompt_tokens, completion_tokens, _reported_cached(state, cached_tokens) - ), - } - ) + if (timings := timings_block( + prompt_tokens, completion_tokens, cached_tokens, + ((first_output_at if first_output_at is not None else ended) - started) * 1000.0, + (last_output_at - first_output_at) * 1000.0 if first_output_at is not None else 0.0, + )) is not None: + final["timings"] = timings + yield _sse(final) yield b"data: [DONE]\n\n" diff --git a/python/freetoken/server/responses_api.py b/python/freetoken/server/responses_api.py index 71d6eed873..7ae5b4cd15 100644 --- a/python/freetoken/server/responses_api.py +++ b/python/freetoken/server/responses_api.py @@ -75,6 +75,7 @@ resolve_sampling, split_tool_lists, submit_generation, + timings_block, with_keepalive, ) from .request_logger import log_request @@ -454,14 +455,20 @@ def build_responses_response( result.cached_tokens if cache_report else 0, ), incomplete_reason="max_output_tokens" if truncated else None, + timings=timings_block( + result.prompt_tokens, result.completion_tokens, result.cached_tokens, + result.prompt_ms, result.predicted_ms, + ), ) def _response_obj( response_id: str, created: int, model: str, output: list[Any], *, status: str, usage: ResponseUsage | None, error: ResponseError | None = None, - incomplete_reason: str | None = None, + incomplete_reason: str | None = None, timings: dict[str, Any] | None = None, ) -> Response: + # Response allows extra fields, so timings rides along without forking the schema. + extra: dict[str, Any] = {"timings": timings} if timings is not None else {} return Response( id=response_id, created_at=created, @@ -475,6 +482,7 @@ def _response_obj( parallel_tool_calls=True, tool_choice="auto", tools=[], + **extra, ) @@ -502,10 +510,10 @@ async def responses_stream_generator( ) -> AsyncIterator[str]: seq = _Seq() - def snapshot(status, output, usage=None, incomplete_reason=None): + def snapshot(status, output, usage=None, incomplete_reason=None, timings=None): return _response_obj( response_id, created, req.model, output, - status=status, usage=usage, incomplete_reason=incomplete_reason, + status=status, usage=usage, incomplete_reason=incomplete_reason, timings=timings, ) yield _sse(ResponseCreatedEvent( @@ -522,6 +530,8 @@ def snapshot(status, output, usage=None, incomplete_reason=None): usage_pt = 0 usage_ct = 0 usage_cached = 0 + usage_prompt_ms = 0.0 + usage_predicted_ms = 0.0 def close_current() -> list[str]: nonlocal current, output_index @@ -707,6 +717,8 @@ def args_delta_frame(fragment: str) -> str: finish_reason = ev.finish_reason usage_pt, usage_ct = ev.prompt_tokens, ev.completion_tokens usage_cached = ev.cached_tokens if cache_report else 0 + usage_prompt_ms, usage_predicted_ms = ev.prompt_ms, ev.predicted_ms + final_timings = timings_block(usage_pt, usage_ct, ev.cached_tokens, usage_prompt_ms, usage_predicted_ms) for f in close_current(): yield f if finish_reason == "length": @@ -716,6 +728,7 @@ def args_delta_frame(fragment: str) -> str: "incomplete", output_items, usage=_usage(usage_pt, usage_ct, usage_cached), incomplete_reason="max_output_tokens", + timings=final_timings, ), )) else: @@ -724,6 +737,7 @@ def args_delta_frame(fragment: str) -> str: response=snapshot( "completed", output_items, usage=_usage(usage_pt, usage_ct, usage_cached), + timings=final_timings, ), )) except GenerationError as exc: diff --git a/tests/server/test_generation_accounting.py b/tests/server/test_generation_accounting.py index a6d2bdb916..96cfc397be 100644 --- a/tests/server/test_generation_accounting.py +++ b/tests/server/test_generation_accounting.py @@ -24,6 +24,7 @@ GenSpec, generate_events, generate_full, + timings_block, ) @@ -206,3 +207,46 @@ def test_ttft_mean_is_zero_without_samples(): request_ring.reset() request_ring.record_request(_row(ttft_ms=None)) assert request_ring.requests_ttft_mean_ms() == 0 + + +def test_timings_block_follows_llama_cpp_semantics(): + t = timings_block(prompt_tokens=1000, completion_tokens=21, cached_tokens=800, + prompt_ms=2500.0, predicted_ms=2000.0) + # prompt_n excludes the cache; decode steps are n - 1 (llama.cpp semantics). + assert t["cache_n"] == 800 + assert t["prompt_n"] == 200 + assert t["prompt_per_second"] == 80.0 # 200 evaluated tokens / 2.5 s + assert t["predicted_n"] == 21 + assert t["predicted_per_second"] == 10.0 # 20 decode steps / 2.0 s + assert t["prompt_ms"] == 2500.0 and t["predicted_ms"] == 2000.0 + + +def test_timings_block_rates_survive_the_edge_shapes(): + assert timings_block(10, 0, 0, 0.0, 0.0) is None # nothing was measured + t = timings_block(10, 1, 0, 100.0, 0.0) # one token: no decode steps + assert t["predicted_per_second"] == 0.0 and t["predicted_per_token_ms"] == 0.0 + t = timings_block(0, 2, 0, 100.0, 50.0) # fully cached / empty prompt + assert t["prompt_per_second"] == 0.0 and t["prompt_per_token_ms"] == 0.0 + + +def test_gen_done_and_result_carry_measured_spans(): + st = FakeState([_ack(prompt=7, completion=1, out="x"), _ack(completion=4, out="yz", finished=True)]) + + async def drain(): + done = None + async for ev in generate_events(42, _spec(), st): + if isinstance(ev, GenDone): + done = ev + return done + + done = asyncio.run(drain()) + assert done.prompt_ms > 0.0 + assert done.predicted_ms >= 0.0 + assert timings_block(done.prompt_tokens, done.completion_tokens, done.cached_tokens, + done.prompt_ms, done.predicted_ms) is not None + + result = asyncio.run(generate_full(42, _spec(), FakeState( + [_ack(prompt=5, completion=1, out="a"), _ack(completion=2, out="bc", finished=True)] + ))) + assert result.prompt_ms > 0.0 + assert result.predicted_ms >= 0.0