From a76d9ecc1f5043aa3c8eebd5880eb0334ddbe80b Mon Sep 17 00:00:00 2001 From: yajun Date: Thu, 17 Sep 2026 16:42:16 +0800 Subject: [PATCH] feat(server): serve per-request inference metrics under --enable-metrics-report - the scheduler measures the prefill span itself (admission to the token sampled off the request's last prefill chunk) and ships it as a duration on that token's DetokenizeMsg: it is the one boundary a client timing SSE frames cannot see, and a duration keeps the wire free of any cross-process clock assumption - the remaining timings come off the raw ack stream, not the emitted GenEvents: a reasoning or tool-call parser holds text back, sometimes to the end of the stream, so an event-timed TTFT runs long, and acks are the one path the streaming and buffered generators share - cached_prompt_tokens reports the real prefix-cache hit whether or not --enable-cache-report is set; that flag governs the billing fields in usage, and gating the hit here would leave prefill_tokens_per_second computed over tokens that were never forwarded - decode_tokens_per_second divides by decode_tokens - 1, as vLLM's and sglang's serving benchmarks do: the first token falls out of prefill and spans no decode interval - served on /v1/chat/completions, /v1/messages and /v1/responses; /v1/completions does not go through the shared generation core and is left out Closes #503 --- docs/cli.md | 56 ++++++++ python/freetoken/message/frontend.py | 3 + python/freetoken/message/tokenizer.py | 5 + python/freetoken/scheduler/scheduler.py | 22 ++++ python/freetoken/server/anthropic_api.py | 26 +++- python/freetoken/server/anthropic_models.py | 6 + python/freetoken/server/args.py | 20 +++ python/freetoken/server/generation.py | 128 +++++++++++++++++-- python/freetoken/server/openai_api.py | 45 +++++-- python/freetoken/server/responses_api.py | 37 +++++- python/freetoken/tokenizer/server.py | 1 + tests/scheduler/test_cost_accounting_core.py | 17 +++ tests/server/test_anthropic_api.py | 56 +++++++- tests/server/test_generation_accounting.py | 113 +++++++++++++++- tests/server/test_openai_api.py | 62 +++++++++ tests/server/test_responses_api.py | 41 +++++- 16 files changed, 599 insertions(+), 39 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 0c3394398..ab4429e0d 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -98,6 +98,62 @@ See [models.md](models.md#moe-strategies) for what each strategy does. | `--tool-call-parser` | auto | Tool-call format; auto-inferred from the model family | | `--reasoning-parser` | auto | Splits chain-of-thought into `reasoning_content`; auto-inferred; `off` disables | | `--enable-cache-report` | off | Report prefix-cache hits in each response's usage block | +| `--enable-metrics-report` | off | Serve a per-request `metrics` object next to usage ([below](#per-request-performance-metrics)) | + +### Per-request performance metrics + +`--enable-metrics-report` adds a `metrics` object to each response on `/v1/chat/completions`, +`/v1/messages` and `/v1/responses`. It is not part of any of those protocols, which is why it is +off by default; clients that do not read it are unaffected. + +```json +{ + "usage": { "prompt_tokens": 8192, "completion_tokens": 256, "total_tokens": 8448 }, + "metrics": { + "ttft_ms": 320.5, + "prefill_tokens": 2048, + "cached_prompt_tokens": 6144, + "prefill_time_ms": 410.0, + "prefill_tokens_per_second": 4995.12, + "decode_tokens": 256, + "decode_time_ms": 5120.8, + "decode_tokens_per_second": 49.8, + "total_time_ms": 5480.2 + } +} +``` + +| Field | Meaning | +| --- | --- | +| `ttft_ms` | Time to the first generated token, from the point the request entered the generation path | +| `prefill_tokens` | Prompt tokens actually forwarded: `prompt_tokens - cached_prompt_tokens` | +| `cached_prompt_tokens` | Prompt tokens served from the prefix cache instead of recomputed | +| `prefill_time_ms` | The scheduler's own prefill span: admission to the token sampled off the last prefill chunk | +| `prefill_tokens_per_second` | `prefill_tokens / prefill_time_ms` | +| `decode_tokens` | Generated tokens (same as `usage.completion_tokens`) | +| `decode_time_ms` | First generated token to the last | +| `decode_tokens_per_second` | `(decode_tokens - 1) / decode_time_ms`; the first token comes out of prefill, so it spans no decode interval | +| `total_time_ms` | Request latency up to the terminal engine reply, excluding HTTP framing | + +Reading the numbers: + +- `prefill_time_ms` is measured inside the scheduler, which is the point of the flag: a client + timing SSE frames cannot see when prefill starts or ends. The others are measured at the API + layer and carry the same IPC hops a client's own timestamps would. +- `ttft_ms` is larger than `prefill_time_ms` by the time the request spent queued before + admission. Under load that gap is the queue, not the model. +- Every span is **this request's share of shared work**. Its prefill chunk is co-scheduled with + other prompts and its decode steps are batched with other requests, so the throughputs + describe this request under that load, not isolated engine benchmarks. For those, + send one request at a time. +- `cached_prompt_tokens` here always reports the real prefix-cache hit, whether or not + `--enable-cache-report` is set; that flag governs the billing fields in `usage`. Without it + `prefill_tokens_per_second` would be computed over tokens that were never forwarded. + +Streaming responses carry `metrics` on the same final message as usage, so the request has to ask +for usage too: `stream_options: {"include_usage": true}` on `/v1/chat/completions`. On +`/v1/messages` it rides `message_delta`; on `/v1/responses`, `response.completed`. +`/v1/completions` does not serve `metrics`: it does not go through the shared generation core. ### Image input diff --git a/python/freetoken/message/frontend.py b/python/freetoken/message/frontend.py index 24725567b..392faea10 100644 --- a/python/freetoken/message/frontend.py +++ b/python/freetoken/message/frontend.py @@ -45,6 +45,9 @@ class UserReply(BaseFrontendMsg): swa_total_tokens: int = 0 # Bytes the engine process holds on the GPU (torch reserved pool). 0 when not reported. gpu_mem_bytes: int = 0 + # Scheduler-measured prefill span (see DetokenizeMsg.prefill_ms). Arrives once, on the + # reply carrying the request's first generated token; 0.0 on every other reply. + prefill_ms: float = 0.0 # Set (with finished=True) when a request failed before producing output — e.g. a chat # template that the tokenizer cannot render, or a prompt that exceeds the KV budget the # scheduler can serve. Carries a human-readable reason. Without this, such a request would diff --git a/python/freetoken/message/tokenizer.py b/python/freetoken/message/tokenizer.py index 9442ee059..f0336f73d 100644 --- a/python/freetoken/message/tokenizer.py +++ b/python/freetoken/message/tokenizer.py @@ -48,6 +48,11 @@ class DetokenizeMsg(BaseTokenizerMsg): swa_total_tokens: int = 0 # Bytes this engine process holds on the GPU (torch reserved pool). 0 on CPU. gpu_mem_bytes: int = 0 + # Scheduler-measured prefill span for this request: admission (the first prefill batch + # was prepared) to the token sampled off its last prefill chunk. Set on that first token + # only, 0.0 on every later one -- it is the one span a client cannot derive from HTTP + # timestamps, which is why it rides the wire instead of being re-estimated frontend-side. + prefill_ms: float = 0.0 @dataclass diff --git a/python/freetoken/scheduler/scheduler.py b/python/freetoken/scheduler/scheduler.py index ac1bf322e..88b455874 100644 --- a/python/freetoken/scheduler/scheduler.py +++ b/python/freetoken/scheduler/scheduler.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import time from typing import TYPE_CHECKING, List, NamedTuple, NoReturn, Set, Tuple, TypeAlias @@ -111,6 +112,11 @@ def __init__(self, config: SchedulerConfig): # tombstone so an abort-before-admission request can never be resurrected after its # terminal accounting acknowledgement has already been published. self._abort_tombstones: dict[int, None] = {} + # uid -> monotonic clock at admission, for the per-request prefill span shipped on the + # first sampled token. An entry existing IS the "this is the first token" test, so it is + # popped there; _free_req_resources pops too, so a request aborted mid-prefill (which + # never samples) cannot leave one behind. + self._prefill_start: dict[int, float] = {} self._forward_iter = 0 # global forward counter; drives the SWA proactive-eviction cadence # The launched-but-not-yet-drained batch (overlap): set at the top of each overlap_loop # iteration so the abort handler can tell whether a request's forward is still in flight @@ -370,6 +376,10 @@ def _process_last_data(self, last_data: ForwardData | None) -> None: and not finished ): req.toolcall_anchor_len = req.input_ids.numel() + # Present only until this request's first token is sampled, i.e. off its last + # prefill chunk -- so popping it both ends the prefill span and marks this as + # the one reply that carries it. + started_at = self._prefill_start.pop(req.uid, None) reply.append( DetokenizeMsg( uid=req.uid, @@ -378,6 +388,10 @@ def _process_last_data(self, last_data: ForwardData | None) -> None: finish_reason=finish_reason, matched_stop=matched_stop, stop_strs=req.sampling_params.stop_strs or None, + prefill_ms=( + 0.0 if started_at is None + else (time.monotonic() - started_at) * 1000.0 + ), ) ) @@ -631,6 +645,9 @@ def _free_req_resources(self, req: Req) -> None: # slots to two later requests. table_idx == -1 marks an already-freed request. if req.table_idx == -1: return + # A request freed without ever sampling (aborted mid-prefill) would otherwise keep its + # admission timestamp forever; a normally-finished one popped it at its first token. + self._prefill_start.pop(req.uid, None) # Polymorphic free: the DSV4 manager returns the request's window pages + cmp/idx blocks # to their tier free-lists; the generic manager frees its KV pages (it reads # page_table[req.table_idx], so free the table entry after). @@ -886,6 +903,11 @@ def _report_prompt_admissions(self, batch: Batch) -> None: """ if not batch.is_prefill or not batch.prompt_admissions: return + # Before the forward, which is what makes this the prefill START: _schedule_next_batch + # has prepared the batch but not run it. + now = time.monotonic() + for uid, _, _ in batch.prompt_admissions: + self._prefill_start[uid] = now self.send_result( [ PromptAdmittedMsg(uid=uid, prompt_tokens=prompt_tokens, cached_tokens=cached_tokens) diff --git a/python/freetoken/server/anthropic_api.py b/python/freetoken/server/anthropic_api.py index 0d60940c9..883c285bf 100644 --- a/python/freetoken/server/anthropic_api.py +++ b/python/freetoken/server/anthropic_api.py @@ -44,9 +44,11 @@ ToolCallArgsDelta, ToolCallsDelta, ToolCallStart, + build_metrics, count_prompt_tokens, generate_events, generate_full, + metrics_enabled, render_messages, resolve_sampling, split_tool_lists, @@ -126,10 +128,11 @@ async def handle_anthropic_messages( return _anthropic_error_response(400, "invalid_request_error", str(exc)) cache_report = getattr(state.config, "enable_cache_report", False) + metrics = metrics_enabled(state) if req.stream: events = anthropic_event_stream( generate_events(uid, spec, state, source="/v1/messages"), - req.model, uid, cache_report=cache_report, + req.model, uid, cache_report=cache_report, metrics=metrics, ) if request is not None: events = state.stream_with_cancellation(events, request, uid) @@ -139,7 +142,9 @@ async def handle_anthropic_messages( result = await generate_full(uid, spec, state, source="/v1/messages") except GenerationError as exc: return _anthropic_error_response(400, "invalid_request_error", str(exc)) - response = anthropic_full_response(result, req.model, uid, cache_report=cache_report) + response = anthropic_full_response( + result, req.model, uid, cache_report=cache_report, metrics=metrics + ) return JSONResponse(content=response.model_dump(exclude_none=True)) @@ -375,7 +380,7 @@ def _tool_result_parts(content) -> tuple[str, list[dict[str, Any]]]: # Output formatting: GenResult / GenEvent -> Anthropic response / events # --------------------------------------------------------------------------- # def anthropic_full_response( - result: GenResult, model: str, uid: int, cache_report: bool = False + result: GenResult, model: str, uid: int, cache_report: bool = False, metrics: bool = False ) -> AnthropicMessagesResponse: content: list[AnthropicContentBlock] = [] if result.reasoning: @@ -402,6 +407,12 @@ def anthropic_full_response( usage=_anthropic_usage( result.prompt_tokens, result.completion_tokens, result.cached_tokens, cache_report ), + metrics=build_metrics( + prompt_tokens=result.prompt_tokens, + completion_tokens=result.completion_tokens, + cached_tokens=result.cached_tokens, + timings=result.timings, + ) if metrics else None, ) @@ -420,7 +431,8 @@ def _anthropic_usage( async def anthropic_event_stream( - events: AsyncIterator[Any], model: str, uid: int, cache_report: bool = False + events: AsyncIterator[Any], model: str, uid: int, cache_report: bool = False, + metrics: bool = False, ) -> AsyncIterator[str]: """Format the protocol-neutral GenEvent stream into Anthropic SSE events. @@ -576,6 +588,12 @@ def _tool_args_delta(fragment: str) -> str: usage=_anthropic_usage( ev.prompt_tokens, ev.completion_tokens, ev.cached_tokens, cache_report ), + metrics=build_metrics( + prompt_tokens=ev.prompt_tokens, + completion_tokens=ev.completion_tokens, + cached_tokens=ev.cached_tokens, + timings=ev.timings, + ) if metrics else None, )) yield _event(AnthropicStreamEvent(type="message_stop")) # Anthropic streams terminate on message_stop — no OpenAI-style diff --git a/python/freetoken/server/anthropic_models.py b/python/freetoken/server/anthropic_models.py index 42ab545be..2a6232ee6 100644 --- a/python/freetoken/server/anthropic_models.py +++ b/python/freetoken/server/anthropic_models.py @@ -153,6 +153,10 @@ class AnthropicMessagesResponse(BaseModel): ) = None stop_sequence: str | None = None usage: AnthropicUsage | None = None + # FreeToken extension, served only under --enable-metrics-report: per-request inference + # timings. Anthropic's wire has no such field, so it stays absent by default rather than + # being emitted as null -- both response paths dump with exclude_none. + metrics: dict[str, Any] | None = None def model_post_init(self, __context): if not self.id: @@ -176,3 +180,5 @@ class AnthropicStreamEvent(BaseModel): index: int | None = None error: AnthropicError | None = None usage: AnthropicUsage | None = None + # Rides message_delta next to usage, under --enable-metrics-report only. + metrics: dict[str, Any] | None = None diff --git a/python/freetoken/server/args.py b/python/freetoken/server/args.py index ab2fb9b74..5b37b6b55 100644 --- a/python/freetoken/server/args.py +++ b/python/freetoken/server/args.py @@ -59,6 +59,10 @@ class ServerArgs(SchedulerConfig): # prompt_tokens_details.cached_tokens, Anthropic cache_read_input_tokens, Responses # input_tokens_details.cached_tokens). Mirrors sglang's --enable-cache-report. enable_cache_report: bool = False + # Serve a per-request `metrics` object (TTFT, prefill/decode times and throughputs, + # prefix-cache hit) alongside usage. Off by default: it is a non-standard field on + # every protocol we speak. + enable_metrics_report: bool = False # Comma-separated hostname allowlist for client-supplied image URLs; empty admits any domain. allowed_media_domains: str = "" # Directory file:// image refs may be read from; empty rejects local files. @@ -524,6 +528,22 @@ def _infer_reasoning_parser(model_path: str) -> str | None: ), ) + parser.add_argument( + "--enable-metrics-report", + action="store_true", + default=ServerArgs.enable_metrics_report, + help=( + "Serve a per-request `metrics` object next to usage on /v1/chat/completions, " + "/v1/completions, /v1/messages and /v1/responses: ttft_ms, prefill_time_ms and " + "prefill_tokens_per_second (the prefill span measured by the scheduler itself), " + "decode_time_ms and decode_tokens_per_second, cached_prompt_tokens and " + "total_time_ms. Streaming responses carry it on the same final chunk as usage, so " + "the request must also ask for usage (OpenAI stream_options.include_usage). " + "Non-standard on every protocol, hence opt-in. Under concurrency the spans are " + "this request's share of shared batches, not isolated engine throughput." + ), + ) + parser.add_argument( "--sampling-defaults", type=str, diff --git a/python/freetoken/server/generation.py b/python/freetoken/server/generation.py index 91056e8d1..108b666b6 100644 --- a/python/freetoken/server/generation.py +++ b/python/freetoken/server/generation.py @@ -107,6 +107,26 @@ class ToolCallsDelta: calls: list[ToolCallItem] +@dataclass +class GenTimings: + """Per-request inference timings, in milliseconds. + + ``prefill_ms`` is the scheduler's own span (admission to the token sampled off the + request's last prefill chunk), shipped over the wire because it is the one boundary a + client cannot see; the rest are measured here, around the engine ack loop, so they carry + the same IPC hops the client's own HTTP timestamps would. + + Under concurrency every span is this request's SHARE of shared work -- its prefill chunk + is co-scheduled with other prompts, its decode steps batched with other requests -- so + the derived rates describe this request under that load, not isolated engine throughput. + """ + + ttft_ms: float = 0.0 + prefill_ms: float = 0.0 + decode_ms: float = 0.0 + total_ms: float = 0.0 + + @dataclass class GenDone: finish_reason: str @@ -114,6 +134,7 @@ class GenDone: completion_tokens: int matched_stop: str | None = None cached_tokens: int = 0 + timings: GenTimings = field(default_factory=GenTimings) GenEvent = ReasoningDelta | ContentDelta | ToolCallStart | ToolCallArgsDelta | ToolCallsDelta | GenDone @@ -129,6 +150,7 @@ class GenResult: completion_tokens: int matched_stop: str | None = None cached_tokens: int = 0 + timings: GenTimings = field(default_factory=GenTimings) @dataclass @@ -491,6 +513,83 @@ async def with_keepalive(events: AsyncIterator[GenEvent], interval: float): task.cancel() +class _Timer: + """Times one generation off the raw ack stream. + + Deliberately fed from acks rather than from emitted GenEvents: a reasoning or tool-call + parser holds text back, sometimes to the end of the stream, so the first EVENT can lag the + first generated token by a lot. Acks are also the one path both the streaming and the + buffered generator share, so TTFT means the same thing on either. + """ + + def __init__(self) -> None: + self.start = time.monotonic() + self.first_token_at: float | None = None + self.end: float | None = None + self.prefill_ms = 0.0 + + def observe(self, ack: Any) -> None: + if ack.prefill_ms: + self.prefill_ms = ack.prefill_ms + # completion_tokens_delta, not incremental_output: the detokenizer holds back a + # trailing partial-stop prefix, so a real generated token can arrive with empty text. + if self.first_token_at is None and ack.completion_tokens_delta: + self.first_token_at = time.monotonic() + if ack.finished: + # Stamped here, not in finish(): the callers run the reasoning split and the + # tool-call drain between the terminal ack and building their result, and that + # post-processing is not inference time. + self.end = time.monotonic() + + def finish(self) -> GenTimings: + # A stream cut short (client disconnect) never saw a terminal ack; time it to here. + end = self.end if self.end is not None else time.monotonic() + first = self.first_token_at + return GenTimings( + ttft_ms=0.0 if first is None else (first - self.start) * 1000.0, + prefill_ms=self.prefill_ms, + decode_ms=0.0 if first is None else (end - first) * 1000.0, + total_ms=(end - self.start) * 1000.0, + ) + + +def build_metrics( + *, prompt_tokens: int, completion_tokens: int, cached_tokens: int, timings: GenTimings +) -> dict[str, Any]: + """The `metrics` object served under --enable-metrics-report, identical on every protocol. + + ``cached_prompt_tokens`` is the real prefix-cache hit whatever --enable-cache-report says: + that flag governs BILLING fields in `usage`, and gating the hit here instead would leave + prefill_tokens_per_second silently computed over tokens that were never forwarded. + + Decode throughput divides by completion_tokens - 1, not completion_tokens: the first token + falls out of prefill, so decode_time_ms spans only the intervals after it (the convention + vLLM's and sglang's serving benchmarks use). + """ + prefill_tokens = max(prompt_tokens - cached_tokens, 0) + prefill_s = timings.prefill_ms / 1000.0 + decode_s = timings.decode_ms / 1000.0 + return { + "ttft_ms": round(timings.ttft_ms, 3), + "prefill_tokens": prefill_tokens, + "cached_prompt_tokens": cached_tokens, + "prefill_time_ms": round(timings.prefill_ms, 3), + "prefill_tokens_per_second": round(prefill_tokens / prefill_s, 2) if prefill_s > 0 else 0.0, + "decode_tokens": completion_tokens, + "decode_time_ms": round(timings.decode_ms, 3), + "decode_tokens_per_second": ( + round((completion_tokens - 1) / decode_s, 2) + if decode_s > 0 and completion_tokens > 1 + else 0.0 + ), + "total_time_ms": round(timings.total_ms, 3), + } + + +def metrics_enabled(state: Any) -> bool: + return bool(getattr(state.config, "enable_metrics_report", False)) + + def _record_generation( *, source: str | None, @@ -532,27 +631,24 @@ async def generate_events( """Wraps `_generate_events_impl` to log the request with its totals, read off the terminal `GenDone`. The `finally` still records the row on a mid-stream disconnect — but with 0 tokens if the drop lands before `GenDone`, the only event carrying the totals.""" - start = time.monotonic() + timer = _Timer() prompt_tokens = 0 completion_tokens = 0 - first_token_at: float | None = None error: str | None = None try: - async for ev in _generate_events_impl(uid, spec, state): + async for ev in _generate_events_impl(uid, spec, state, timer): if isinstance(ev, GenDone): prompt_tokens = ev.prompt_tokens completion_tokens = ev.completion_tokens - elif first_token_at is None: - first_token_at = time.monotonic() yield ev except GenerationError as exc: error = str(exc) raise finally: _record_generation( - source=source, stream=True, start=start, + source=source, stream=True, start=timer.start, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, error=error, - first_token_at=first_token_at, + first_token_at=timer.first_token_at, ) @@ -561,25 +657,29 @@ async def generate_full( ) -> GenResult: """Wraps `_generate_full_impl` to log the request with its totals; the `finally` also records a `GenerationError` as a failed row.""" - start = time.monotonic() + timer = _Timer() result: GenResult | None = None error: str | None = None try: - result = await _generate_full_impl(uid, spec, state) + result = await _generate_full_impl(uid, spec, state, timer) return result except GenerationError as exc: error = str(exc) raise finally: _record_generation( - source=source, stream=False, start=start, + source=source, stream=False, start=timer.start, prompt_tokens=result.prompt_tokens if result else 0, completion_tokens=result.completion_tokens if result else 0, error=error, + # No first_token_at: non-streaming rows deliberately carry no TTFT, because + # requests_ttft_mean_ms averages only the rows that have one. `metrics` still does. ) -async def _generate_events_impl(uid: int, spec: GenSpec, state: Any) -> AsyncIterator[GenEvent]: +async def _generate_events_impl( + uid: int, spec: GenSpec, state: Any, timer: _Timer +) -> AsyncIterator[GenEvent]: """Protocol-neutral streaming generation. Yields semantic events (reasoning / content / tool-call deltas) terminated by exactly one GenDone. Produces no wire format — the OpenAI/Anthropic/Responses streamers format these into their own. @@ -686,6 +786,7 @@ def _route_tool_text(piece: str) -> list[GenEvent]: async for ack in state.wait_for_ack(uid): if getattr(ack, "error", None): raise GenerationError(ack.error, getattr(ack, "error_code", None)) + timer.observe(ack) prompt_tokens += ack.prompt_tokens_delta completion_tokens += ack.completion_tokens_delta cached_tokens += ack.cached_tokens @@ -771,10 +872,11 @@ def _route_tool_text(piece: str) -> list[GenEvent]: yield GenDone( finish_reason, prompt_tokens, completion_tokens, matched_stop=engine_matched_stop, cached_tokens=cached_tokens, + timings=timer.finish(), ) -async def _generate_full_impl(uid: int, spec: GenSpec, state: Any) -> GenResult: +async def _generate_full_impl(uid: int, spec: GenSpec, state: Any, timer: _Timer) -> GenResult: """Protocol-neutral non-streaming generation: accumulate, split reasoning, parse tool calls, strip special tokens. The adapters format the GenResult into their wire.""" full_content = "" @@ -786,6 +888,7 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any) -> GenResult: async for ack in state.wait_for_ack(uid): if getattr(ack, "error", None): raise GenerationError(ack.error, getattr(ack, "error_code", None)) + timer.observe(ack) prompt_tokens += ack.prompt_tokens_delta completion_tokens += ack.completion_tokens_delta cached_tokens += ack.cached_tokens @@ -815,4 +918,5 @@ 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, + timings=timer.finish(), ) diff --git a/python/freetoken/server/openai_api.py b/python/freetoken/server/openai_api.py index dc2f73a97..4b08bc83a 100644 --- a/python/freetoken/server/openai_api.py +++ b/python/freetoken/server/openai_api.py @@ -28,12 +28,15 @@ GenDone, GenerationError, GenSpec, + GenTimings, ReasoningDelta, ToolCallArgsDelta, ToolCallsDelta, ToolCallStart, + build_metrics, generate_events, generate_full, + metrics_enabled, prerender_error, render_messages, resolve_sampling, @@ -217,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()), @@ -235,6 +238,14 @@ async def handle_chat_completion( _reported_cached(state, result.cached_tokens), ), } + if metrics_enabled(state): + response["metrics"] = build_metrics( + prompt_tokens=result.prompt_tokens, + completion_tokens=result.completion_tokens, + cached_tokens=result.cached_tokens, + timings=result.timings, + ) + return response async def stream_chat_completion_chunks( @@ -257,6 +268,7 @@ async def stream_chat_completion_chunks( prompt_tokens = 0 completion_tokens = 0 cached_tokens = 0 + timings = GenTimings() tool_calls_sent = 0 open_tool: dict[str, Any] | None = None events = generate_events(uid, spec, state, source="/v1/chat/completions") @@ -366,21 +378,28 @@ async def stream_chat_completion_chunks( prompt_tokens = ev.prompt_tokens completion_tokens = ev.completion_tokens cached_tokens = ev.cached_tokens + timings = ev.timings yield _sse(_chat_chunk(req, uid, [{"delta": {}, "index": 0, "finish_reason": ev.finish_reason}])) 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 metrics_enabled(state): + final["metrics"] = build_metrics( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + cached_tokens=cached_tokens, + 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 71d6eed87..e333fb70e 100644 --- a/python/freetoken/server/responses_api.py +++ b/python/freetoken/server/responses_api.py @@ -69,8 +69,10 @@ ToolCallArgsDelta, ToolCallsDelta, ToolCallStart, + build_metrics, generate_events, generate_full, + metrics_enabled, render_messages, resolve_sampling, split_tool_lists, @@ -163,10 +165,11 @@ async def handle_responses( return _error_response(400, str(exc)) cache_report = getattr(state.config, "enable_cache_report", False) + metrics = metrics_enabled(state) if req.stream: events = responses_stream_generator( generate_events(uid, spec, state, source="/v1/responses"), req, response_id, created, - cache_report=cache_report, + cache_report=cache_report, metrics=metrics, ) if request is not None: events = state.stream_with_cancellation(events, request, uid) @@ -176,7 +179,9 @@ async def handle_responses( result = await generate_full(uid, spec, state, source="/v1/responses") except GenerationError as exc: return _error_response(400, str(exc), exc.code) - response = build_responses_response(result, req, response_id, created, cache_report=cache_report) + response = build_responses_response( + result, req, response_id, created, cache_report=cache_report, metrics=metrics + ) return JSONResponse(content=response.model_dump(mode="json")) @@ -410,6 +415,7 @@ def build_responses_response( response_id: str, created: int, cache_report: bool = False, + metrics: bool = False, ) -> Response: truncated = result.finish_reason == "length" item_status = "incomplete" if truncated else "completed" @@ -454,15 +460,21 @@ def build_responses_response( result.cached_tokens if cache_report else 0, ), incomplete_reason="max_output_tokens" if truncated else None, + metrics=build_metrics( + prompt_tokens=result.prompt_tokens, + completion_tokens=result.completion_tokens, + cached_tokens=result.cached_tokens, + timings=result.timings, + ) if metrics else None, ) 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, metrics: dict[str, Any] | None = None, ) -> Response: - return Response( + response = Response( id=response_id, created_at=created, model=model, @@ -476,6 +488,11 @@ def _response_obj( tool_choice="auto", tools=[], ) + if metrics is not None: + # The SDK model allows extras, so the FreeToken-only `metrics` object rides the real + # Response type rather than forcing a parallel dict-shaped response path. + response.metrics = metrics + return response def _usage(prompt_tokens: int, completion_tokens: int, cached_tokens: int = 0) -> ResponseUsage: @@ -499,13 +516,15 @@ async def responses_stream_generator( response_id: str, created: int, cache_report: bool = False, + metrics: bool = False, ) -> AsyncIterator[str]: seq = _Seq() - def snapshot(status, output, usage=None, incomplete_reason=None): + def snapshot(status, output, usage=None, incomplete_reason=None, request_metrics=None): return _response_obj( response_id, created, req.model, output, status=status, usage=usage, incomplete_reason=incomplete_reason, + metrics=request_metrics, ) yield _sse(ResponseCreatedEvent( @@ -707,6 +726,12 @@ 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 + request_metrics = build_metrics( + prompt_tokens=ev.prompt_tokens, + completion_tokens=ev.completion_tokens, + cached_tokens=ev.cached_tokens, + timings=ev.timings, + ) if metrics else None for f in close_current(): yield f if finish_reason == "length": @@ -716,6 +741,7 @@ def args_delta_frame(fragment: str) -> str: "incomplete", output_items, usage=_usage(usage_pt, usage_ct, usage_cached), incomplete_reason="max_output_tokens", + request_metrics=request_metrics, ), )) else: @@ -724,6 +750,7 @@ def args_delta_frame(fragment: str) -> str: response=snapshot( "completed", output_items, usage=_usage(usage_pt, usage_ct, usage_cached), + request_metrics=request_metrics, ), )) except GenerationError as exc: diff --git a/python/freetoken/tokenizer/server.py b/python/freetoken/tokenizer/server.py index d3365bfec..8182c4551 100644 --- a/python/freetoken/tokenizer/server.py +++ b/python/freetoken/tokenizer/server.py @@ -225,6 +225,7 @@ def tokenize_worker( swa_used_tokens=msg.swa_used_tokens, swa_total_tokens=msg.swa_total_tokens, gpu_mem_bytes=msg.gpu_mem_bytes, + prefill_ms=msg.prefill_ms, ) for msg, reply in zip(detokenize_msg, replies, strict=True) ] diff --git a/tests/scheduler/test_cost_accounting_core.py b/tests/scheduler/test_cost_accounting_core.py index 045010c6b..37df2b79a 100644 --- a/tests/scheduler/test_cost_accounting_core.py +++ b/tests/scheduler/test_cost_accounting_core.py @@ -94,6 +94,7 @@ def test_schedule_reports_admission_only_after_prepare_succeeds(): scheduler.prefill_budget = 99 scheduler.prefill_manager = SimpleNamespace(schedule_next_batch=lambda budget: batch) scheduler.decode_manager = SimpleNamespace(schedule_next_batch=lambda: None) + scheduler._prefill_start = {} events = [] def prepare(value): @@ -110,6 +111,8 @@ def send(messages): assert events[0] == ("prepared", batch) sent = events[1][1] assert [(m.uid, m.prompt_tokens, m.cached_tokens) for m in sent] == [(1, 12, 4), (2, 34, 0)] + # Admission also opens each request's prefill span, closed at its first sampled token. + assert sorted(scheduler._prefill_start) == [1, 2] def test_prepare_failure_emits_no_prompt_admission(): @@ -130,6 +133,20 @@ def fail_prepare(_batch): assert sent == [] +def test_freeing_a_request_that_never_sampled_closes_its_prefill_span(): + """The first sampled token is what normally pops the span. A request aborted mid-prefill + never reaches that point, so the free path has to pop it -- otherwise a long-lived server + leaks one entry per aborted request.""" + scheduler = Scheduler.__new__(Scheduler) + scheduler._prefill_start = {7: 100.0, 9: 200.0} + scheduler.cache_manager = SimpleNamespace(cache_req=lambda req, finished: None) + scheduler.table_manager = SimpleNamespace(free=lambda table_idx: None) + + Scheduler._free_req_resources(scheduler, SimpleNamespace(uid=7, table_idx=3)) + + assert scheduler._prefill_start == {9: 200.0} # only the freed request's span is dropped + + def test_scheduler_rejection_emits_error_but_no_admission(): scheduler = Scheduler.__new__(Scheduler) scheduler.engine = SimpleNamespace(max_seq_len=4) diff --git a/tests/server/test_anthropic_api.py b/tests/server/test_anthropic_api.py index 327bfcd87..8f7ef1544 100644 --- a/tests/server/test_anthropic_api.py +++ b/tests/server/test_anthropic_api.py @@ -33,6 +33,7 @@ ContentDelta, GenDone, GenResult, + GenTimings, ReasoningDelta, ToolCallsDelta, ) @@ -43,12 +44,14 @@ async def _aiter(items): yield it -def _collect_events(events, model="claude-x", uid=1, cache_report=False): +def _collect_events(events, model="claude-x", uid=1, cache_report=False, metrics=False): """Run the Anthropic event stream over neutral GenEvents; return [(type, data), ...].""" async def run(): out = [] - async for frame in A.anthropic_event_stream(_aiter(events), model, uid, cache_report=cache_report): + async for frame in A.anthropic_event_stream( + _aiter(events), model, uid, cache_report=cache_report, metrics=metrics + ): etype = None data = None for line in frame.split("\n"): @@ -976,3 +979,52 @@ def test_image_only_tool_result_keeps_an_empty_tool_message(): "role": "user", "content": [{"type": "image", "freetoken_ref": {"kind": "b64", "data": "aGk="}}], } + + +# --------------------------------------------------------------------------- # +# Per-request metrics (--enable-metrics-report) +# --------------------------------------------------------------------------- # +def _timed_result() -> GenResult: + return GenResult( + reasoning="", content="hi", tool_calls=[], finish_reason="stop", + prompt_tokens=11, completion_tokens=5, cached_tokens=8, + timings=GenTimings(ttft_ms=30.0, prefill_ms=12.0, decode_ms=40.0, total_ms=70.0), + ) + + +def test_full_response_metrics_absent_unless_asked_for(): + """Anthropic's wire has no `metrics` field, so /v1/messages must stay byte-identical for + every client that did not opt in -- absent, not null.""" + resp = A.anthropic_full_response(_timed_result(), "claude-x", uid=9) + assert "metrics" not in resp.model_dump(exclude_none=True) + + +def test_full_response_metrics_report_the_untouched_prompt_split(): + """cache_report rewrites usage.input_tokens to exclude the cached prefix; metrics must + still describe the real prefill, which is prompt_tokens - cached_tokens.""" + resp = A.anthropic_full_response( + _timed_result(), "claude-x", uid=9, cache_report=True, metrics=True + ) + body = resp.model_dump(exclude_none=True) + assert body["usage"]["input_tokens"] == 3 + assert body["metrics"]["prefill_tokens"] == 3 + assert body["metrics"]["cached_prompt_tokens"] == 8 + assert body["metrics"]["prefill_time_ms"] == 12.0 + assert body["metrics"]["ttft_ms"] == 30.0 + + +def test_stream_message_delta_carries_metrics(): + events = [ + ContentDelta("hi"), + GenDone("stop", 4, 2, cached_tokens=1, + timings=GenTimings(ttft_ms=5.0, prefill_ms=2.0, decode_ms=9.0, total_ms=14.0)), + ] + collected = _collect_events(events, metrics=True) + md = next(e[1] for e in collected if e[0] == "message_delta") + assert md["metrics"]["prefill_time_ms"] == 2.0 + assert md["metrics"]["prefill_tokens"] == 3 + # Only the terminal event carries them. + assert sum(1 for _, data in collected if isinstance(data, dict) and "metrics" in data) == 1 + + off = _collect_events(events) + assert all("metrics" not in data for _, data in off if isinstance(data, dict)) diff --git a/tests/server/test_generation_accounting.py b/tests/server/test_generation_accounting.py index 8276163b0..5615460ec 100644 --- a/tests/server/test_generation_accounting.py +++ b/tests/server/test_generation_accounting.py @@ -22,6 +22,8 @@ from freetoken.server.generation import ( # noqa: E402 GenDone, GenSpec, + GenTimings, + build_metrics, generate_events, generate_full, ) @@ -63,13 +65,22 @@ async def wait_for_ack(self, uid: int): yield reply -def _ack(prompt: int = 0, completion: int = 0, out: str = "", finished: bool = False) -> UserReply: +def _ack( + prompt: int = 0, + completion: int = 0, + out: str = "", + finished: bool = False, + cached: int = 0, + prefill_ms: float = 0.0, +) -> UserReply: return UserReply( uid=42, incremental_output=out, finished=finished, prompt_tokens_delta=prompt, completion_tokens_delta=completion, + cached_tokens=cached, + prefill_ms=prefill_ms, finish_reason="stop" if finished else None, ) @@ -193,3 +204,103 @@ 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 + + +# ------------------------------------------------------- per-request metrics +# The timings ride the same ack stream the token totals do, so they are covered here rather +# than in each adapter's file; the adapter tests only check that the object reaches the wire. +def test_prefill_ms_from_the_engine_reaches_the_result_unchanged(): + """The scheduler measures the prefill span; the generation layer must pass it through + rather than re-deriving it from ack arrival times (which batch together).""" + st = FakeState([ + _ack(prompt=100, cached=40), + _ack(completion=1, out="a", prefill_ms=12.5), + _ack(completion=1, out="b", finished=True), + ]) + result = asyncio.run(generate_full(42, _spec(), st)) + assert result.timings.prefill_ms == 12.5 + assert result.cached_tokens == 40 + + +# A real gap between acks, so a TTFT taken at the wrong ack is distinguishable from one taken +# at the right one; FakeState yields its whole list within a single event-loop tick. +_GAP_S = 0.02 +_GAP_MS = _GAP_S * 1000 + + +class PacedState(FakeState): + async def wait_for_ack(self, uid: int): + assert uid == 42 + for index, reply in enumerate(self._replies): + if index: + await asyncio.sleep(_GAP_S) + yield reply + + +def test_ttft_is_measured_at_the_first_generated_token_not_the_admission_ack(): + """The admission ack carries prompt_tokens and no token; timing TTFT from it would report + roughly the queue time for every request.""" + st = PacedState([ + _ack(prompt=5), + _ack(completion=1, out="hi"), + _ack(completion=1, out="!", finished=True), + ]) + result = asyncio.run(generate_full(42, _spec(), st)) + # One gap to the first token (not zero, as an admission-timed TTFT would give), two to the end. + assert _GAP_MS * 0.5 < result.timings.ttft_ms < _GAP_MS * 1.9 + assert result.timings.total_ms > result.timings.ttft_ms + _GAP_MS * 0.5 + assert result.timings.decode_ms > _GAP_MS * 0.5 + + +def test_ttft_counts_a_token_the_detokenizer_held_back(): + """A token whose text is withheld (a trailing partial-stop prefix) is still a generated + token: TTFT keys off completion_tokens_delta, not off non-empty output -- otherwise it + would skip to the ack after it.""" + st = PacedState([ + _ack(prompt=5), + _ack(completion=1, out=""), + _ack(completion=1, out="done", finished=True), + ]) + result = asyncio.run(generate_full(42, _spec(), st)) + assert _GAP_MS * 0.5 < result.timings.ttft_ms < _GAP_MS * 1.9 + + +def test_stream_gendone_carries_the_same_timings(): + st = FakeState([ + _ack(prompt=8, cached=2), + _ack(completion=1, out="x", prefill_ms=7.0), + _ack(completion=1, out="y", finished=True), + ]) + + async def drain(): + return [ev async for ev in generate_events(42, _spec(), st) if isinstance(ev, GenDone)] + + (done,) = asyncio.run(drain()) + assert done.timings.prefill_ms == 7.0 + assert done.timings.ttft_ms > 0.0 + + +def test_build_metrics_splits_prompt_tokens_into_cached_and_prefilled(): + metrics = build_metrics( + prompt_tokens=8192, completion_tokens=256, cached_tokens=6144, + timings=GenTimings(ttft_ms=320.5, prefill_ms=410.0, decode_ms=5120.0, total_ms=5480.2), + ) + # The identity the issue asks for: prompt_tokens == cached_prompt_tokens + prefill_tokens. + assert metrics["cached_prompt_tokens"] + metrics["prefill_tokens"] == 8192 + assert metrics["prefill_tokens"] == 2048 + assert metrics["prefill_tokens_per_second"] == round(2048 / 0.410, 2) + # Decode divides by completion_tokens - 1: the first token came out of prefill. + assert metrics["decode_tokens_per_second"] == round(255 / 5.120, 2) + assert metrics["decode_tokens"] == 256 + assert metrics["ttft_ms"] == 320.5 + assert metrics["total_time_ms"] == 5480.2 + + +def test_build_metrics_rates_are_zero_rather_than_dividing_by_zero(): + metrics = build_metrics( + prompt_tokens=10, completion_tokens=1, cached_tokens=0, timings=GenTimings() + ) + assert metrics["prefill_tokens_per_second"] == 0.0 + # One generated token spans no decode interval, so there is no rate to report. + assert metrics["decode_tokens_per_second"] == 0.0 + assert metrics["prefill_tokens"] == 10 diff --git a/tests/server/test_openai_api.py b/tests/server/test_openai_api.py index e33018e19..ce675a250 100644 --- a/tests/server/test_openai_api.py +++ b/tests/server/test_openai_api.py @@ -687,3 +687,65 @@ def test_minimax_http_non_stream_forces_implicit_reasoning_without_request_knob( message = response["choices"][0]["message"] assert message["reasoning_content"] == "private thought" assert message["content"] == "visible answer" + + +# ----------------------------------------------------- per-request metrics +def _metrics_replies() -> list[UserReply]: + return [ + UserReply(uid=42, incremental_output="", finished=False, prompt_tokens_delta=10, cached_tokens=4), + UserReply(uid=42, incremental_output="hi", finished=True, completion_tokens_delta=2, prefill_ms=8.0), + ] + + +def test_non_stream_chat_metrics_only_with_flag(): + state = FakeState(_metrics_replies()) + state.config.enable_metrics_report = True + response = run(handle_chat_completion(chat_request(tools=None), request=None, state=state, model_sampling={})) + metrics = response["metrics"] + assert metrics["prefill_time_ms"] == 8.0 + assert metrics["prefill_tokens"] == 6 and metrics["cached_prompt_tokens"] == 4 + assert metrics["decode_tokens"] == 2 + assert metrics["total_time_ms"] > 0 + + off = run(handle_chat_completion(chat_request(tools=None), request=None, state=FakeState(_metrics_replies()), model_sampling={})) + assert "metrics" not in off + + +def test_non_stream_chat_metrics_report_the_cache_hit_without_enable_cache_report(): + """--enable-cache-report governs the billing fields in `usage`. Gating the hit inside + `metrics` on it too would leave prefill_tokens_per_second computed over tokens that were + never forwarded.""" + state = FakeState(_metrics_replies()) + state.config.enable_metrics_report = True + response = run(handle_chat_completion(chat_request(tools=None), request=None, state=state, model_sampling={})) + assert "prompt_tokens_details" not in response["usage"] + assert response["metrics"]["cached_prompt_tokens"] == 4 + + +def test_stream_chat_metrics_ride_the_usage_chunk(): + state = FakeState(_metrics_replies()) + state.config.enable_metrics_report = True + req = chat_request(tools=None, stream_options={"include_usage": True}) + + async def collect(): + return [chunk async for chunk in stream_chat_completion_chunks(42, req, state)] + + events = parse_sse(run(collect())) + final = next(e for e in reversed(events) if isinstance(e, dict) and e.get("usage")) + assert final["metrics"]["prefill_time_ms"] == 8.0 + assert final["metrics"]["decode_tokens"] == 2 + # No metrics on the content chunks -- exactly one carries them. + assert sum(1 for e in events if isinstance(e, dict) and "metrics" in e) == 1 + + +def test_stream_chat_without_include_usage_has_no_metrics_chunk(): + """Metrics ride the usage chunk, so a client that opted out of usage keeps the plain + OpenAI stream rather than getting an extra trailing chunk it never asked for.""" + state = FakeState(_metrics_replies()) + state.config.enable_metrics_report = True + + async def collect(): + return [chunk async for chunk in stream_chat_completion_chunks(42, chat_request(tools=None), state)] + + events = parse_sse(run(collect())) + assert not any(isinstance(e, dict) and "metrics" in e for e in events) diff --git a/tests/server/test_responses_api.py b/tests/server/test_responses_api.py index 887e78e57..6e7721b89 100644 --- a/tests/server/test_responses_api.py +++ b/tests/server/test_responses_api.py @@ -338,10 +338,11 @@ def test_stream_tool_call_events(): # Route smoke tests # --------------------------------------------------------------------------- # class FakeState: - def __init__(self, outputs, finish_reason=None, cached_tokens=0): + def __init__(self, outputs, finish_reason=None, cached_tokens=0, prefill_ms=0.0): self._outputs = outputs self._finish_reason = finish_reason # stamped on the terminal ack self._cached_tokens = cached_tokens # stamped on the first ack (admission reply) + self._prefill_ms = prefill_ms # the scheduler stamps it on the first generated token self.maintenance_state = "serving" self.config = SimpleNamespace( mm=SimpleNamespace(text_model_only=False, disabled_encoders=frozenset()), @@ -364,7 +365,8 @@ async def wait_for_ack(self, uid): yield UserReply(uid=uid, incremental_output=text, finished=finished, finish_reason=self._finish_reason if finished else None, prompt_tokens_delta=pt, completion_tokens_delta=ct, - cached_tokens=self._cached_tokens if i == 0 else 0) + cached_tokens=self._cached_tokens if i == 0 else 0, + prefill_ms=self._prefill_ms if i == 0 else 0.0) async def stream_with_cancellation(self, gen, request, uid): async for chunk in gen: @@ -967,3 +969,38 @@ def test_convert_function_call_output_text_list_stays_a_plain_tool_message(): spec = RP.convert_responses_to_genspec(req, {}) assert [m["role"] for m in spec.messages] == ["user", "assistant", "tool"] assert spec.messages[2]["content"] == "ab" + + +# --------------------------------------------------------------------------- # +# Per-request metrics (--enable-metrics-report) +# --------------------------------------------------------------------------- # +def test_route_nonstream_metrics_only_with_flag(): + fake = FakeState([("Hello world", True, 5, 2)], cached_tokens=3, prefill_ms=6.0) + fake.config.enable_metrics_report = True + body = _client(fake).post("/v1/responses", json={"model": "gpt-x", "input": "hi"}).json() + assert body["metrics"]["prefill_time_ms"] == 6.0 + assert body["metrics"]["prefill_tokens"] == 2 and body["metrics"]["cached_prompt_tokens"] == 3 + assert body["metrics"]["total_time_ms"] > 0 + + off = _client(FakeState([("Hello world", True, 5, 2)])).post( + "/v1/responses", json={"model": "gpt-x", "input": "hi"} + ).json() + assert "metrics" not in off + + +def test_route_stream_completed_event_carries_metrics(): + fake = FakeState([("Hello world", True, 5, 2)], cached_tokens=3, prefill_ms=6.0) + fake.config.enable_metrics_report = True + r = _client(fake).post( + "/v1/responses", json={"model": "gpt-x", "input": "hi", "stream": True} + ) + completed = None + for block in r.text.split("\n\n"): + for line in block.split("\n"): + if line.startswith("data:"): + payload = json.loads(line[len("data:"):].strip()) + if payload.get("type") == "response.completed": + completed = payload + assert completed is not None + assert completed["response"]["metrics"]["prefill_time_ms"] == 6.0 + assert completed["response"]["metrics"]["cached_prompt_tokens"] == 3