Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions docs/ops/distributed-prefill-kv-network.md
Original file line number Diff line number Diff line change
Expand Up @@ -302,6 +302,24 @@ curl -fsS https://kakeya.ai/v1/network/benchmarks/<run-id>
The `Benchmarks` dashboard tab shows live progress, phase comparison, history,
and complete redacted stage details.

### Generator/Critic Agent GAN demo

Run two logical agents through real multi-round model inference:

```bash
bash scripts/run_agent_gan_demo.sh \
--rounds 2 \
--output-tokens 64 \
--report /tmp/kakeya-agent-gan-demo.json
```

For every Generator and Critic turn the task first asks allens to Prefill the
complete agent context, then performs the actual inference against the promoted
Primary hot snapshot. Terminal output contains the real proposal/critique;
persisted reports contain only output length/hash and metrics. The report
separates inference-only KV hit rate from whole-workload hit rate including
warmup, plus per-agent and aggregate token throughput/latency.

## Rollback

The cache is an optimization; inference correctness does not depend on it.
Expand Down
36 changes: 35 additions & 1 deletion inference_engine/bench/prefill_fleet_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,13 @@
import statistics
from typing import Any, Sequence

PHASES = ("remote_compute", "primary_hot_hit", "allens_cold_restore")
PHASES = (
"remote_compute",
"primary_hot_hit",
"allens_cold_restore",
"agent_generator",
"agent_critic",
)
HIT_SOURCES = ("remote_worker", "primary_hot", "allens_offload", "unknown")
_PRIVATE_KEYS = {
"prompt",
Expand Down Expand Up @@ -60,6 +66,20 @@ def summarize_stages(stages: Sequence[dict[str, Any]]) -> dict[str, Any]:
for stage in normalized:
sources[stage["hit_source"]] += 1
decode = [stage["decode_tok_s"] for stage in normalized]
prefix_tokens = sum(stage["prefix_tokens"] for stage in normalized)
hit_tokens = sum(
int(stage.get("delta", {}).get("tokens_reused", 0))
for stage in normalized
)
warmup_prefix_tokens = sum(
int(stage.get("warmup_prefix_tokens", 0)) for stage in normalized
)
warmup_hit_tokens = sum(
int(stage.get("warmup_tokens_reused", 0)) for stage in normalized
)
decode_tokens = sum(stage["output_tokens"] for stage in normalized)
decode_seconds = sum(stage["decode_s"] for stage in normalized)
e2e_seconds = sum(stage["e2e_s"] for stage in normalized)
return {
"stages_total": len(normalized),
"stages_failed": sum(not stage.get("ok", False) for stage in normalized),
Expand All @@ -72,6 +92,20 @@ def summarize_stages(stages: Sequence[dict[str, Any]]) -> dict[str, Any]:
"generation_latency_ms_p50": _median(
stage["generation_latency_ms_per_token"] for stage in normalized
),
"inference_kv_token_hit_rate": (
hit_tokens / prefix_tokens if prefix_tokens else 0.0
),
"workload_kv_token_hit_rate": (
(hit_tokens + warmup_hit_tokens)
/ (prefix_tokens + warmup_prefix_tokens)
if prefix_tokens + warmup_prefix_tokens else 0.0
),
"aggregate_decode_tok_s": (
decode_tokens / decode_seconds if decode_seconds else 0.0
),
"aggregate_e2e_tok_s": (
decode_tokens / e2e_seconds if e2e_seconds else 0.0
),
"bytes_received": sum(
int(stage.get("delta", {}).get("bytes_received", 0))
for stage in normalized
Expand Down
8 changes: 4 additions & 4 deletions inference_engine/network/dashboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ def dashboard_html() -> str:
</section>
<section id="nodes" class="tab hidden"><h2>Registered inference nodes</h2><div class="card"><table><thead><tr><th>Node</th><th>Role</th><th>Region</th><th>Model / Cache</th><th>Link</th><th>Status</th></tr></thead><tbody id="nodesBody"></tbody></table></div></section>
<section id="groups" class="tab hidden"><h2>Paired inference groups</h2><div id="groupCards" class="grid3"></div><div class="card register"><h3>Create group</h3><label>Name<input id="groupName" placeholder="Snow Fox Studio"></label><label>Node IDs (comma separated)<input id="groupNodes"></label><button class="primary" id="createGroup">Create group</button></div></section>
<section id="benchmarks" class="tab hidden"><h2>Live benchmark</h2><div class="card" id="benchmarkLive">No benchmark running.</div><h2>Phase comparison</h2><div class="grid3" id="benchmarkPhases"><div class="card muted">Select a benchmark run.</div></div><h2>Recent benchmark history</h2><div class="card"><table><thead><tr><th>Run</th><th>Status</th><th>Started</th><th>Hit sources</th><th>Prefill tok/s</th><th>Decode tok/s</th><th>E2E tok/s</th></tr></thead><tbody id="benchmarkHistory"></tbody></table></div><h2>Run detail</h2><div class="card" id="benchmarkDetail">Select a run from history.</div></section>
<section id="benchmarks" class="tab hidden"><h2>Live benchmark</h2><div class="card" id="benchmarkLive">No benchmark running.</div><h2>Phase comparison</h2><div class="grid3" id="benchmarkPhases"><div class="card muted">Select a benchmark run.</div></div><h2>Recent benchmark history</h2><div class="card"><table><thead><tr><th>Run</th><th>Status</th><th>Started</th><th>Hit sources</th><th>KV hit</th><th>Prefill tok/s</th><th>Decode tok/s</th><th>Latency/token</th><th>E2E tok/s</th></tr></thead><tbody id="benchmarkHistory"></tbody></table></div><h2>Run detail</h2><div class="card" id="benchmarkDetail">Select a run from history.</div></section>
<p class="muted">Exact IPs, raw prompt hashes and cache keys are administrator-only. Region is operator-selected and coarse.</p>
</main>
<script>
Expand All @@ -49,13 +49,13 @@ def dashboard_html() -> str:
$('createRegistration').onclick=async()=>{let r=await fetch('/v1/network/nodes/register',{method:'POST',headers:writeHeaders(),body:JSON.stringify({alias:$('alias').value,address:$('address').value,region:$('region').value,role:'hybrid'})});let j=await r.json();$('pairing').textContent=r.ok?`Pairing token: ${j.pairing_token}\nExpires: ${new Date(j.expires_at*1000).toLocaleTimeString()}`:`Error: ${j.detail||r.status}`;$('pairing').classList.remove('hidden');load()};
$('createGroup').onclick=async()=>{await fetch('/v1/network/groups',{method:'POST',headers:writeHeaders(),body:JSON.stringify({name:$('groupName').value,node_ids:$('groupNodes').value.split(',').map(x=>x.trim()).filter(Boolean)})});load()};
function nodePosition(i,total){let a=(i/Math.max(total,1))*Math.PI*2;return {x:50+38*Math.cos(a),y:53+35*Math.sin(a)}}
function phaseCards(stages){$('benchmarkPhases').innerHTML=stages.map(x=>`<div class="card"><b>${esc(x.name)}</b><p>${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}</p><div class="muted">TTFT ${num(x.ttft_s)}s<br>Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s<br>Decode ${num(x.decode_tok_s)} tok/s<br>Generation ${num(x.generation_latency_ms_per_token)} ms/token<br>E2E ${num(x.e2e_tok_s)} tok/s</div></div>`).join('')||'<div class="card muted">No stages yet.</div>'}
async function showBenchmark(id){let r=await fetch('/v1/network/benchmarks/'+encodeURIComponent(id)).then(x=>x.json());phaseCards(r.stages||[]);$('benchmarkDetail').innerHTML=`<b>${esc(r.id)} · ${esc(r.status)}</b><p class="muted">${esc(r.kind)} · ${new Date(r.started_at*1000).toLocaleString()}</p><table><thead><tr><th>Phase</th><th>Source</th><th>TTFT</th><th>Prefill/restore</th><th>Decode</th><th>Latency/token</th><th>E2E</th></tr></thead><tbody>${(r.stages||[]).map(x=>`<tr><td>${esc(x.name)}</td><td>${esc(x.hit_source)}</td><td>${num(x.ttft_s)}s</td><td>${num(x.prefill_or_restore_tok_s)}</td><td>${num(x.decode_tok_s)}</td><td>${num(x.generation_latency_ms_per_token)}ms</td><td>${num(x.e2e_tok_s)}</td></tr>`).join('')}</tbody></table>`}
function phaseCards(stages){$('benchmarkPhases').innerHTML=stages.map(x=>`<div class="card"><b>${esc(x.agent?`${x.agent} R${x.round}`:x.name)}</b><p>${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}</p><div class="muted">TTFT ${num(x.ttft_s)}s<br>Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s<br>Decode ${num(x.decode_tok_s)} tok/s<br>Generation ${num(x.generation_latency_ms_per_token)} ms/token<br>E2E ${num(x.e2e_tok_s)} tok/s</div></div>`).join('')||'<div class="card muted">No stages yet.</div>'}
async function showBenchmark(id){let r=await fetch('/v1/network/benchmarks/'+encodeURIComponent(id)).then(x=>x.json());phaseCards(r.stages||[]);$('benchmarkDetail').innerHTML=`<b>${esc(r.id)} · ${esc(r.status)}</b><p class="muted">${esc(r.kind)} · ${new Date(r.started_at*1000).toLocaleString()}</p><table><thead><tr><th>Phase</th><th>Source</th><th>TTFT</th><th>Prefill/restore</th><th>Decode</th><th>Latency/token</th><th>E2E</th></tr></thead><tbody>${(r.stages||[]).map(x=>`<tr><td>${esc(x.agent?`${x.agent} R${x.round}`:x.name)}</td><td>${esc(x.hit_source)}</td><td>${num(x.ttft_s)}s</td><td>${num(x.prefill_or_restore_tok_s)}</td><td>${num(x.decode_tok_s)}</td><td>${num(x.generation_latency_ms_per_token)}ms</td><td>${num(x.e2e_tok_s)}</td></tr>`).join('')}</tbody></table>`}
async function load(){let [s,n,g,live,runs]=await Promise.all([fetch('/v1/network/summary').then(r=>r.json()),fetch('/v1/network/nodes').then(r=>r.json()),fetch('/v1/network/groups').then(r=>r.json()),fetch('/v1/network/benchmarks/live').then(r=>r.json()),fetch('/v1/network/benchmarks?limit=20').then(r=>r.json())]);
$('online').textContent=s.online_nodes;$('groupCount').textContent=s.groups;$('tokens').textContent=fmt(s.completed_tokens);$('hitRate').textContent=(s.kv_hit_rate*100).toFixed(0)+'%';$('cache').textContent=gb(s.cache_bytes_used+s.cache_bytes_free)+' GB';let p=s.prefill||{};$('remoteJobs').textContent=fmt(p.remote_jobs);$('remoteHits').textContent=fmt(p.remote_hits);$('reusedTokens').textContent=fmt(p.tokens_reused);$('evictions').textContent=fmt(s.cache_evictions);$('publishFailures').textContent=fmt(p.publish_failures);
let total=s.cache_bytes_used+s.cache_bytes_free,pct=total?s.cache_bytes_used/total*100:0;$('capacityLabel').textContent=`${gb(s.cache_bytes_used)} / ${gb(total)} GB`;$('capacityBar').style.width=pct+'%';
$('map').innerHTML=n.map((x,i)=>{let p=nodePosition(i,n.length);return `<div class="dot ${x.role.includes('head')?'head':''}" style="left:${p.x}%;top:${p.y}%"><i></i><small>${x.region}<br>${x.alias}</small></div>`}).join('');
$('nodesBody').innerHTML=n.map(x=>`<tr><td>${x.alias}</td><td>${x.role}</td><td>${x.region}</td><td>${x.cache?(x.cache.model_id+' / '+x.cache.format):'—'}</td><td>${x.endpoint.network} ${x.endpoint.rtt_ms?x.endpoint.rtt_ms+'ms':''}</td><td class="${x.status}">${x.status}</td></tr>`).join('');
$('groupCards').innerHTML=g.map(x=>`<div class="card"><b>${x.name}</b><p class="muted">${x.online} / ${x.node_ids.length} online</p><div>${x.node_ids.join(' · ')}</div></div>`).join('')||'<div class="card muted">Create the first inference group.</div>';$('benchmarkLive').innerHTML=live?`<b>${esc(live.id)} · ${esc(live.status)}</b><p class="muted">${esc(live.kind)} · ${(live.stages||[]).length} stages completed</p>`:'No benchmark running.';$('benchmarkHistory').innerHTML=runs.map(r=>{let s=r.summary||{},h=s.hit_source_counts||{};return `<tr data-run="${esc(r.id)}"><td><button onclick="showBenchmark('${esc(r.id)}')">${esc(r.id)}</button></td><td>${esc(r.status)}</td><td>${new Date(r.started_at*1000).toLocaleString()}</td><td>worker ${h.remote_worker||0} · hot ${h.primary_hot||0} · offload ${h.allens_offload||0}</td><td>${num(s.prefill_tok_s_p50)}</td><td>${num(s.decode_tok_s_p50)}</td><td>${num(s.e2e_tok_s_p50)}</td></tr>`}).join('')||'<tr><td colspan="7" class="muted">No benchmark history.</td></tr>';if(live)phaseCards(live.stages||[])}
$('groupCards').innerHTML=g.map(x=>`<div class="card"><b>${x.name}</b><p class="muted">${x.online} / ${x.node_ids.length} online</p><div>${x.node_ids.join(' · ')}</div></div>`).join('')||'<div class="card muted">Create the first inference group.</div>';let last=live&&(live.stages||[]).slice(-1)[0];$('benchmarkLive').innerHTML=live?`<b>${esc(live.id)} · ${esc(live.status)}</b><p class="muted">${esc(live.kind)} · ${(live.stages||[]).length} stages completed${last?`<br>${esc(last.name)} · ${esc(last.hit_source)} · TTFT ${num(last.ttft_s)}s · Decode ${num(last.decode_tok_s)} tok/s`:''}</p>`:'No benchmark running.';$('benchmarkHistory').innerHTML=runs.map(r=>{let s=r.summary||{},h=s.hit_source_counts||{};return `<tr data-run="${esc(r.id)}"><td><button onclick="showBenchmark('${esc(r.id)}')">${esc(r.id)}</button></td><td>${esc(r.status)}</td><td>${new Date(r.started_at*1000).toLocaleString()}</td><td>worker ${h.remote_worker||0} · hot ${h.primary_hot||0} · offload ${h.allens_offload||0}</td><td>${num((s.workload_kv_token_hit_rate||0)*100)}%</td><td>${num(s.prefill_tok_s_p50)}</td><td>${num(s.decode_tok_s_p50)}</td><td>${num(s.generation_latency_ms_p50)}ms</td><td>${num(s.e2e_tok_s_p50)}</td></tr>`}).join('')||'<tr><td colspan="9" class="muted">No benchmark history.</td></tr>';if(live)phaseCards(live.stages||[])}
load();setInterval(load,5000);
</script></body></html>"""
Loading
Loading