[NV][AgentX] Add glm5.1-fp8-b200-tilert-agentic (TileRT PD disagg) / [NV][AgentX] 新增 glm5.1-fp8-b200-tilert-agentic（TileRT PD 分离）

benchmarks/multi_node/agentic/glm5.1_fp8_b200_tilert-disagg.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+
+source "$(dirname "$0")/../../benchmark_lib.sh"
+
+check_env_vars \
+    CONC_LIST \
+    DURATION \
+    IMAGE \
+    SPEC_DECODING \
+    MODEL_PATH \
+    PREFILL_NUM_WORKERS \
+    PREFILL_TP \
+    PREFILL_EP \
+    PREFILL_DP_ATTN \
+    DECODE_NUM_WORKERS \
+    DECODE_TP \
+    DECODE_EP \
+    DECODE_DP_ATTN \
+    PREFILL_NODES \
+    DECODE_NODES \
+    FRAMEWORK
+
+require_agentic_kv_offload_none
+
+export MODEL_NAME=glm5
+export TILERT_MODEL_TYPE=glm-5
+
+export DECODE_KV_DTYPE=fp8
+export PREFILL_KV_DTYPE=fp8_ds_mla
+
+export TILERT_PARSER=none
+
+exec bash "$(dirname "$0")/../tilert_utils/submit.sh"

benchmarks/multi_node/tilert_utils/build_queue_wheel.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Build InferenceX's queueing backport from the official TileRT post2 wheel."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import shutil
+import subprocess
+import sys
+import tempfile
+import urllib.request
+import zipfile
+from pathlib import Path
+
+
+UPSTREAM_VERSION = "0.1.5.post2"
+PATCHED_VERSION = "0.1.5.post2+inferencex.1"
+UPSTREAM_WHEEL = "tilert-0.1.5.post2-cp312-cp312-manylinux_2_28_x86_64.whl"
+UPSTREAM_URL = (
+    "https://github.com/tile-ai/TileRT/releases/download/"
+    f"v{UPSTREAM_VERSION}/{UPSTREAM_WHEEL}"
+)
+UPSTREAM_SHA256 = "e65b876ccfc1a419b0047a6d6b395f619ea35c15194ad1892f171c78476fe407"
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def download_upstream(destination: Path) -> None:
+    with urllib.request.urlopen(UPSTREAM_URL) as response:  # noqa: S310 (fixed URL)
+        with destination.open("wb") as output:
+            shutil.copyfileobj(response, output)
+    actual = sha256(destination)
+    if actual != UPSTREAM_SHA256:
+        raise RuntimeError(
+            f"upstream wheel SHA256 mismatch: expected {UPSTREAM_SHA256}, got {actual}"
+        )
+
+
+def update_metadata(unpacked: Path) -> None:
+    old_dist_info = unpacked / f"tilert-{UPSTREAM_VERSION}.dist-info"
+    new_dist_info = unpacked / f"tilert-{PATCHED_VERSION}.dist-info"
+    old_dist_info.rename(new_dist_info)
+    metadata = new_dist_info / "METADATA"
+    text = metadata.read_text()
+    old_version = f"Version: {UPSTREAM_VERSION}\n"
+    if text.count(old_version) != 1:
+        raise RuntimeError("expected exactly one upstream Version field in METADATA")
+    metadata.write_text(text.replace(old_version, f"Version: {PATCHED_VERSION}\n"))
+
+
+def build(output_dir: Path) -> Path:
+    patch = Path(__file__).with_name("patches") / "tilert-0.1.5.post2-queue.patch"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix="tilert-queue-wheel-") as temporary:
+        work = Path(temporary)
+        upstream = work / UPSTREAM_WHEEL
+        unpacked = work / "unpacked"
+        download_upstream(upstream)
+        with zipfile.ZipFile(upstream) as archive:
+            archive.extractall(unpacked)
+        subprocess.run(
+            ["patch", "-p1", "--batch", "--forward", "-i", str(patch)],
+            cwd=unpacked,
+            check=True,
+        )
+        update_metadata(unpacked)
+        subprocess.run(
+            [sys.executable, "-m", "wheel", "pack", "--dest-dir", str(output_dir), "."],
+            cwd=unpacked,
+            check=True,
+        )
+    wheels = list(output_dir.glob("tilert-0.1.5.post2+inferencex.1-*.whl"))
+    if len(wheels) != 1:
+        raise RuntimeError(f"expected one patched wheel, found {len(wheels)}")
+    return wheels[0]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("output_dir", type=Path)
+    args = parser.parse_args()
+    wheel = build(args.output_dir.resolve())
+    print(f"{sha256(wheel)}  {wheel}")
+
+
+if __name__ == "__main__":
+    main()

benchmarks/multi_node/tilert_utils/patches/tilert-0.1.5.post2-queue.patch

@@ -0,0 +1,133 @@
+--- a/tilert/pd_vllm/pd_router.py
++++ b/tilert/pd_vllm/pd_router.py
+@@ -4,7 +4,8 @@
+ non-streaming.
+
+ Flow per request (phase-1 hybrid, see design doc):
+-  1. pick a free decode node (in-memory busy tracking; all busy -> 429)
++  1. wait up to --queue-timeout for a free decode node (in-memory busy
++     tracking; timeout -> 429)
+   2. forward to vLLM with max_tokens=1 + logprobs and inject
+      kv_transfer_params {tilert_host, tilert_ctrl_port} — the connector
+      claims the request and RDMA-sends state to the decode node
+@@ -56,19 +57,27 @@
+ class Pool:
+     def __init__(self, nodes: list[DecodeNode]):
+         self.nodes = nodes
+-        self._lock = threading.Lock()
++        self._available = threading.Condition()
+
+-    def acquire(self) -> DecodeNode | None:
+-        with self._lock:
+-            for n in self.nodes:
+-                if not n.busy:
+-                    n.busy = True
+-                    return n
+-        return None
++    def acquire(self, timeout: float = 0.0) -> DecodeNode | None:
++        deadline = time.monotonic() + timeout
++        with self._available:
++            while True:
++                for n in self.nodes:
++                    if not n.busy:
++                        n.busy = True
++                        return n
++                if timeout <= 0:
++                    return None
++                remaining = deadline - time.monotonic()
++                if remaining <= 0:
++                    return None
++                self._available.wait(timeout=remaining)
+
+     def release(self, node: DecodeNode) -> None:
+-        with self._lock:
++        with self._available:
+             node.busy = False
++            self._available.notify()
+
+
+ def first_token_from_logprobs(resp: dict, is_chat: bool) -> int:
+@@ -123,11 +134,19 @@
+ class RouterCtx:
+     """Immutable per-process context (tokenizer, parser factory, config)."""
+
+-    def __init__(self, vllm_url: str, pool: Pool, tokenizer, parser_name: str):
++    def __init__(
++        self,
++        vllm_url: str,
++        pool: Pool,
++        tokenizer,
++        parser_name: str,
++        queue_timeout: float = 0.0,
++    ):
+         self.vllm_url = vllm_url
+         self.pool = pool
+         self.tokenizer = tokenizer
+         self.parser_name = parser_name
++        self.queue_timeout = queue_timeout
+         self._parsers = {}
+         if parser_name != "none":
+             if tokenizer is None:
+@@ -178,9 +197,12 @@
+     # ── non-streaming ────────────────────────────────────────────────────
+     def _handle(path: str, body: dict):
+         is_chat = path.endswith("chat/completions")
+-        node = pool.acquire()
++        node = pool.acquire(ctx.queue_timeout)
+         if node is None:
+-            return JSONResponse({"error": "all decode nodes busy"}, status_code=429)
++            return JSONResponse(
++                {"error": f"all decode nodes busy after {ctx.queue_timeout:g}s"},
++                status_code=429,
++            )
+         t0 = time.time()
+         try:
+             prefill = _prefill(path, body, node)
+@@ -257,9 +280,12 @@
+     async def _handle_stream(path: str, body: dict, request: Request):
+         from starlette.concurrency import run_in_threadpool
+
+-        node = pool.acquire()
++        node = await run_in_threadpool(pool.acquire, ctx.queue_timeout)
+         if node is None:
+-            return JSONResponse({"error": "all decode nodes busy"}, status_code=429)
++            return JSONResponse(
++                {"error": f"all decode nodes busy after {ctx.queue_timeout:g}s"},
++                status_code=429,
++            )
+
+         try:
+             prefill = await run_in_threadpool(_prefill, path, body, node)
+@@ -445,7 +472,15 @@
+         default="glm47",
+         help="output parser (reasoning + tool calls)",
+     )
++    ap.add_argument(
++        "--queue-timeout",
++        type=float,
++        default=0.0,
++        help="seconds to wait for a free decode node before returning HTTP 429",
++    )
+     args = ap.parse_args()
++    if args.queue_timeout < 0:
++        ap.error("--queue-timeout must be non-negative")
+
+     nodes = []
+     for spec in args.decode:
+@@ -460,14 +495,15 @@
+             args.model_path, trust_remote_code=True
+         )  # nosec B615
+
+-    ctx = RouterCtx(args.vllm_url, Pool(nodes), tokenizer, args.parser)
++    ctx = RouterCtx(args.vllm_url, Pool(nodes), tokenizer, args.parser, args.queue_timeout)
+     app = build_app(ctx)
+     logger.info(
+-        "router on :%d -> vllm=%s, %d decode node(s), parser=%s",
++        "router on :%d -> vllm=%s, %d decode node(s), parser=%s, queue_timeout=%gs",
+         args.port,
+         args.vllm_url,
+         len(nodes),
+         args.parser,
++        args.queue_timeout,
+     )
+     uvicorn.run(app, host=args.host, port=args.port, log_level="warning")

benchmarks/multi_node/tilert_utils/run_node.sh

@@ -23,6 +23,18 @@ PREFILL_KV_DTYPE=${PREFILL_KV_DTYPE:-fp8_ds_mla}
 PREFILL_SPEC=(--speculative-config '{"method":"mtp","num_speculative_tokens":1}')
 DECODE_MTP=(--with-mtp)
 
+TILERT_IS_AGENTIC=0
+if [[ "${IS_AGENTIC:-0}" == "1" || "${SCENARIO_TYPE:-}" == "agentic-coding" ]]; then
+    TILERT_IS_AGENTIC=1
+fi
+
+if [[ "$TILERT_IS_AGENTIC" == "1" ]]; then
+    TILERT_QUEUE_TIMEOUT=${TILERT_QUEUE_TIMEOUT:-1800}
+fi
+TILERT_QUEUE_TIMEOUT=${TILERT_QUEUE_TIMEOUT:-0}
+
+AGENTIC_LOGS_DIR=${AGENTIC_LOGS_DIR:-$RESULT_DIR/LOGS/agentic}
+
 : "${DECODE_HOST:?DECODE_HOST is unset -- submit.sh must export it}"
 : "${PREFILL_HOST:?PREFILL_HOST is unset -- submit.sh must export it}"
 : "${TILERT_ROLE:?TILERT_ROLE is unset -- submit.sh must set it to decode or prefill}"
@@ -156,8 +168,10 @@ start_decode() {
 }
 
 start_prefill() {
+    local served=("$MODEL_NAME")
+    [[ -n "${MODEL:-}" && "$MODEL" != "$MODEL_NAME" ]] && served+=("$MODEL")
     local cmd=(vllm serve "$MODEL_PATH"
-        --served-model-name "$MODEL_NAME" --port "$PREFILL_PORT"
+        --served-model-name "${served[@]}" --port "$PREFILL_PORT"
         --tensor-parallel-size "$PREFILL_TP" --max-model-len "$MAX_MODEL_LEN"
         --enforce-eager --trust-remote-code --return-tokens-as-token-ids
         --gpu-memory-utilization "$GPU_MEM_UTIL" --kv-cache-dtype "$PREFILL_KV_DTYPE"
@@ -171,7 +185,8 @@ start_router() {
     local cmd=(env CUDA_VISIBLE_DEVICES= "${PY:-python}" -m tilert.pd_vllm.pd_router
         --vllm-url "http://$PREFILL_HOST:$PREFILL_PORT"
         --decode "$DECODE_HOST:$DECODE_CTRL_PORT:$DECODE_HTTP_PORT"
-        --port "$ROUTER_PORT" --model-path "$MODEL_PATH" --parser "$TILERT_PARSER")
+        --port "$ROUTER_PORT" --model-path "$MODEL_PATH" --parser "$TILERT_PARSER"
+        --queue-timeout "$TILERT_QUEUE_TIMEOUT")
     log_and_run_bg router "$BENCHMARK_LOGS_DIR/tilert_router.log" "${cmd[@]}"
     ROUTER_PID=$LAST_BG_PID
 }
@@ -211,16 +226,37 @@ run_bench_and_eval() {
             --result-filename "$(bench_result_stem "$conc")" --result-dir "$RESULT_DIR" \
             || { rc=$?; echo "[bench] WARNING: conc=$conc failed/timed out (rc=$rc)"; }
     done
-    if [[ "${RUN_EVAL}" = "true" ]]; then
-        if [[ -n "${EVAL_CONC:-}" ]]; then
-            export EVAL_CONCURRENT_REQUESTS="$EVAL_CONC"
-        else
-            export EVAL_CONCURRENT_REQUESTS="$(tr ' ' '\n' <<< "$CONC_LIST" | sort -n | tail -1)"
-        fi
-        export CONC="$EVAL_CONCURRENT_REQUESTS"
-        run_eval --framework lm-eval --port "$ROUTER_PORT"
-        append_lm_eval_summary
+    run_lm_eval
+    return $rc
+}
+
+run_lm_eval() {
+    [[ "${RUN_EVAL}" = "true" ]] || return 0
+    if [[ -n "${EVAL_CONC:-}" ]]; then
+        export EVAL_CONCURRENT_REQUESTS="$EVAL_CONC"
+    else
+        export EVAL_CONCURRENT_REQUESTS="$(tr ' ' '\n' <<< "$CONC_LIST" | sort -n | tail -1)"
     fi
+    export CONC="$EVAL_CONCURRENT_REQUESTS"
+    run_eval --framework lm-eval --port "$ROUTER_PORT"
+    append_lm_eval_summary
+}
+
+run_agentic_replay() {
+    wait_for_server_ready --port "$ROUTER_PORT" \
+        --server-log "$BENCHMARK_LOGS_DIR/tilert_router.log" --server-pid "$ROUTER_PID"
+    local rc=0 conc conc_result_dir
+    local result_filename_base="$RESULT_FILENAME"
+    for conc in $CONC_LIST; do
+        conc_result_dir="$AGENTIC_LOGS_DIR/conc_${conc}"
+        mkdir -p "$conc_result_dir"
+        export CONC="$conc"
+        export RESULT_FILENAME="${result_filename_base}_conc${conc}"
+        build_replay_cmd "$conc_result_dir"
+        run_agentic_replay_and_write_outputs "$conc_result_dir" \
+            || { rc=$?; echo "[agentic] WARNING: conc=$conc failed/timed out (rc=$rc)"; }
+    done
+    export RESULT_FILENAME="$result_filename_base"
     return $rc
 }
 
@@ -247,13 +283,21 @@ case "$TILERT_ROLE" in
     prefill)
         rdma_preflight || exit 1
         rm -f "$DONE_SENTINEL"
+        if [[ "$TILERT_IS_AGENTIC" == "1" ]]; then
+            resolve_trace_source
+            install_agentic_deps
+        fi
         wait_for_tcp "$DECODE_HOST" "$DECODE_CTRL_PORT" "$DECODE_WAIT" \
             || echo "[prefill] WARNING: timed out waiting for the decode ctrl port ($DECODE_HOST:$DECODE_CTRL_PORT), starting anyway"
         start_prefill
         wait_for_tcp "$PREFILL_HOST" "$PREFILL_PORT" "${PREFILL_WAIT:-3600}" \
             || echo "[prefill] WARNING: timed out waiting for the vLLM port ($PREFILL_HOST:$PREFILL_PORT), continuing (see $BENCHMARK_LOGS_DIR/tilert_prefill.log)"
         start_router
-        run_bench_and_eval; BENCH_RC=$?
+        if [[ "$TILERT_IS_AGENTIC" == "1" ]]; then
+            run_agentic_replay; BENCH_RC=$?
+        else
+            run_bench_and_eval; BENCH_RC=$?
+        fi
         touch "$DONE_SENTINEL"
         kill "$ROUTER_PID" "$PREFILL_PID" 2>/dev/null || true
         exit $BENCH_RC