Files
agentic-kvc/replayer/__main__.py
Gahow Wang 48ae72467a Replayer: closed-loop inter-turn think-time mode
Add --inter-turn-think (env REPLAY_INTER_TURN_THINK_S): turn 1 fires on
session admission, each later turn a FIXED think-time after the previous
turn COMPLETES, ignoring absolute trace timestamps. Combined with
--max-inflight-sessions (env REPLAY_MAX_INFLIGHT) this is a stable N-user
closed loop, removing the open-loop "fire immediately because timestamp is
in the past" retrigger artifact. Needed for the dispatch-coupling
(wall-clock amplification) sweep.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-05-29 18:19:12 +08:00

65 lines
2.7 KiB
Python

"""CLI entry point: python -m replayer replay ..."""
from __future__ import annotations
import argparse
import asyncio
import logging
import os
from pathlib import Path
from .replay import ReplayConfig, replay_trace
def main() -> None:
p = argparse.ArgumentParser(description="Trace replayer for vLLM benchmarking")
p.add_argument("--trace", type=Path, required=True, help="Sampled trace JSONL")
p.add_argument("--output", type=Path, required=True, help="Output metrics JSONL")
p.add_argument("--endpoint", type=str, required=True,
help="vLLM server URL (e.g. http://localhost:8000)")
p.add_argument("--model", type=str, default="default", help="Model name for API")
p.add_argument("--concurrency-limit", type=int, default=2000,
help="Max concurrent HTTP requests (safety limit)")
_env_inflight = os.environ.get("REPLAY_MAX_INFLIGHT")
p.add_argument("--max-inflight-sessions", type=int,
default=int(_env_inflight) if _env_inflight else None,
help="Cap on concurrent sessions (None = unlimited; "
"trace-driven dispatch otherwise). Env: REPLAY_MAX_INFLIGHT")
_env_think = os.environ.get("REPLAY_INTER_TURN_THINK_S")
p.add_argument("--inter-turn-think", type=float,
default=float(_env_think) if _env_think else None,
help="Closed-loop think-time (s) after each turn completes; "
"ignore absolute trace schedule. Env: REPLAY_INTER_TURN_THINK_S")
p.add_argument("--request-timeout", type=float, default=600.0)
p.add_argument("--request-limit", type=int, default=None,
help="Limit number of requests to replay")
p.add_argument("-v", "--verbose", action="store_true")
args = p.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
config = ReplayConfig(
trace_path=args.trace,
output_path=args.output,
endpoint_url=args.endpoint.rstrip("/"),
model_name=args.model,
concurrency_limit=args.concurrency_limit,
request_timeout_s=args.request_timeout,
request_limit=args.request_limit,
max_inflight_sessions=args.max_inflight_sessions,
inter_turn_think_s=args.inter_turn_think,
)
results = asyncio.run(replay_trace(config))
succeeded = sum(1 for r in results if r.error is None)
print(f"\nDone: {succeeded}/{len(results)} requests succeeded")
print(f"Metrics: {args.output}")
print(f"Summary: {args.output.with_suffix('.summary.json')}")
if __name__ == "__main__":
main()