front/broker: publish tokens/s for brokered streams on the Tasks page

The broker streaming relay counted tokens into engine.active["step"] but never
set ["rate"], so _merge_engine_tasks (which overlays both onto the running task)
showed token progression with a frozen 0 speed. Compute tok/s from the first
streamed token (so the model-load/queue wait doesn't drag the average down) and
publish it as m["rate"], mirroring the engine-side text path.
Co-Authored-By: 's avatarClaude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RdMufYvtTbtGDWsiZVoXce
parent bd5868be
...@@ -521,6 +521,7 @@ class FrontProxy: ...@@ -521,6 +521,7 @@ class FrontProxy:
_meas = (_status == 200 and "text/event-stream" _meas = (_status == 200 and "text/event-stream"
in (rp_resp.headers.get("content-type") or "")) in (rp_resp.headers.get("content-type") or ""))
ntok = 0 ntok = 0
_gen_t0 = None # wall-clock of the first streamed token (for tok/s)
async for raw in rp_resp.aiter_raw(): async for raw in rp_resp.aiter_raw():
if not raw: if not raw:
continue continue
...@@ -529,6 +530,16 @@ class FrontProxy: ...@@ -529,6 +530,16 @@ class FrontProxy:
m = (engine.active or {}).get(_rid) m = (engine.active or {}).get(_rid)
if m is not None: if m is not None:
m["step"] = ntok m["step"] = ntok
# Publish tokens/s too, measured from the first token so the
# model-load/queue wait doesn't drag the average down. The
# Tasks page reads m["rate"] (see _merge_engine_tasks); without
# this it showed token progress but a frozen 0 speed.
if _gen_t0 is None:
_gen_t0 = _t.time()
else:
_el = _t.time() - _gen_t0
if _el > 0:
m["rate"] = round(ntok / _el, 1)
yield raw.decode("utf-8", "replace") yield raw.decode("utf-8", "replace")
await rp_resp.aclose() await rp_resp.aclose()
break break
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment