diff --git a/.gitignore b/.gitignore index 9ef909acc2..9e4ebf9af8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ **/__pycache__/** **/.coverage experimental/multiturn/vllm_benchmark/results/ +.openloop/ diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index bd5d1a57eb..6969c82a85 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -3583,7 +3583,9 @@ build_replay_cmd_yaml() { REPLAY_CMD+=" --tokenizer $MODEL" if [ "$SERVING_SEARCH" != "1" ]; then check_env_vars CONC - REPLAY_CMD+=" --concurrency $CONC" + # open-loop scenarios: CONC is the target number of concurrent calls (it sets the arrival rate in + # openloop_trace); --concurrency only caps open sessions, so it is set well above it there + REPLAY_CMD+=" --concurrency ${IX_CLIENT_CONCURRENCY:-$CONC}" fi # else: adaptive_search sweeps the concurrency itself (scenario YAML's sweep: # block; SLOs come from sweep.sla_filters, optionally overridden above). @@ -3601,6 +3603,25 @@ build_replay_cmd_yaml() { REPLAY_CMD+=" --output-artifact-dir $result_dir/aiperf_artifacts" } +# Open-loop scenarios (scenarios/*-openloop.yaml): rewrite the fetched LiveKit trace so calls arrive by the clock +# at CONC / call duration per second (utils/serving/livekit_openloop.py), point IX_CUSTOM_DATASET_FILE at it and +# lift aiperf's session cap above CONC. Settings: IX_OPENLOOP_{CALL_DURATION_S,RAMP_S,TAIL_S,SEED} (serving-YAML +# env), defaults in the script. The trace goes to node-local disk; openloop.json (arrivals, scoring window) to +# the run dir for utils/serving/livekit_ttft.py. +openloop_trace() { + check_env_vars INFMAX_CONTAINER_WORKSPACE AIPERF_PYTHON IX_CUSTOM_DATASET_FILE CONC SERVING_SCENARIO + local result_dir="$1" dir="${IX_OPENLOOP_DIR:-$INFMAX_CONTAINER_WORKSPACE/.openloop}" derived + mkdir -p "$dir" + derived="$dir/${SERVING_SCENARIO}-c${CONC}.jsonl" + "$AIPERF_PYTHON" "$INFMAX_CONTAINER_WORKSPACE/utils/serving/livekit_openloop.py" \ + "$IX_CUSTOM_DATASET_FILE" "$derived" "$result_dir/openloop.json" --ccu "$CONC" \ + --scenario "$INFMAX_CONTAINER_WORKSPACE/scenarios/${SERVING_SCENARIO}.yaml" \ + ${IX_OPENLOOP_CALL_DURATION_S:+--call-duration-s "$IX_OPENLOOP_CALL_DURATION_S"} \ + ${IX_OPENLOOP_RAMP_S:+--ramp-s "$IX_OPENLOOP_RAMP_S"} ${IX_OPENLOOP_TAIL_S:+--tail-s "$IX_OPENLOOP_TAIL_S"} \ + ${IX_OPENLOOP_SEED:+--seed "$IX_OPENLOOP_SEED"} || return 1 + export IX_CUSTOM_DATASET_FILE="$derived" IX_CLIENT_CONCURRENCY=$(( CONC * 4 )) +} + # Exit 0 when is a lock-pinned file dataset (scenarios/datasets/lock.yaml), # 1 when it is not, 2 when the lock cannot be read. is_custom_dataset() { diff --git a/benchmarks/single_node/agentic/serve_generic.sh b/benchmarks/single_node/agentic/serve_generic.sh index 3264420f8a..32e414cd99 100755 --- a/benchmarks/single_node/agentic/serve_generic.sh +++ b/benchmarks/single_node/agentic/serve_generic.sh @@ -66,6 +66,10 @@ else *) echo "ERROR: cannot tell whether $SERVING_LOADER is a lock-pinned dataset (exit $custom_rc)" >&2; exit 1 ;; esac fi +# open-loop scenarios replay the fetched trace with calls arriving by the clock (benchmark_lib.sh openloop_trace) +if [[ "$SERVING_SCENARIO" == *-openloop ]]; then + openloop_trace "$RESULT_DIR" +fi # the user's env may set HF_HUB_OFFLINE=1 for the engine; the pre-download must reach the Hub HF_HUB_OFFLINE= TRANSFORMERS_OFFLINE= HF_DATASETS_OFFLINE= "$AIPERF_HF_CLI" download "$MODEL" diff --git a/scenarios/custom-agentic-livekit111868-v1.1-warm-openloop.yaml b/scenarios/custom-agentic-livekit111868-v1.1-warm-openloop.yaml new file mode 100644 index 0000000000..cf5a8885fd --- /dev/null +++ b/scenarios/custom-agentic-livekit111868-v1.1-warm-openloop.yaml @@ -0,0 +1,44 @@ +# custom-agentic family, open loop: the V1.1-warm LiveKit 111868 trace (V1 append-ideal restricted to the 671 rooms +# with >= 4 real requests, a dial warm-up row before each room's first turn; HF README) with calls arriving by the clock instead of when a lane frees up, so every server receives the same calls per hour +# at the same concurrency. conc = target mean concurrent calls. Before aiperf runs, serve_generic.sh rewrites the +# trace with utils/serving/livekit_openloop.py: calls open as a seeded Poisson process at conc / 127 s per second +# (127 s = the trace's mean call length at production reply speed), 120 s linear ramp, arrivals stop 60 s before +# the end so the last scored calls can finish. Inside a call each request keeps its logged gap after the previous +# reply ends. Override with serving-YAML env: IX_OPENLOOP_CALL_DURATION_S, IX_OPENLOOP_RAMP_S, IX_OPENLOOP_TAIL_S, +# IX_OPENLOOP_SEED. Score with utils/serving/livekit_ttft.py: it keeps the calls opened in [ramp, duration - tail) +# (openloop.json in the run dir) and drops the dial warm-ups. +# Client policy: a request with no answer after 10 s is abandoned (the agent's side-request timeout; agent turns +# time out at 5 s, counted by the scorer). +schemaVersion: "2.0" +benchmark: + unsafeOverride: true + endpoint: + type: chat + path: /v1/chat/completions + streaming: true + useServerTokenCount: true + timeout: 10.0 + datasets: + - name: livekit-111868-v1.1-warm + type: file + format: mooncake_trace + path: ${IX_CUSTOM_DATASET_FILE} + # fixed_schedule sends each call at its timestamp, so file order does not matter; calls are drawn by the + # rewrite's seed + sampling: sequential + phases: + - name: profiling + kind: profiling + type: fixed_schedule + autoOffset: true + # 120 s ramp + 1800 s scored + 60 s tail. Hard stop: requests of calls still running then are not sent. + duration: 1980.0 + gracePeriod: 30.0 + failedRequestThreshold: 0.1 + artifacts: + sliceDuration: 1.0 + tokenizer: + trustRemoteCode: true + runtime: + statsInterval: 30.0 +randomSeed: 42 diff --git a/scenarios/datasets/lock.yaml b/scenarios/datasets/lock.yaml index 9f5a63b1c0..e967b168fd 100644 --- a/scenarios/datasets/lock.yaml +++ b/scenarios/datasets/lock.yaml @@ -29,3 +29,15 @@ livekit-111868-v2: revision: 7e36287967360851a1b05ba05730ff1d256b60c7 file: trace.v2-observed.jsonl sha256: 190af677563ea782d060456bbc44f92e802e1d63e4322b147ef0290a49efd4f3 +# V1.1-warm and its per-request plan (from bench/gemma4-livekit-dyn-swa-arms); the plan is read by +# utils/serving/livekit_ttft.py as -plan. +livekit-111868-v1.1-warm: + repo: Green-Lab/2026-10-16-111868-agentic-traces + revision: 6e0934c957565a42d52cafac096ec758a4302355 + file: trace.v1.1-append-ideal-warm.jsonl + sha256: e26542392a202764849c05018f2f8ea3f9574277ffa3400615f0b4df23b0eb2f +livekit-111868-v1.1-warm-plan: + repo: Green-Lab/2026-10-16-111868-agentic-traces + revision: 6e0934c957565a42d52cafac096ec758a4302355 + file: plan.v1.1-append-ideal-warm.csv + sha256: fc5f69bc810a483905adc7d0bcd40258cf78bae869465ee9f19d3261217dc391 diff --git a/serving/gemma4-livekit-1gpu-v11-openloop.yaml b/serving/gemma4-livekit-1gpu-v11-openloop.yaml new file mode 100644 index 0000000000..6e0c6297ef --- /dev/null +++ b/serving/gemma4-livekit-1gpu-v11-openloop.yaml @@ -0,0 +1,30 @@ +# gemma4-livekit-1gpu-v1 (Gemma-4-31B FP8-block, prod vLLM flags minus MTP, 1 GPU) on the open-loop V1.1-warm +# scenario: calls arrive by the clock, so a faster server gets the same calls per hour instead of more (scenario +# header). Score each conc with: python utils/serving/livekit_ttft.py (window from openloop.json). +name: gemma4-livekit-1gpu-v11-openloop +runner: hardware-hcm-8x_02 +image: vllm/vllm-openai:v0.25.1 +model: RedHatAI/gemma-4-31B-it-FP8-block +gpus: 1 +precision: fp8 +env: + VLLM_DISABLE_COMPILE_CACHE: "1" + NCCL_P2P_LEVEL: NVL +serve: | + vllm serve RedHatAI/gemma-4-31B-it-FP8-block + --served-model-name RedHatAI/gemma-4-31B-it-FP8-block + --tensor-parallel-size 1 + --gpu-memory-utilization 0.92 + --max-model-len 262144 + --max-num-seqs 64 + --max-num-batched-tokens 16384 + --enable-chunked-prefill + --long-prefill-token-threshold 8192 + --kv-cache-metrics + --enable-prompt-tokens-details + --enable-auto-tool-choice + --tool-call-parser gemma4 + --reasoning-parser gemma4 +scenario: custom-agentic-livekit111868-v1.1-warm-openloop +# target mean concurrent calls (calls arrive by the clock at conc / 127 s per second) +conc: [8, 16, 32] diff --git a/utils/serving/livekit_openloop.py b/utils/serving/livekit_openloop.py new file mode 100644 index 0000000000..9a33640b12 --- /dev/null +++ b/utils/serving/livekit_openloop.py @@ -0,0 +1,150 @@ +"""Open-loop LiveKit trace: calls arrive by the clock, not when a lane frees up. + + python utils/serving/livekit_openloop.py --ccu N + [--call-duration-s 127] [--duration-s T | --scenario scenarios/.yaml] [--ramp-s 120] [--tail-s 60] + [--seed 42] + +The lane replay (custom-agentic-livekit111868-*) starts a lane's next call the moment its previous one ends, and +each turn the logged gap after the previous reply ends, so a faster server shortens every call and receives more +calls per hour at the same concurrency. Here the load is a parameter: calls open as a Poisson process at rate +CCU / call_duration_s per second (call_duration_s = the trace's mean call length at production reply speed), ramping +linearly from zero over the first ramp_s seconds, then steady until duration_s - tail_s. Every server sees the same +arrivals for the same seed. + +Each arrival takes the next call of a seeded shuffle of the trace. Its first request gets the arrival `timestamp` +(aiperf fixed_schedule); later requests keep their logged `delay` (sent that long after the previous reply ends). +When more arrivals than calls are needed the shuffle cycles: the k-th play of a call (k >= 2) gets session id +`~k` and a `play ` marker at the start of its first user message, so the call's own context is cold again +while the shared system prompt still hits, as for a new caller. livekit_ttft.py strips the suffix to find the plan. + +openloop.json records the arrivals and the scoring window: calls that open inside [ramp_s, duration_s - tail_s) +are scored; earlier ones are the ramp, later ones only keep the server loaded while the scored calls finish. +Standard library only (PyYAML for --scenario); two streaming passes over the trace (1.4 GB for V1.1-warm). +""" + +from __future__ import annotations + +import argparse +import json +import random +import re +import sys + +SID_RE = re.compile(r'"session_id":\s*"([^"]*)"') + + +def arrivals(rng: random.Random, rate: float, ramp_s: float, horizon_s: float) -> list[float]: + """Poisson arrival times (s) in [0, horizon_s) at rate * min(1, t / ramp_s), by thinning.""" + out: list[float] = [] + if rate <= 0: + return out + t = 0.0 + while True: + t += rng.expovariate(rate) + if t >= horizon_s: + return out + ramp = min(1.0, t / ramp_s) if ramp_s > 0 else 1.0 + if rng.random() < ramp: + out.append(t) + + +def scan_rooms(path: str) -> list[str]: + rooms: list[str] = [] + seen: set[str] = set() + with open(path) as f: + for line in f: + m = SID_RE.search(line) + if m and m.group(1) not in seen: + seen.add(m.group(1)) + rooms.append(m.group(1)) + return rooms + + +def mark(row: dict, play: int) -> dict: + """Play marker at the start of the first user message: the call's own text goes cold, the system prompt does not.""" + msgs = [dict(m) for m in row.get("messages") or []] + for m in msgs: + if m.get("role") == "user" and isinstance(m.get("content"), str): + m["content"] = f"play{play} " + m["content"] + break + return {**row, "messages": msgs} + + +def scenario_duration(path: str) -> float: + import yaml + + phases = yaml.safe_load(open(path))["benchmark"]["phases"] + return float(next(p for p in phases if p.get("kind", p.get("name")) == "profiling")["duration"]) + + +def main(argv: list[str]) -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("src") + ap.add_argument("dst") + ap.add_argument("sidecar") + ap.add_argument("--ccu", type=float, required=True, help="target mean concurrent calls") + ap.add_argument("--call-duration-s", type=float, default=127.0, help="mean call length the rate is set from") + group = ap.add_mutually_exclusive_group(required=True) + group.add_argument("--duration-s", type=float, help="whole profiling phase: ramp + scored + tail") + group.add_argument("--scenario", help="read the profiling phase duration from this scenario YAML") + ap.add_argument("--ramp-s", type=float, default=120.0) + ap.add_argument("--tail-s", type=float, default=60.0) + ap.add_argument("--seed", type=int, default=42) + a = ap.parse_args(argv) + + duration = a.duration_s if a.duration_s is not None else scenario_duration(a.scenario) + horizon = duration - a.tail_s + if horizon <= a.ramp_s: + sys.exit(f"duration {duration:g} s leaves no scored window after ramp {a.ramp_s:g} s and tail {a.tail_s:g} s") + rate = a.ccu / a.call_duration_s + rng = random.Random(a.seed) + rooms = scan_rooms(a.src) + if not rooms: + sys.exit("no rooms in trace") + order = rooms[:] + rng.shuffle(order) + times = arrivals(rng, rate, a.ramp_s, horizon) + # arrival i -> (room, play): the shuffle cycles once every room has been opened + plays: dict[str, list[tuple[int, int]]] = {} + for i, t in enumerate(times): + plays.setdefault(order[i % len(order)], []).append((i // len(order) + 1, int(round(t * 1000)))) + + first_seen: set[str] = set() + written = 0 + with open(a.src) as fin, open(a.dst, "w") as fout: + for line in fin: + m = SID_RE.search(line) if line.strip() else None + sid = m.group(1) if m else None + if sid is None or sid not in plays: + continue + row = json.loads(line) + opening = sid not in first_seen + first_seen.add(sid) + for play, t_ms in plays[sid]: + out = dict(row) if play == 1 else mark(row, play) + if play > 1: + out["session_id"] = f"{sid}~{play}" + if opening: + out.pop("delay", None) + out["timestamp"] = t_ms + fout.write(json.dumps(out) + "\n") + written += 1 + + t0 = times[0] if times else 0.0 + sidecar = { + "ccu": a.ccu, "call_duration_s": a.call_duration_s, "calls_per_s": rate, "ramp_s": a.ramp_s, + "duration_s": duration, "tail_s": a.tail_s, "seed": a.seed, "arrivals": len(times), + "rooms_in_trace": len(rooms), "max_play": max((p for v in plays.values() for p, _ in v), default=0), + "requests": written, "first_arrival_s": t0, + # aiperf's autoOffset and the scorer both take the first arrival as t = 0 + "score_window_s": [a.ramp_s - t0, horizon - t0], + } + with open(a.sidecar, "w") as f: + json.dump(sidecar, f, indent=2) + print(f"open loop: {len(times)} calls at {rate * 60:.1f}/min (ccu {a.ccu:g} / {a.call_duration_s:g} s), " + f"max play {sidecar['max_play']}, {written} requests -> {a.dst}") + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/utils/serving/livekit_ttft.py b/utils/serving/livekit_ttft.py new file mode 100644 index 0000000000..7f8edb3bf8 --- /dev/null +++ b/utils/serving/livekit_ttft.py @@ -0,0 +1,207 @@ +"""TTFT of a LiveKit (custom-agentic-livekit111868-*) run, split by a room's first turn vs later turns. + + python utils/serving/livekit_ttft.py ... [--plan ] + +A directory is searched for every aiperf_artifacts/profile_export.jsonl (one per conc); each is read +with the plan of the dataset its run used: dataset.txt names it (e.g. livekit-111868-v1), and the +lock entry -plan in scenarios/datasets/lock.yaml pins the plan CSV that ships with the trace +(fetched like the trace: GREENLAB_HF_TOKEN). --plan uses a local CSV instead. + +First turns (no parent in the plan) carry the room's whole system prompt uncached, so they set the +tail; "5s turns" are the requests whose logged client timeout is 5 s. Only the profiling phase counts. +Dial warm-up rows (warm=1 in the plan of a -warm dataset: the first prompt minus the user's utterance, +max_tokens 1, sent before the room's first turn) are left out of every view and get their own line: +they are not on the user's path, but must finish before the first turn arrives. + +Open-loop runs (custom-agentic-livekit111868-*-openloop): replayed calls are `~k` (the plan row is the +room's); only calls whose first request started inside the scoring window of the run's openloop.json are kept +(the ramp and the tail are left out); "5s turns" also get the time to the first sentence (TTFS: TTFT plus the +next TTFS_TOKENS - 1 chunks, about what the TTS needs to start speaking). +""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +import sys +from collections import defaultdict +from pathlib import Path + +UNCACHED_BUCKETS = ((0, 5000), (5000, 11000), (11000, 20000), (20000, None)) +TTFS_TOKENS = 12 +PLAY_RE = re.compile(r"~\d+$") # open-loop replays of a room: ~k + + +def quantile(values: list[float], p: float) -> float: + values = sorted(values) + return values[min(len(values) - 1, int(p * len(values)))] if values else float("nan") + + +def load_plan(path: str | Path) -> dict[str, list[dict]]: + plan: dict[str, list[dict]] = defaultdict(list) + with open(path, newline="") as f: + for row in csv.DictReader(f): + plan[row["session_id"]].append(row) + for turns in plan.values(): + turns.sort(key=lambda r: int(r["i"])) + return plan + + +def ttfs(metrics: dict, ttft_s: float) -> float | None: + """TTFT plus the first TTFS_TOKENS - 1 inter-chunk gaps (s); None for a reply shorter than that.""" + gaps = (metrics.get("inter_chunk_latency") or {}).get("value") or [] + if len(gaps) < TTFS_TOKENS - 1: + return None + return ttft_s + sum(gaps[: TTFS_TOKENS - 1]) / 1e3 + + +def load_requests(profile_export: Path, plan: dict[str, list[dict]]) -> list[dict]: + """One dict per profiling request with a TTFT: ttft (s), ttfs (s or None), first, timeout_5s, + uncached (tokens), warm, call (conversation id, ~k kept), start (ns or None).""" + rows = [] + with open(profile_export) as f: + for line in f: + record = json.loads(line) + meta, metrics = record["metadata"], record["metrics"] + if meta.get("benchmark_phase") != "profiling" or "time_to_first_token" not in metrics: + continue + turn = plan[PLAY_RE.sub("", meta["conversation_id"])][meta["turn_index"]] + ttft_s = metrics["time_to_first_token"]["value"] / 1e3 + rows.append({ + "ttft": ttft_s, + "ttfs": ttfs(metrics, ttft_s), + "call": meta["conversation_id"], + "start": meta.get("request_start_ns"), + "first": turn["parent"] == "", + "timeout_5s": float(turn["timeout"]) == 5.0, + "uncached": int(turn["isl"]) - int(turn["shared"]), + "warm": turn.get("warm") == "1", + }) + return rows + + +def describe(name: str, rows: list[dict]) -> list[str]: + ttft = [r["ttft"] for r in rows] + p95 = quantile(ttft, 0.95) + over = [r for r in rows if r["ttft"] > p95] + first = [r for r in rows if r["first"]] + lines = [(f"{name}: n={len(rows)} p95={p95:.2f}s first-turn share={len(first) / len(rows):.1%} " + f"first-turn share of >p95: {sum(r['first'] for r in over) / max(len(over), 1):.0%}")] + for label, sel in (("first", first), ("later", [r for r in rows if not r["first"]])): + x = [r["ttft"] for r in sel] + if not x: + continue + lines.append(f" {label}: n={len(x):5d} p50={quantile(x, .5):.2f} p90={quantile(x, .9):.2f} " + f"p95={quantile(x, .95):.2f} p99={quantile(x, .99):.2f} " + f">1s={sum(v > 1 for v in x) / len(x):.0%} " + f"median uncached={quantile([r['uncached'] for r in sel], .5)}") + return lines + + +def score_window(profile_export: Path) -> tuple[float, float] | None: + """[t0, t1) s after the first request, from the run's openloop.json (next to aiperf_artifacts/); None if none.""" + for parent in profile_export.parents: + sidecar = parent / "openloop.json" + if sidecar.is_file(): + t0, t1 = json.loads(sidecar.read_text())["score_window_s"] + return float(t0), float(t1) + return None + + +def in_window(rows: list[dict], window: tuple[float, float]) -> list[dict]: + """Rows of the calls whose first request started inside the window (all of a kept call's requests stay).""" + starts = [r["start"] for r in rows if r["start"] is not None] + if not starts: + raise SystemExit("ERROR open-loop run without request_start_ns in profile_export.jsonl") + zero = min(starts) + opened: dict[str, float] = {} + for r in rows: + if r["start"] is not None: + opened[r["call"]] = min(opened.get(r["call"], float("inf")), (r["start"] - zero) / 1e9) + return [r for r in rows if window[0] <= opened.get(r["call"], -1.0) < window[1]] + + +def describe_ttfs(rows: list[dict]) -> list[str]: + x = [r["ttfs"] for r in rows if r["ttfs"] is not None] + if not x: + return [] + return [f" TTFS ({TTFS_TOKENS} tokens): n={len(x)} p50={quantile(x, .5):.2f} p90={quantile(x, .9):.2f} " + f"p95={quantile(x, .95):.2f} p99={quantile(x, .99):.2f}"] + + +def describe_warm(rows: list[dict]) -> list[str]: + x = [r["ttft"] for r in rows] + return [(f"warm-ups (excluded above): n={len(x)} p50={quantile(x, .5):.2f} p90={quantile(x, .9):.2f} " + f"p95={quantile(x, .95):.2f} p99={quantile(x, .99):.2f} max={max(x):.2f} done " + + " ".join(f"<={t:g}s={sum(v <= t for v in x) / len(x):.1%}" for t in (0.5, 1, 1.5, 2)))] + + +def report(rows: list[dict]) -> str: + warm = [r for r in rows if r["warm"]] + rows = [r for r in rows if not r["warm"]] + if not rows: + return "no profiling requests with a TTFT" + lines = describe("all", rows) + turns = [r for r in rows if r["timeout_5s"]] + lines += describe("5s turns", turns) + describe_ttfs(turns) if turns else [] + first = [r for r in rows if r["first"]] + for lo, hi in UNCACHED_BUCKETS: + x = [r["ttft"] for r in first if r["uncached"] >= lo and (hi is None or r["uncached"] < hi)] + if x: + lines.append(f" first turns uncached {lo}-{hi or ''}: n={len(x)} p50={quantile(x, .5):.2f} " + f"p95={quantile(x, .95):.2f} >1s={sum(v > 1 for v in x) / len(x):.0%}") + lines += describe_warm(warm) if warm else [] + return "\n".join(lines) + + +def find_exports(path: Path) -> list[Path]: + if path.is_file(): + return [path] + return sorted(path.rglob("profile_export.jsonl")) + + +def dataset_name(profile_export: Path) -> str: + """The dataset name from the run's dataset.txt, next to aiperf_artifacts/.""" + for parent in profile_export.parents: + txt = parent / "dataset.txt" + if txt.is_file(): + return txt.read_text().split()[0] + raise SystemExit(f"ERROR {profile_export}: no dataset.txt above it; pass --plan") + + +def plan_path(name: str) -> str: + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from fetch_custom_dataset import fetch, load_lock + + if f"{name}-plan" not in load_lock(): + raise SystemExit(f"ERROR dataset {name}: no {name}-plan in scenarios/datasets/lock.yaml; pass --plan") + return fetch(f"{name}-plan") + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("paths", nargs="+", type=Path, help="artifact dirs or profile_export.jsonl files") + parser.add_argument("--plan", help="plan CSV to use instead of the locked one") + args = parser.parse_args(argv) + exports = [e for p in args.paths for e in find_exports(p)] + if not exports: + raise SystemExit("ERROR no profile_export.jsonl under " + " ".join(map(str, args.paths))) + plans: dict[str, dict] = {} + for export in exports: + key = args.plan or dataset_name(export) + if key not in plans: + plans[key] = load_plan(args.plan or plan_path(key)) + rows = load_requests(export, plans[key]) + window = score_window(export) + print(f"== {export}") + if window is not None: + rows = in_window(rows, window) + print(f"open loop: calls opened in [{window[0]:.0f}, {window[1]:.0f}) s") + print(report(rows)) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/utils/test_serving_livekit_openloop.py b/utils/test_serving_livekit_openloop.py new file mode 100644 index 0000000000..b3627ac5e2 --- /dev/null +++ b/utils/test_serving_livekit_openloop.py @@ -0,0 +1,124 @@ +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[1] +SCRIPT = REPO / "utils/serving/livekit_openloop.py" + + +def trace_rows(): + """Three rooms; a dial warm-up row first (no delay), then turns with logged delays.""" + rows = [] + for room in ("room-a", "room-b", "room-c"): + msgs = [{"role": "system", "content": "shared head"}, {"role": "user", "content": f"{room} hello"}] + rows.append({"session_id": room, "messages": msgs, "output_length": 1}) + rows.append({"session_id": room, "messages": msgs, "output_length": 20, "delay": 3000}) + rows.append({"session_id": room, "messages": msgs + [{"role": "assistant", "content": "hi"}, + {"role": "user", "content": "more"}], + "output_length": 30, "delay": 6500}) + return rows + + +def generate(tmp_path, *extra, name="out"): + src = tmp_path / "trace.jsonl" + if not src.exists(): + src.write_text("".join(json.dumps(r) + "\n" for r in trace_rows())) + dst, sidecar = tmp_path / f"{name}.jsonl", tmp_path / f"{name}.json" + args = ["--ccu", "8", "--call-duration-s", "40", "--duration-s", "600", "--ramp-s", "60", "--tail-s", "60", *extra] + out = subprocess.run([sys.executable, str(SCRIPT), str(src), str(dst), str(sidecar), *args], + capture_output=True, text=True) + assert out.returncode == 0, out.stderr + return [json.loads(line) for line in dst.read_text().splitlines()], json.loads(sidecar.read_text()) + + +def test_same_seed_same_trace_other_seed_differs(tmp_path): + a, _ = generate(tmp_path, name="a") + b, _ = generate(tmp_path, name="b") + c, _ = generate(tmp_path, "--seed", "7", name="c") + assert a == b + assert a != c + + +def test_openings_get_arrival_times_later_turns_keep_their_gap(tmp_path): + rows, side = generate(tmp_path) + by_call: dict[str, list[dict]] = {} + for r in rows: + by_call.setdefault(r["session_id"], []).append(r) + assert len(by_call) == side["arrivals"] + horizon_ms = (side["duration_s"] - side["tail_s"]) * 1000 + for call in by_call.values(): + assert len(call) == 3 + assert "delay" not in call[0] and 0 <= call[0]["timestamp"] < horizon_ms + assert [r["delay"] for r in call[1:]] == [3000, 6500] + assert all("timestamp" not in r for r in call[1:]) + + +def test_replayed_calls_are_renamed_and_their_own_context_goes_cold(tmp_path): + rows, side = generate(tmp_path) + assert side["max_play"] >= 2 # 8 calls / 40 s over 540 s needs ~100 arrivals from 3 rooms + replay = [r for r in rows if r["session_id"] == "room-a~2"] + assert len(replay) == 3 + for r in replay: + assert r["messages"][0] == {"role": "system", "content": "shared head"} + assert r["messages"][1]["content"] == "play2 room-a hello" + assert all(r["messages"][1]["content"] == "room-a hello" for r in rows if r["session_id"] == "room-a") + + +def test_rate_and_scoring_window(tmp_path): + rows, side = generate(tmp_path) + assert side["calls_per_s"] == pytest.approx(8 / 40) + # linear 60 s ramp then steady until 600 - 60: expected 0.2 * (30 + 480) = 102 arrivals + assert 70 <= side["arrivals"] <= 135 + t0 = side["first_arrival_s"] + assert min(r["timestamp"] for r in rows if "timestamp" in r) == round(t0 * 1000) + assert side["score_window_s"] == pytest.approx([60 - t0, 540 - t0]) + + +def test_duration_from_the_scenario(tmp_path): + src = tmp_path / "trace.jsonl" + src.write_text("".join(json.dumps(r) + "\n" for r in trace_rows())) + out = subprocess.run([sys.executable, str(SCRIPT), str(src), str(tmp_path / "o.jsonl"), str(tmp_path / "o.json"), + "--ccu", "4", "--scenario", + str(REPO / "scenarios/custom-agentic-livekit111868-v1.1-warm-openloop.yaml")], + capture_output=True, text=True) + assert out.returncode == 0, out.stderr + side = json.loads((tmp_path / "o.json").read_text()) + assert (side["duration_s"], side["ramp_s"], side["tail_s"], side["call_duration_s"]) == (1980, 120, 60, 127) + + +def test_no_scored_window_is_an_error(tmp_path): + src = tmp_path / "trace.jsonl" + src.write_text("".join(json.dumps(r) + "\n" for r in trace_rows())) + out = subprocess.run([sys.executable, str(SCRIPT), str(src), str(tmp_path / "o.jsonl"), str(tmp_path / "o.json"), + "--ccu", "4", "--duration-s", "300", "--ramp-s", "180", "--tail-s", "300"], + capture_output=True, text=True) + assert out.returncode != 0 + assert "no scored window" in out.stderr + + +def test_openloop_trace_rewrites_the_dataset_and_lifts_the_session_cap(tmp_path): + src = tmp_path / "trace.jsonl" + src.write_text("".join(json.dumps(r) + "\n" for r in trace_rows())) + script = f''' + set -e + export INFMAX_CONTAINER_WORKSPACE="$PWD" AIPERF_CLI=aiperf + export IX_CUSTOM_DATASET_FILE={src} CONC=8 PORT=8888 MODEL=m SERVING_SEARCH=0 + export SERVING_SCENARIO=custom-agentic-livekit111868-v1.1-warm-openloop IX_OPENLOOP_DIR={tmp_path}/ol + export IX_OPENLOOP_SEED=3 + source benchmarks/benchmark_lib.sh + export AIPERF_PYTHON={sys.executable} # benchmark_lib.sh points it at the container venv + openloop_trace {tmp_path} + build_replay_cmd_yaml {tmp_path} + printf '%s\\n%s\\n' "$IX_CUSTOM_DATASET_FILE" "$REPLAY_CMD" + ''' + out = subprocess.run(["bash", "-c", script], cwd=REPO, capture_output=True, text=True) + assert out.returncode == 0, out.stderr + dataset, cmd = out.stdout.splitlines()[-2:] + assert dataset == f"{tmp_path}/ol/custom-agentic-livekit111868-v1.1-warm-openloop-c8.jsonl" + assert Path(dataset).stat().st_size > 0 + assert " --concurrency 32 " in cmd + side = json.loads((tmp_path / "openloop.json").read_text()) + assert (side["ccu"], side["seed"]) == (8, 3) diff --git a/utils/test_serving_livekit_ttft.py b/utils/test_serving_livekit_ttft.py new file mode 100644 index 0000000000..bddd7d8ce0 --- /dev/null +++ b/utils/test_serving_livekit_ttft.py @@ -0,0 +1,149 @@ +import hashlib +import json +import os +import shutil +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[1] +PLAN = textwrap.dedent("""\ + session_id,i,parent,shared,isl,rendered_isl,lcp_parent,osl,delay_ms,timeout + room-1,0,,0,12000,12000,0,10,,5.0 + room-1,1,0,11990,12100,12100,11990,10,800,5.0 + room-1,2,1,12090,12150,12150,12090,10,800,10.0 + room-2,0,,0,3000,3000,0,10,,5.0 +""") +FAKE_HUB = textwrap.dedent('''\ + import os + from . import errors + def hf_hub_download(repo_id, filename, revision=None, repo_type=None, token=None): + with open(os.environ["FAKE_HF_LOG"], "a") as f: + f.write(filename + "\\n") + return os.environ["FAKE_HF_FILE"] +''') +FAKE_ERRORS = textwrap.dedent('''\ + class HfHubHTTPError(Exception): + response = None + class RepositoryNotFoundError(HfHubHTTPError): pass +''') + + +def record(room, turn, ttft_ms, phase="profiling"): + return {"metadata": {"conversation_id": room, "turn_index": turn, "benchmark_phase": phase}, + "metrics": {"time_to_first_token": {"value": ttft_ms, "unit": "ms"}}} + + +@pytest.fixture() +def world(tmp_path): + """A mini repo with the script and a lock, a fake huggingface_hub serving the plan, and one run artifact.""" + repo = tmp_path / "repo" + (repo / "utils/serving").mkdir(parents=True) + (repo / "scenarios/datasets").mkdir(parents=True) + for name in ("livekit_ttft.py", "fetch_custom_dataset.py"): + shutil.copy(REPO / "utils/serving" / name, repo / "utils/serving/") + plan = tmp_path / "plan.csv" + plan.write_text(PLAN) + sha = hashlib.sha256(PLAN.encode()).hexdigest() + (repo / "scenarios/datasets/lock.yaml").write_text( + f"lk-v1-plan: {{repo: Green-Lab/x, revision: {'a' * 40}, file: plan.v1.csv, sha256: {sha}}}\n") + fake = tmp_path / "fake/huggingface_hub" + fake.mkdir(parents=True) + (fake / "__init__.py").write_text(FAKE_HUB) + (fake / "errors.py").write_text(FAKE_ERRORS) + + run = tmp_path / "artifacts/agentic_x_c96/aiperf_artifacts" + run.mkdir(parents=True) + (run.parent / "dataset.txt").write_text("lk-v1 Green-Lab/x@aaa file=trace.v1.jsonl sha256=0 path=/x\n") + rows = [record("room-1", 0, 900), record("room-1", 1, 100), record("room-1", 2, 300), + record("room-2", 0, 1500), record("room-2", 0, 50, phase="warmup")] + (run / "profile_export.jsonl").write_text("".join(json.dumps(r) + "\n" for r in rows)) + env = {**os.environ, "PYTHONPATH": str(tmp_path / "fake"), "FAKE_HF_LOG": str(tmp_path / "calls"), + "FAKE_HF_FILE": str(plan)} + env.pop("GREENLAB_HF_TOKEN", None) + env.pop("RESULT_DIR", None) + return {"repo": repo, "env": env, "plan": plan, "artifacts": tmp_path / "artifacts", + "calls": tmp_path / "calls"} + + +def run(world, *args): + return subprocess.run([sys.executable, str(world["repo"] / "utils/serving/livekit_ttft.py"), *map(str, args)], + env=world["env"], capture_output=True, text=True, check=False) + + +def test_artifact_dir_uses_the_locked_plan_of_its_dataset(world): + out = run(world, world["artifacts"]) + assert out.returncode == 0, out.stderr + assert world["calls"].read_text() == "plan.v1.csv\n" + lines = out.stdout.splitlines() + # 4 profiling requests (the warmup one is dropped); first turns 0.9 s and 1.5 s, later 0.1 s and 0.3 s. + assert lines[1] == "all: n=4 p95=1.50s first-turn share=50.0% first-turn share of >p95: 0%" + assert lines[2].startswith(" first: n= 2 p50=1.50 ") and ">1s=50%" in lines[2] + assert "median uncached=12000" in lines[2] + assert lines[3].startswith(" later: n= 2 p50=0.30 ") and "median uncached=110" in lines[3] + # The 10 s-timeout request leaves the "5s turns" view. + assert lines[4].startswith("5s turns: n=3 ") + assert " first turns uncached 0-5000: n=1 p50=1.50" in out.stdout + assert " first turns uncached 11000-20000: n=1 p50=0.90" in out.stdout + + +def test_local_plan_skips_the_fetch(world): + export = world["artifacts"] / "agentic_x_c96/aiperf_artifacts/profile_export.jsonl" + out = run(world, export, "--plan", world["plan"]) + assert out.returncode == 0, out.stderr + assert not world["calls"].exists() + assert "all: n=4 p95=1.50s" in out.stdout + + +def test_dataset_without_a_plan_entry_fails_with_a_hint(world): + (world["artifacts"] / "agentic_x_c96/dataset.txt").write_text("tau2-cs Green-Lab/y@b file=t sha256=0 path=/y\n") + out = run(world, world["artifacts"]) + assert out.returncode == 1 + assert "no tau2-cs-plan in scenarios/datasets/lock.yaml; pass --plan" in out.stderr + + +def test_warm_up_rows_are_reported_apart(world): + # A -warm plan: room-1's dial warm-up is turn 0, so its first turn is turn 1. + world["plan"].write_text(textwrap.dedent("""\ + session_id,i,parent,shared,isl,rendered_isl,lcp_parent,osl,delay_ms,timeout,warm + room-1,0,,0,11936,11936,,1,,5.0,1 + room-1,1,,0,12000,12000,0,10,3000,5.0, + room-1,2,1,11990,12100,12100,11990,10,800,5.0, + room-2,0,,0,3000,3000,0,10,,5.0, + """)) + export = world["artifacts"] / "agentic_x_c96/aiperf_artifacts/profile_export.jsonl" + rows = [record("room-1", 0, 700), record("room-1", 1, 120), record("room-1", 2, 100), record("room-2", 0, 1500)] + export.write_text("".join(json.dumps(r) + "\n" for r in rows)) + out = run(world, export, "--plan", world["plan"]) + assert out.returncode == 0, out.stderr + lines = out.stdout.splitlines() + assert lines[1].startswith("all: n=3 ") + assert lines[2].startswith(" first: n= 2 p50=1.50 ") + assert lines[-1] == ("warm-ups (excluded above): n=1 p50=0.70 p90=0.70 p95=0.70 p99=0.70 max=0.70 " + "done <=0.5s=0.0% <=1s=100.0% <=1.5s=100.0% <=2s=100.0%") + + +def test_open_loop_window_replayed_calls_and_ttfs(world): + # openloop.json next to aiperf_artifacts: calls opened in [60, 600) s after the first request are scored. + run_dir = world["artifacts"] / "agentic_x_c96" + (run_dir / "openloop.json").write_text(json.dumps({"score_window_s": [60.0, 600.0]})) + + def timed(room, turn, ttft_ms, start_s, gaps_ms=()): + r = record(room, turn, ttft_ms) + r["metadata"]["request_start_ns"] = int(start_s * 1e9) + r["metrics"]["inter_chunk_latency"] = {"value": list(gaps_ms), "unit": "ms"} + return r + + rows = [timed("room-2", 0, 2000, 0), # ramp: left out + timed("room-1~2", 0, 900, 100), timed("room-1~2", 1, 200, 700, [10] * 11), # opened at 100 s: kept + timed("room-2~3", 0, 400, 650)] # tail: left out + (run_dir / "aiperf_artifacts/profile_export.jsonl").write_text("".join(json.dumps(r) + "\n" for r in rows)) + out = run(world, world["artifacts"], "--plan", world["plan"]) + assert out.returncode == 0, out.stderr + assert "open loop: calls opened in [60, 600) s" in out.stdout + assert "all: n=2 " in out.stdout + # the replayed room-1~2 reads room-1's plan rows; its turn 1 TTFS = 0.2 s + 11 x 10 ms + assert " TTFS (12 tokens): n=1 p50=0.31 " in out.stdout