TensorRT-LLM을 사용한 트레이스 재생 기반 에이전틱 LLM 서빙 벤치마크
README 링크에서 발견 · 2 days ago
에이전트 워크로드에 맞춰 TensorRT-LLM 서빙을 튜닝하기 위해 에이전트 트레이스를 재생했습니다.
What they did
그들은 자신들의 에이전트 스택에서 추론 레이어가 실제로 병목인지 알고 싶었기 때문에, 단일 요청 지연 수치를 신뢰하는 대신 TensorRT-LLM을 중심으로 벤치마크 하네스를 구축했습니다. 먼저 실제 에이전트 트래픽을 캡처했습니다. 가변 프롬프트 길이를 가진 멀티턴 도구 호출 세션, 턴 간 KV 캐시 재사용, 버스트성 동시성을 포함한 트래픽을 재생 가능한 트레이스로 저장했습니다. 그다음 TensorRT-LLM 서버를 세우고 지원되는 모델을 로드한 뒤, 제어된 도착률로 트레이스를 공급하면서 요청별 타이밍을 기록하는 재생 드라이버를 작성했습니다. 토큰/초만 보고하는 대신 작업 수준 지표를 계산했습니다. 에이전트 작업당 종단 간 시간, 부하 상태에서의 첫 토큰까지 걸리는 시간, 큐잉 지연, 그리고 작업 지연이 예산을 초과한 빈도입니다. 배치 크기, CUDA 그래프 설정, KV 캐시 구성을 훑어보며 각 구성을 동일한 트레이스와 비교해 결과가 동등한 조건에서 비교되도록 했습니다. 마지막으로 작업 수준 관점을 사용해 서빙 구성을 선택하고, 에이전트 트래픽과 일반 채팅 트래픽에서 어떤 노브가 중요한지 문서화했습니다.
How it went
trace-replay 접근 방식은 합성 벤치마크보다 훨씬 명확한 그림을 제공했습니다. 어떤 agent job이 큐잉에 의해 지배되는지, 생성에 의해 지배되는지를 확인할 수 있었고, 실제 도착 패턴에 맞춰 튜닝된 서빙 구성을 선택했습니다. 이 harness는 모델이나 구성 변경 후 회귀 테스트에 재사용할 수 있습니다. 완성도는 중간 수준입니다. 그들의 워크로드에는 작동하지만 범용 벤치마크 제품은 아니며, 결과는 그들이 캡처한 trace에 한정됩니다.
Worth knowing
NVIDIA GPU와 TensorRT-LLM이 지원하는 모델이 필요합니다. 설정에는 버전에 민감할 수 있는 CUDA, PyTorch, MPI 의존성이 포함됩니다. Trace 캡처는 동시성과 프롬프트 구조를 충실히 보존해야 하며, 그렇지 않으면 replay가 오해를 불러일으킵니다. Job 수준 지표는 job 경계를 어떻게 정의하느냐에 따라 달라집니다. 공개 데이터셋이나 바로 사용할 수 있는 스크립트를 의미하는 것은 아니며, trace와 모델 가중치는 직접 제공해야 합니다.
Wiring
순서: TensorRT-LLM 릴리스에 맞는 CUDA toolkit과 PyTorch를 설치하고, TensorRT-LLM을 설치한 뒤(pip wheel 또는 소스 빌드), 지원되는 모델 체크포인트를 다운로드하고, OpenAI 호환 엔드포인트를 노출하는 서버(예: trtllm-serve)를 실행합니다. 리플레이 클라이언트는 별도의 Python 프로세스로, 트레이스 파일(도착 타임스탬프, prompt, max_tokens, job_id가 담긴 JSONL)을 읽고 asyncio와 HTTP 클라이언트를 사용해 예정된 오프셋에 요청을 발사하며, job_id를 키로 한 요청별 타이밍을 로깅합니다. 메트릭 집계기는 요청 로그를 job 단위 요약으로 조인하여 CSV/JSON으로 기록합니다. 스윕 러너는 서로 다른 설정 플래그(배치 크기, KV 캐시 비율, CUDA graph 배치 크기)로 서버를 재시작하도록 셸 아웃하고, 헬스 체크를 기다린 뒤 동일한 트레이스를 실행하고 결과를 하나의 비교 테이블로 수집합니다. 설정은 리플레이 클라이언트와 스윕 러너가 공유하는 YAML 파일에 존재하여 트레이스와 설정이 재현 가능하게 유지됩니다.
Try it yourself
# Agentic Serving Benchmark for TensorRT-LLM
Below is a complete implementation plan with code for each step. I'll keep dependencies minimal (stdlib + `aiohttp` + `numpy` for percentiles, or pure stdlib if you prefer).
---
## Repo layout
```
benchmarks/agentic/
├── README.md
├── trace_format.md
├── requirements.txt
├── trace.py # trace schema + reader/writer
├── gen_synthetic_trace.py # step 5
├── replay_client.py # step 2 + 3
├── aggregate.py # step 3 (job-level metrics)
├── sweep.py # step 4
└── traces/
└── synthetic_1k.jsonl
```
---
## Step 1 — Trace format
`benchmarks/agentic/trace_format.md`
```markdown
# Agentic Trace Format (JSONL)
One JSON object per line. Each line is one request.
| field | type | required | notes |
|----------------|-----------------|----------|-------|
| `request_id` | string | yes | unique per request |
| `job_id` | string | yes | groups multi-turn agent calls |
| `turn` | int | no | 0-based turn index within job |
| `arrival_ms` | float | yes | ms since trace start (t=0) |
| `prompt` | string | one of | raw text prompt |
| `prompt_ids` | list[int] | one of | token ids (preferred) |
| `max_tokens` | int | yes | generation cap |
| `temperature` | float | no | default 0.0 |
| `metadata` | object | no | free-form |
Exactly one of `prompt` / `prompt_ids` must be present.
`arrival_ms` must be non-decreasing across the file (sorted).
```
`benchmarks/agentic/trace.py`
```python
"""Trace schema, reader, and writer for the agentic serving benchmark."""
from __future__ import annotations
import json
from dataclasses import dataclass, field, asdict
from typing import Any, Iterable, Iterator, Optional
@dataclass
class TraceRecord:
request_id: str
job_id: str
arrival_ms: float
max_tokens: int
prompt: Optional[str] = None
prompt_ids: Optional[list[int]] = None
turn: int = 0
temperature: float = 0.0
metadata: dict[str, Any] = field(default_factory=dict)
def validate(self) -> None:
if (self.prompt is None) == (self.prompt_ids is None):
raise ValueError(
f"{self.request_id}: exactly one of prompt/prompt_ids required"
)
if self.max_tokens <= 0:
raise ValueError(f"{self.request_id}: max_tokens must be > 0")
if self.arrival_ms < 0:
raise ValueError(f"{self.request_id}: arrival_ms must be >= 0")
def to_json(self) -> str:
d = asdict(self)
# drop nulls for compactness
d = {k: v for k, v in d.items() if v is not None and v != {}}
return json.dumps(d, separators=(",", ":"))
def read_trace(path: str) -> list[TraceRecord]:
records: list[TraceRecord] = []
with open(path, "r", encoding="utf-8") as f:
for lineno, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError as e:
raise ValueError(f"{path}:{lineno}: bad JSON: {e}") from e
rec = TraceRecord(**obj)
rec.validate()
records.append(rec)
records.sort(key=lambda r: r.arrival_ms)
return records
def write_trace(path: str, records: Iterable[TraceRecord]) -> None:
with open(path, "w", encoding="utf-8") as f:
for rec in records:
rec.validate()
f.write(rec.to_json() + "\n")
```
---
## Step 2 — Replay client
`benchmarks/agentic/replay_client.py`
```python
"""Replay a trace against a TensorRT-LLM OpenAI-compatible server.
Usage:
python replay_client.py --trace traces/synthetic_1k.jsonl \
--base-url http://localhost:8000 --concurrency 64 \
--out results/run.jsonl
"""
from __future__ import annotations
import argparse
import asyncio
import json
import time
from dataclasses import dataclass, asdict
from typing import Optional
import aiohttp
from trace import TraceRecord, read_trace
@dataclass
class RequestResult:
request_id: str
job_id: str
turn: int
arrival_ms: float
start_ms: float # when client actually sent
first_token_ms: Optional[float]
end_ms: float
ttft_ms: Optional[float]
total_ms: float
output_tokens: int
prompt_tokens: int
status: str # "ok" | "error"
error: Optional[str] = None
async def _issue(
session: aiohttp.ClientSession,
rec: TraceRecord,
base_url: str,
t0: float,
timeout_s: float,
) -> RequestResult:
# Wait until scheduled arrival
target = t0 + rec.arrival_ms / 1000.0
now = time.monotonic()
if target > now:
await asyncio.sleep(target - now)
start = time.monotonic()
payload = {
"model": "tensorrt_llm",
"max_tokens": rec.max_tokens,
"temperature": rec.temperature,
"stream": True,
"stream_options": {"include_usage": True},
}
if rec.prompt_ids is not None:
payload["prompt"] = rec.prompt_ids # TRT-LLM accepts token-id arrays
else:
payload["prompt"] = rec.prompt
first_token_ms: Optional[float] = None
output_tokens = 0
prompt_tokens = 0
status = "ok"
err: Optional[str] = None
try:
async with session.post(
f"{base_url}/v1/completions",
json=payload,
timeout=aiohttp.ClientTimeout(total=timeout_s),
) as resp:
if resp.status != 200:
body = await resp.text()
raise RuntimeError(f"HTTP {resp.status}: {body[:200]}")
async for raw in resp.content:
line = raw.strip()
if not line or not line.startswith(b"data:"):
continue
data = line[len(b"data:"):].strip()
if data == b"[DONE]":
break
chunk = json.loads(data)
if chunk.get("usage"):
prompt_tokens = chunk["usage"].get("prompt_tokens", prompt_tokens)
output_tokens = chunk["usage"].get("completion_tokens", output_tokens)
for choice in chunk.get("choices", []):
text = choice.get("text") or ""
if text and first_token_ms is None:
first_token_ms = (time.monotonic() - start) * 1000.0
if text:
output_tokens += 1 # fallback if usage missing
except Exception as e: # noqa: BLE001
status = "error"
err = f"{type(e).__name__}: {e}"
end = time.monotonic()
total_ms = (end - start) * 1000.0
return RequestResult(
request_id=rec.request_id,
job_id=rec.job_id,
turn=rec.turn,
arrival_ms=rec.arrival_ms,
start_ms=(start - t0) * 1000.0,
first_token_ms=first_token_ms,
end_ms=(end - t0) * 1000.0,
ttft_ms=first_token_ms,
total_ms=total_ms,
output_tokens=output_tokens,
prompt_tokens=prompt_tokens,
status=status,
error=err,
)
async def replay(
records: list[TraceRecord],
base_url: str,
concurrency: int,
timeout_s: float,
out_path: str,
) -> None:
sem = asyncio.Semaphore(concurrency)
t0 = time.monotonic()
async def bounded(rec: TraceRecord, session: aiohttp.ClientSession):
async with sem:
return await _issue(session, rec, base_url, t0, timeout_s)
connector = aiohttp.TCPConnector(limit=concurrency * 2)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [asyncio.create_task(bounded(r, session)) for r in records]
with open(out_path, "w", encoding="utf-8") as f:
for coro in asyncio.as_completed(tasks):
res = await coro
f.write(json.dumps(asdict(res)) + "\n")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--trace", required=True)
ap.add_argument("--base-url", default="http://localhost:8000")
ap.add_argument("--concurrency", type=int, default=64)
ap.add_argument("--timeout-s", type=float, default=600.0)
ap.add_argument("--out", required=True)
args = ap.parse_args()
records = read_trace(args.trace)
print(f"Loaded {len(records)} requests, "
f"{len({r.job_id for r in records})} jobs")
asyncio.run(replay(records, args.base_url, args.concurrency,
args.timeout_s, args.out))
print(f"Wrote {args.out}")
if __name__ == "__main__":
main()
```
---
## Step 3 — Aggregation
`benchmarks/agentic/aggregate.py`
```python
"""Aggregate per-request results into job-level metrics.
Usage:
python aggregate.py --results results/run.jsonl --out results/summary.json
"""
from __future__ import annotations
import argparse
import json
import statistics
from collections import defaultdict
from typing import Any
def _pct(xs: list[float], p: float) -> float:
if not xs:
return float("nan")
xs = sorted(xs)
k = (len(xs) - 1) * p
lo, hi = int(k), min(int(k) + 1, len(xs) - 1)
return xs[lo] + (xs[hi] - xs[lo]) * (k - lo)
def aggregate(results_path: str) -> dict[str, Any]:
reqs: list[dict] = []
with open(results_path, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
reqs.append(json.loads(line))
ok = [r for r in reqs if r["status"] == "ok"]
err = [r for r in reqs if r["status"] != "ok"]
# Per-request metrics
ttfts = [r["ttft_ms"] for r in ok if r["ttft_ms"] is not None]
totals = [r["total_ms"] for r in ok]
out_toks = [r["output_tokens"] for r in ok]
# Job-level: group by job_id, ordered by turn
jobs: dict[str, list[dict]] = defaultdict(list)
for r in ok:
jobs[r["job_id"]].append(r)
job_e2e: list[float] = []
job_queue: list[float] = [] # sum of (start - arrival) across turns
for jid, turns in jobs.items():
turns.sort(key=lambda r: r["turn"])
first = turns[0]
last = turns[-1]
# end-to-end = last end - first arrival
job_e2e.append(last["end_ms"] - first["arrival_ms"])
for t in turns:
job_queue.append(t["start_ms"] - t["arrival_ms"])
summary = {
"num_requests": len(reqs),
"num_ok": len(ok),
"num_errors": len(err),
"num_jobs": len(jobs),
"request": {
"ttft_ms": {
"mean": statistics.fmean(ttfts) if ttfts else None,
"p50": _pct(ttfts, 0.50),
"p99": _pct(ttfts, 0.99),
},
"total_ms": {
"mean": statistics.fmean(totals) if totals else None,
"p50": _pct(totals, 0.50),
"p99": _pct(totals, 0.99),
},
"output_tokens": {
"mean": statistics.fmean(out_toks) if out_toks else None,
"sum": sum(out_toks),
},
},
"job": {
"e2e_ms": {
"mean": statistics.fmean(job_e2e) if job_e2e else None,
"p50": _pct(job_e2e, 0.50),
"p99": _pct(job_e2e, 0.99),
},
"queueing_ms": {
"mean": statistics.fmean(job_queue) if job_queue else None,
"p50": _pct(job_queue, 0.50),
"p99": _pct(job_queue, 0.99),
},
},
}
return summary
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--results", required=True)
ap.add_argument("--out", required=True)
args = ap.parse_args()
summary = aggregate(args.results)
with open(args.out, "w", encoding="utf-8") as f:
json.dump(summary, f, indent=2)
print(json.dumps(summary, indent=2))
if __name__ == "__main__":
main()
```
---
## Step 4 — Config sweep
`benchmarks/agentic/sweep.py`
```python
"""Sweep TensorRT-LLM server configs and compare on the same trace.
Assumes you launch the server yourself (or via --launch-cmd) for each config.
The script:
1. iterates over a grid of (max_batch_size, cuda_graph, kv_cache_fraction)
2. launches the server (or expects it already running)
3. runs replay_client.py + aggregate.py
4. writes a comparison table (CSV + Markdown)
Usage:
python sweep.py --trace traces/synthetic_1k.jsonl \
--launch-cmd "trtllm-serve /models/llama3-8b --port 8000 \
--max_batch_size {max_batch_size} \
--kv_cache_free_gpu_memory_fraction {kv_cache_fraction} \
{cuda_graph_flag}" \
--out-dir results/sweep
"""
from __future__ import annotations
import argparse
import itertools
import json
import os
import shlex
import subprocess
import sys
import time
from pathlib import Path
import aiohttp # only for readiness probe
import asyncio
def wait_ready(base_url: str, timeout_s: float = 600.0) -> None:
async def _probe():
async with aiohttp.ClientSession() as s:
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
try:
async with s.get(f"{base_url}/health") as r:
if r.status == 200:
return
except Exception:
pass
await asyncio.sleep(2.0)
raise TimeoutError(f"server not ready at {base_url}")
asyncio.run(_probe())
def run_one(
cfg: dict,
trace: str,
base_url: str,
concurrency: int,
out_dir: Path,
launch_cmd: str | None,
) -> dict:
tag = f"mbs{cfg['max_batch_size']}_cg{int(cfg['cuda_graph'])}_kv{cfg['kv_cache_fraction']}"
run_dir = out_dir / tag
run_dir.mkdir(parents=True, exist_ok=True)
proc = None
if launch_cmd:
cmd = launch_cmd.format(
max_batch_size=cfg["max_batch_size"],
kv_cache_fraction=cfg["kv_cache_fraction"],
cuda_graph_flag="--enable_cuda_graph" if cfg["cuda_graph"] else "",
)
print(f"[sweep] launching: {cmd}")
proc = subprocess.Popen(shlex.split(cmd))
try:
wait_ready(base_url)
except Exception:
proc.terminate()
raise
try:
results = run_dir / "results.jsonl"
subprocess.run(
[sys.executable, "replay_client.py",
"--trace", trace,
"--base-url", base_url,
"--concurrency", str(concurrency),
"--out", str(results)],
check=True,
)
summary_path = run_dir / "summary.json"
subprocess.run(
[sys.executable, "aggregate.py",
"--results", str(results),
"--out", str(summary_path)],
check=True,
)
with open(summary_path) as f:
summary = json.load(f)
finally:
if proc is not None:
proc.terminate()
try:
proc.wait(timeout=30)
except subprocess.TimeoutExpired:
proc.kill()
summary["config"] = cfg
return summary
def write_table(rows: list[dict], out_dir: Path) -> None:
header = [
"max_batch_size", "cuda_graph", "kv_cache_fraction",
"req_ttft_p50", "req_ttft_p99",
"job_e2e_p50", "job_e2e_p99",
"job_queue_p50", "job_queue_p99",
"errors",
]
lines = [",".join(header)]
for r in rows:
c = r["config"]
lines.append(",".join(str(x) for x in [
c["max_batch_size"], int(c["cuda_graph"]), c["kv_cache_fraction"],
f"{r['request']['ttft_ms']['p50']:.1f}",
f"{r['request']['ttft_ms']['p99']:.1f}",
f"{r['job']['e2e_ms']['p50']:.1f}",
f"{r['job']['e2e_ms']['p99']:.1f}",
f"{r['job']['queueing_ms']['p50']:.1f}",
f"{r['job']['queueing_ms']['p99']:.1f}",
r["num_errors"],
]))
(out_dir / "comparison.csv").write_text("\n".join(lines) + "\n")
md = ["| " + " | ".join(header) + " |",
"|" + "|".join(["---"] * len(header)) + "|"]
for line in lines[1:]:
md.append("| " + " | ".join(line.split(",")) + " |")
(out_dir / "comparison.md").write_text("\n".join(md) + "\n")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--trace", required=True)
ap.add_argument("--base-url", default="http://localhost:8000")
ap.add_argument("--concurrency", type=int, default=64)
ap.add_argument("--out-dir", required=True)
ap.add_argument("--launch-cmd", default=None,
help="Optional shell command template; {max_batch_size}, "
"{kv_cache_fraction}, {cuda_graph_flag} are substituted. "
"If omitted, assumes a server is already running.")
ap.add_argument("--max-batch-sizes", default="32,64,128")
ap.add_argument("--cuda-graphs", default="0,1")
ap.add_argument("--kv-cache-fractions", default="0.5,0.8,0.9")
args = ap.parse_args()
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
grid = list(itertools.product(
[int(x) for x in args.max_batch_sizes.split(",")],
[bool(int(x)) for x in args.cuda_graphs.split(",")],
[float(x) for x in args.kv_cache_fractions.split(",")],
))
print(f"[sweep] {len(grid)} configs")
rows = []
for mbs, cg, kv in grid:
cfg = {"max_batch_size": mbs, "cuda_graph": cg, "kv_cache_fraction": kv}
print(f"[sweep] config={cfg}")
try:
rows.append(run_one(cfg, args.trace, args.base_url,
args.concurrency, out_dir, args.launch_cmd))
except Exception as e: # noqa: BLE001
print(f"[sweep] FAILED {cfg}: {e}")
write_table(rows, out_dir)
print(f"[sweep] wrote {out_dir/'comparison.csv'} and comparison.md")
if __name__ == "__main__":
main()
```
---
## Step 5 — Synthetic trace generator
`benchmarks/agentic/gen_synthetic_trace.py`
```python
"""Generate a synthetic agentic trace.
Simulates N jobs, each with 1-5 turns. Each turn's arrival is offset from the
previous turn's *expected* completion (prompt_tokens + max_tokens at a target
tok/s), so the trace looks like a real agent loop.
Usage:
python gen_synthetic_trace.py --num-jobs 200 --out traces/synthetic_1k.jsonl
"""
from __future__ import annotations
import argparse
import random
from trace import TraceRecord, write_trace
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--num-jobs", type=int, default=200)
ap.add_argument("--min-turns", type=int, default=1)
ap.add_argument("--max-turns", type=int, default=5)
ap.add_argument("--prompt-tokens-min", type=int, default=200)
ap.add_argument("--prompt-tokens-max", type=int, default=2000)
ap.add_argument("--max-tokens-min", type=int, default=64)
ap.add_argument("--max-tokens-max", type=int, default=512)
ap.add_argument("--target-tok-per-s", type=float, default=50.0,
help="assumed decode speed used to space turns")
ap.add_argument("--think-ms-min", type=float, default=50.0,
help="simulated client-side think time between turns")
ap.add_argument("--think-ms-max", type=float, default=500.0)
ap.add_argument("--seed", type=int, default=0)
ap.add_argument("--out", required=True)
args = ap.parse_args()
rng = random.Random(args.seed)
records: list[TraceRecord] = []
for j in range(args.num_jobs):
job_id = f"job-{j:05d}"
n_turns = rng.randint(args.min_turns, args.max_turns)
# stagger job starts over a 60s window
t_ms = rng.uniform(0, 60_000)
for turn in range(n_turns):
ptoks = rng.randint(args.prompt_tokens_min, args.prompt_tokens_max)
mtoks = rng.randint(args.max_tokens_min, args.max_tokens_max)
records.append(TraceRecord(
request_id=f"{job_id}-t{turn}",
job_id=job_id,
turn=turn,
arrival_ms=t_ms,
prompt_ids=[rng.randint(0, 32000) for _ in range(ptoks)],
max_tokens=mtoks,
temperature=0.0,
))
# next turn arrives after expected completion + think time
expected_ms = (ptoks + mtoks) / args.target_tok_per_s * 1000.0
t_ms += expected_ms + rng.uniform(args.think_ms_min, args.think_ms_max)
records.sort(key=lambda r: r.arrival_ms)
write_trace(args.out, records)
print(f"Wrote {len(records)} requests across {args.num_jobs} jobs to {args.out}")
if __name__ == "__main__":
main()
```
---
## `requirements.txt`
```
aiohttp>=3.9
```
(`numpy` not required — percentiles are computed in pure Python.)
---
## `README.md`
```markdown
# Agentic Serving Benchmark for TensorRT-LLM
Replays a multi-turn agent trace against a TensorRT-LLM OpenAI-compatible
server and reports per-request and per-job latency metrics.
## Requirements
**Hardware**
- NVIDIA GPU with ≥ 24 GB VRAM for a 7–8B model at `max_batch_size=64`
(e.g. L40S, A100, H100). Larger models / batches need proportionally more.
- CUDA 12.x driver compatible with your TensorRT-LLM build.
**Software**
- TensorRT-LLM ≥ 0.14 with `trtllm-serve` (OpenAI-compatible endpoint).
- Python ≥ 3.10.
- `pip install -r requirements.txt` (only `aiohttp`).
**Model**
- Any model served by `trtllm-serve` (e.g. Llama-3-8B-Instruct).
- The trace uses token ids by default; if you use text prompts, ensure the
server's tokenizer matches the one used to build the trace.
## Quick start
```bash
# 1. Generate a synthetic trace
python gen_synthetic_trace.py --num-jobs 200 --out traces/synthetic_1k.jsonl
# 2. Start the server (example)
trtllm-serve /models/llama3-8b --port 8000 \
--max_batch_size 64 --kv_cache_free_gpu_memory_fraction 0.8
# 3. Replay
python replay_client.py --trace traces/synthetic_1k.jsonl \
--base-url http://localhost:8000 --concurrency 64 \
--out results/run.jsonl
# 4. Aggregate
python aggregate.py --results results/run.jsonl --out results/summary.json
```
## Config sweep
```bash
python sweep.py --trace traces/synthetic_1k.jsonl \
--launch-cmd "trtllm-serve /models/llama3-8b --port 8000 \
--max_batch_size {max_batch_size} \
--kv_cache_free_gpu_memory_fraction {kv_cache_fraction} \
{cuda_graph_flag}" \
--max-batch-sizes 32,64,128 \
--cuda-graphs 0,1 \
--kv-cache-fractions 0.5,0.8,0.9 \
--out-dir results/sweep
```
Outputs `results/sweep/comparison.csv` and `comparison.md` with per-config
TTFT p50/p99, job end-to-end p50/p99, and queueing delay p50/p99.
## Metrics
- **TTFT** — time from request send to first streamed token.
- **Total latency** — time from send to stream close.
- **Job end-to-end** — `last_turn.end_ms − first_turn.arrival_ms`.
- **Queueing delay** — `start_ms − arrival_ms` per request (client-side
scheduling + server admission).
## Trace format
See `trace_format.md`. One JSON object per line; `job_id` groups turns.
## Notes
- The replay client uses `stream=true` with `stream_options.include_usage`
so token counts come from the server, not client-side heuristics.
- `arrival_ms` is relative to trace start; the client sleeps until each
request's scheduled time, so concurrency is bounded by `--concurrency`
but arrival timing is preserved.
- For reproducible comparisons, keep the same trace, seed, and concurrency
across sweep configs.
```
---
## Notes on design choices
- **Token-id prompts** are preferred because they remove tokenizer variance between trace generation and serving; the client falls back to text if `prompt` is set.
- **Streaming is required** for TTFT — non-streaming would only give total latency.
- **`as_completed`** writes results as they finish, so a long sweep doesn't lose data if interrupted.
- **Sweep launches the server itself** via `--launch-cmd` so each config gets a clean process; omit it to sweep against an externally managed server (e.g. a Kubernetes deployment).
- **Percentiles** use linear interpolation, matching `numpy.percentile` defaults, without the numpy dependency.