Coverage for orchestrant/benchmark/answers.py: 97%
161 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-18 00:48 +0000
1"""What a reply is: which part was thinking, and whether an answer arrived.
3The speed runner used to score a `<think>` block that never closed as 0 %
4thinking, and to publish the time to the token cap as the "time to a finished
5answer". Both were wrong in the same runs: on 2026-09-24, six of nine
6CPU-lane replies (a thinking Qwen3-4B, `max_tokens` 256) never left
7`<think>`, the printed "86 % thinking" averaged the three that did, and the
8"12.5 s to a finished answer" was 12.5 s to the cap. A reply that ran out of
9budget has no answer, and its time to one was not measured.
11Servers put thinking in three places, and all three are read here: inline
12`<think>` inside `content` (GenieX, llama.cpp without --reasoning-format),
13`reasoning` (Ollama) and `reasoning_content` (llama-server, vLLM). Reading
14only `content` made a reasoning_content server's whole thinking phase count
15as time to first token.
17A reply's decode rate is read here too (decode_fields), and only where the
18stream gave it a window long enough to time.
20A thinking share is not always there to read: a reply cut before any marker
21may be all thinking (thinking_share), and its share is recorded as unknown.
22"""
24from __future__ import annotations
26import json
27import time
30# The shortest decode window a row's rate is read from; decode_fields says why.
31MIN_DECODE_WINDOW_S = 0.05
33# Why a reply with text has no thinking share; thinking_share says when.
34UNKNOWN_SHARE_NOTE = (
35 "cut before any <think>, </think> or reasoning: a chat template that opens "
36 "<think> in the prompt leaves no marker, so this reply may be all thinking"
37)
40def delta_pieces(delta):
41 """(answer, thinking) text of one streamed delta or one whole message."""
42 answer = delta.get("content") or ""
43 thought = delta.get("reasoning_content") or delta.get("reasoning") or ""
44 return answer, thought
47def split_answer(content, reasoning=""):
48 """(thinking characters, answer text) of a whole reply.
50 Inline `<think>` that never closed is ALL thinking: the answer never began.
51 """
52 if "</think>" in content:
53 answer = content.rsplit("</think>", 1)[1]
54 elif "<think>" in content:
55 answer = ""
56 else:
57 answer = content
58 return len(reasoning) + len(content) - len(answer), answer
61def thinking_share(content, reasoning="", cut=False):
62 """(share, note): the thinking share of a whole reply, or None and why.
64 Qwen3 and the Qwen3.8 distills open `<think>` in the PROMPT, so a reply
65 carries only the closing tag, and one cut before it carries neither: the
66 9B distill's two CUT rows of cpu-9b-classic-r3.json read 0.0 beside
67 0.42-0.96 on the seven that finished. Such a reply is all thinking or all
68 answer and cannot say which. A FINISHED reply with no marker keeps 0.0:
69 a template that opened `<think>` closes it before the answer.
70 """
71 total = len(content) + len(reasoning)
72 if not total:
73 return None, None
74 thinking, _ = split_answer(content, reasoning)
75 if cut and not thinking:
76 return None, UNKNOWN_SHARE_NOTE
77 return round(thinking / total, 3), None
80class Reply:
81 """One reply, streamed or not, with the moments that matter."""
83 def __init__(self):
84 self._answer, self._thought = [], []
85 self.usage = None
86 self.finish_reason = None
87 self.first_token_at = None # any token, thinking included: end of prefill
88 self.first_answer_at = None # first visible answer token
89 self.chunks = 0
91 @property
92 def content(self):
93 return "".join(self._answer)
95 @property
96 def reasoning(self):
97 return "".join(self._thought)
99 def add(self, answer, thought, now=None):
100 if not (answer or thought):
101 return
102 self.chunks += 1
103 self._answer.append(answer)
104 self._thought.append(thought)
105 if now is None:
106 return
107 if self.first_token_at is None:
108 self.first_token_at = now
109 if (
110 self.first_answer_at is None
111 and answer
112 and split_answer(self.content)[1].strip()
113 ):
114 self.first_answer_at = now
117def read_stream(lines, clock=time.monotonic):
118 """A Reply from SSE lines. The space after "data:" is optional (GenieX omits it)."""
119 reply = Reply()
120 for line in lines:
121 if not line.startswith("data:"):
122 continue
123 data = line[5:].strip()
124 if data == "[DONE]":
125 break
126 try:
127 chunk = json.loads(data)
128 except json.JSONDecodeError:
129 continue
130 if not isinstance(chunk, dict):
131 continue
132 if chunk.get("usage"): # the final chunk; its choices may be empty
133 reply.usage = chunk["usage"]
134 choices = chunk.get("choices") or []
135 if not choices:
136 continue
137 reply.finish_reason = choices[0].get("finish_reason") or reply.finish_reason
138 answer, thought = delta_pieces(choices[0].get("delta") or {})
139 reply.add(answer, thought, clock())
140 return reply
143def from_body(body):
144 """A Reply from a non-streamed response body."""
145 reply = Reply()
146 choice = (body.get("choices") or [{}])[0]
147 reply.add(*delta_pieces(choice.get("message") or {}))
148 reply.finish_reason = choice.get("finish_reason")
149 reply.usage = body.get("usage")
150 return reply
153def accounting(reply, completion_tokens=None, max_tokens=None):
154 """finish_reason, `answered` and the thinking share of one reply.
156 A reply is answered when visible answer text arrived and the server did not
157 stop it at the budget. Without a finish_reason, reaching max_tokens is read
158 as a cut: a server that does not say why it stopped cannot claim a finish.
159 """
160 content, reasoning = reply.content, reply.reasoning
161 _, answer = split_answer(content, reasoning)
162 cut = reply.finish_reason == "length" or (
163 reply.finish_reason is None
164 and bool(max_tokens)
165 and bool(completion_tokens)
166 and completion_tokens >= max_tokens
167 )
168 share, note = thinking_share(content, reasoning, cut)
169 return {
170 "finish_reason": reply.finish_reason,
171 "answered": bool(answer.strip()) and not cut,
172 "thinking_char_share": share,
173 "thinking_share_note": note,
174 }
177def decode_fields(elapsed, ttft, completion_tokens):
178 """A row's decode window and rate, or its window and why it has no rate.
180 The first token ends the prefill, so the window from it to the end of the
181 reply decoded `completion_tokens - 1`. Ollama sent three 8-12-token
182 replies of ollama-t8-4b-instruct-speed-answer.json in one burst, latency
183 == TTFT: windows of 0.36-0.72 ms read 9,733-26,712 tok/s. The shortest
184 real window in the 25 tracked speed reports is 174 ms (a 7-token reply at
185 34.5 tok/s, cpu-llama3b). MIN_DECODE_WINDOW_S lies between them, and above
186 three ticks of the coarsest clock this runner meets: time.monotonic on
187 Windows before Python 3.13, 15.6 ms. A one-token reply decoded nothing.
189 `decode_s` is kept either way, so speed_summary.decode_tok_s pools every
190 row it pooled when the burst rows still carried their rates.
191 """
192 if ttft is None: # not streamed: no first-token moment to start a window
193 return {"decode_s": None, "decode_tok_per_sec": None, "decode_rate_note": None}
194 window = elapsed - ttft
195 note = None
196 if completion_tokens < 2:
197 note = "under 2 tokens: none decoded after the first"
198 elif window < MIN_DECODE_WINDOW_S:
199 note = (
200 f"decode window {window * 1000:.1f} ms, "
201 f"under the {MIN_DECODE_WINDOW_S * 1000:.0f} ms floor"
202 )
203 return {
204 "decode_s": round(window, 6),
205 "decode_tok_per_sec": (
206 None if note else round((completion_tokens - 1) / window, 2)
207 ),
208 "decode_rate_note": note,
209 }
212def row_decode_rate(row):
213 """A row's decode rate, withholding one an older report stored from a burst.
215 A report written before decode_fields kept such a rate: the tracked t8
216 run's rows 0-2 still read 9,733-26,712 tok/s, and paired against them a
217 rerun that lost 20 % on the other six prompts read "noise +/-40%" and
218 passed (compare_speed). Their window is read back as (completion_tokens -
219 1) / rate; a row decode_fields wrote, with its `decode_s`, already says.
220 """
221 rate, tokens = row.get("decode_tok_per_sec"), row.get("completion_tokens")
222 if not rate or "decode_s" in row or not isinstance(tokens, int):
223 return rate or None
224 return None if (tokens - 1) / rate < MIN_DECODE_WINDOW_S else rate
227def row_thinking_share(row):
228 """A row's thinking share, repairing the ones an older report got wrong.
230 Before `answered` existed, a `<think>` that never closed scored 0.0; such a
231 row's preview still opens with the tag, so it is read back as all thinking.
232 Before `thinking_share_note` existed, a cut reply with no marker scored 0.0
233 too; at three decimals 0.0 means no marker and no reasoning, so a CUT row
234 of such a report is read back as unknown (thinking_share).
235 """
236 share = row.get("thinking_char_share")
237 if share != 0.0 or "thinking_share_note" in row:
238 return share
239 if "answered" not in row and str(
240 row.get("content_preview") or ""
241 ).lstrip().startswith("<think>"):
242 return 1.0
243 return None if _row_cut(row) else share
246def _row_cut(row):
247 """Did a speed, chat or coding row stop at a budget or a deadline?"""
248 unanswered = row.get("answered") is False and row.get("finish_reason") != "stop"
249 return bool(row.get("truncated") or row.get("gave_up") or unanswered)
252def row_thinking_unknown(row):
253 """A row whose reply had text but whose thinking share cannot be read."""
254 return row_thinking_share(row) is None and bool(
255 row.get("thinking_share_note") or row.get("thinking_char_share") == 0.0
256 )
259def row_answer_s(row):
260 """Seconds to a finished answer, or None when the row never finished one."""
261 if "answered" in row:
262 return row.get("wall_s_to_answer") if row["answered"] else None
263 return row.get("wall_s_to_answer", row.get("latency_s"))
266def answered_count(rows):
267 """(answered, rows that say) -- (None, 0) for reports older than the field.
269 A mean time to an answer is taken over the rows that HAVE one, so it must
270 travel with this count: a run that cut 5 of 9 replies averages its 4
271 shortest and would otherwise rank as the fastest.
272 """
273 flagged = [r for r in rows if "answered" in r]
274 if not flagged:
275 return None, 0
276 return sum(1 for r in flagged if r["answered"]), len(flagged)
279def answer_cell(row):
280 """The table's answer column: seconds, or "cut" where none arrived."""
281 seconds = row_answer_s(row)
282 return "cut" if seconds is None else seconds
285def _mean(values):
286 return sum(values) / len(values)
289def summary_lines(results):
290 """The time-to-answer and thinking lines of the speed table's summary."""
291 rows = [r for r in results if "error" not in r]
292 lines = []
293 if rows and all("answered" in r for r in rows):
294 done = [r for r in rows if r["answered"]]
295 cut = len(rows) - len(done)
296 line = f" Answered: {len(done)}/{len(rows)} inside max_tokens"
297 if cut:
298 line += f" ({cut} cut: their time to an answer was NOT measured)"
299 lines.append(line)
300 if done:
301 line = (
302 f" Time to answer: {_mean([r['wall_s_to_answer'] for r in done]):.1f}s "
303 f"avg over the {len(done)} answered"
304 )
305 lines.append(
306 line + " <-- rank models by THIS, not tok/s"
307 if not cut
308 else line + " -- raise --max-tokens until all answer, then rank"
309 )
310 ttfa = [r["ttfa_s"] for r in done if r.get("ttfa_s") is not None]
311 if ttfa:
312 lines.append(
313 f" First answer: {_mean(ttfa):.2f}s avg (first token after any thinking)"
314 )
315 else: # a report written before `answered` existed
316 answers = [s for s in map(row_answer_s, rows) if s is not None]
317 if answers:
318 lines.append(
319 f" Time to answer: {_mean(answers):.1f}s avg (older report: "
320 "rows cut at max_tokens are included and not identified)"
321 )
322 thinking = _thinking_line(rows)
323 return [*lines, thinking] if thinking else lines
326def _thinking_line(rows):
327 """The mean thinking share, and how many cut replies it cannot count.
329 Averaging only the rows that show a share reads a run whose cut replies
330 were all thinking as the share of the ones that finished.
331 """
332 shares = [s for s in map(row_thinking_share, rows) if s is not None]
333 unknown = sum(map(row_thinking_unknown, rows))
334 cut = f"{unknown} cut before any <think> marker"
335 if not shares:
336 return f" Thinking share: unknown: {cut}" if unknown else None
337 if not (unknown or any(shares)):
338 return None
339 line = f" Thinking share: {100 * _mean(shares):.0f}% of output was thinking"
340 if not unknown:
341 return line + " (pure latency for an agent)"
342 them = "reply that shows" if len(shares) == 1 else "replies that show"
343 return f"{line} on the {len(shares)} {them} it; {cut}: unknown"