diff --git a/examples/10_DeepSeekV4Pro_Example/README.md b/examples/10_DeepSeekV4Pro_Example/README.md new file mode 100644 index 000000000..fbebd1391 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/README.md @@ -0,0 +1,286 @@ +# DeepSeek-V4-Pro Benchmark (SGLang) + +End-to-end example for benchmarking `deepseek-ai/DeepSeek-V4-Pro` with SGLang, using the same +performance and accuracy datasets as the [GPT-OSS-120B example](../04_GPTOSS120B_Example/Readme.md) +(AIME25, GPQA, LiveCodeBench). + +The server uses `/v1/chat/completions` (`api_type: openai`) with text prompts from the dataset. +The server applies the DeepSeek-V4 chat template and reasoning parser. + +## Getting the Dataset + +The performance dataset must be obtained from the LLM task-force (parquet format). Place it at: + +``` +examples/04_GPTOSS120B_Example/data/perf_eval_ref.parquet +``` + +The accuracy datasets (AIME25, GPQA, LiveCodeBench) are downloaded automatically from HuggingFace. + +## Environment Setup + +```bash +export HF_HOME= +export HF_TOKEN= # required for GPQA (gated) and faster HF downloads +export MODEL_NAME=deepseek-ai/DeepSeek-V4-Pro +export MODEL_DIR=/data/workloads-inference/models +export MODEL_PATH=${MODEL_DIR}/deepseek-ai/DeepSeek-V4-Pro +export TOKENIZER_MODEL_PATH=${MODEL_PATH} # host path for ISL/OSL/TPOT metrics +``` + +Preflight scripts (`run_benchmark.sh`, `run_sglang_benchmark.sh`, `run_*_accuracy_benchmark.sh`) probe the inference server with `GET /health` and `GET /v1/models`. Override the base URL or wait time while a server is starting: + +```bash +export SGLANG_BASE_URL=http://127.0.0.1:30000 # default when SGLANG_PORT=30000 +export WAIT_FOR_SGLANG_S=120 # seconds; 0 = single attempt +``` + +## Download Model + +Download weights to the shared model store and mount them into the serving container: + +```bash +mkdir -p "${MODEL_PATH}" +hf download "${MODEL_NAME}" --local-dir "${MODEL_PATH}" +``` + +--- + +## SGLang (ROCm / MI35x) + +SGLang serves DeepSeek-V4-Pro on ROCm using the DSv4-tuned image and FP4-experts env flags from the +`amd/deepseek_v4` branch. Before launch, patch `config.json` so `model_type` is `deepseek_v3` +(SGLang registry compatibility). + +### Launch Server + +**Option A: helper script (host or container)** + +On a ROCm host with SGLang installed: + +```bash +export MODEL_PATH=/data/workloads-inference/models/deepseek-ai/DeepSeek-V4-Pro +export SGLANG_PORT=30000 +export TP=8 +export CONC=512 +./examples/10_DeepSeekV4Pro_Example/start_sglang_server.sh +``` + +Inside the DSv4 Docker image (model directory must exist on the host): + +```bash +export MODEL_PATH=/data/workloads-inference/models/deepseek-ai/DeepSeek-V4-Pro +export RUN_MODE=docker +export SGLANG_IMAGE=rocm/sgl-dev:rocm720-mi35x-f96ac98-20260526-DSv4 +./examples/10_DeepSeekV4Pro_Example/start_sglang_server.sh +``` + +Optional overrides: + +| Variable | Default | Description | +| --------------------------- | --------- | ----------------------------------------------------------------------------------------------------------------------------- | +| `SGLANG_PORT` / `HTTP_PORT` | `30000` | HTTP listen port (`start_sglang_server.sh` unsets `SGLANG_PORT` before launch — SGLang uses that name for internal ZMQ ports) | +| `TP` | `8` | Tensor parallel size | +| `CONC` | `512` | `--max-running-requests` | +| `MAX_MODEL_LEN` | `98304` | `--context-length` | +| `DP_ATTENTION` | `false` | Set `true` to enable DP attention | +| `EP_SIZE` | `1` | Expert parallel size (`>1` adds `--ep-size`) | +| `CHAT_TEMPLATE` | _(unset)_ | Optional path to `deepseek_v4_thinking.jinja` | + +The script exports the FP4-experts ROCm flags (`SGLANG_DSV4_FP4_EXPERTS=True`, +`SGLANG_FORCE_TRITON_MOE_FP8=0`, `SGLANG_REASONING_EFFORT=max`, etc.) and launches: + +```text +python3 -m sglang.launch_server \ + --model-path $MODEL \ + --tensor-parallel-size $TP \ + --attention-backend compressed \ + --reasoning-parser deepseek-v4 \ + --tool-call-parser deepseekv4 \ + ... +``` + +**Option B: manual launch** (same flags as `start_sglang_server.sh`) + +```bash +# Patch HF config (once per cache checkout) +python3 <<'PYEOF' +import json +from huggingface_hub import hf_hub_download +path = hf_hub_download(repo_id="deepseek-ai/DeepSeek-V4-Pro", filename="config.json") +with open(path) as f: + config = json.load(f) +if config.get("model_type") == "deepseek_v4": + config["model_type"] = "deepseek_v3" + with open(path, "w") as f: + json.dump(config, f, indent=2) +PYEOF + +# Export env block from start_sglang_server.sh, then: +python3 -m sglang.launch_server \ + --model-path "${MODEL_PATH}" \ + --host 0.0.0.0 \ + --port 30000 \ + --tensor-parallel-size 8 \ + --trust-remote-code \ + --disable-radix-cache \ + --attention-backend compressed \ + --max-running-requests 512 \ + --mem-fraction-static 0.90 \ + --swa-full-tokens-ratio 0.15 \ + --page-size 256 \ + --context-length 98304 \ + --chunked-prefill-size 8192 \ + --disable-shared-experts-fusion \ + --tool-call-parser deepseekv4 \ + --reasoning-parser deepseek-v4 \ + --watchdog-timeout 1800 +``` + +Verify: + +```bash +curl http://127.0.0.1:30000/health +``` + +### Run Benchmark (SGLang) + +[`sglang_deepseek_v4_pro_example.yaml`](sglang_deepseek_v4_pro_example.yaml) targets +`http://localhost:30000` with the performance + AIME25 + GPQA + LiveCodeBench datasets: + +```bash +uv run inference-endpoint benchmark from-config \ + -c examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_example.yaml \ + --timeout 60 +``` + +Or use the helper script: + +```bash +./examples/10_DeepSeekV4Pro_Example/run_sglang_benchmark.sh +``` + +Performance-only: + +```bash +uv run inference-endpoint benchmark from-config \ + -c examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_perf.yaml \ + --timeout 60 +``` + +### Run Accuracy (SGLang) + +Same workflow as [GPT-OSS `run.py`](../04_GPTOSS120B_Example/Readme.md#accuracy-suite-script): +start SGLang, start `lcb-service`, set `HF_TOKEN` (required for gated GPQA), then run the +accuracy helper (checks all prerequisites, tees logs under `results/docker_logs/accuracy/`): + +```bash +export HF_TOKEN= +export HF_HOME=~/.cache/huggingface + +./examples/10_DeepSeekV4Pro_Example/start_sglang_server.sh +./examples/10_DeepSeekV4Pro_Example/start_lcb_service.sh +./examples/10_DeepSeekV4Pro_Example/run_sglang_accuracy_benchmark.sh +``` + +YAML-only (equivalent): + +```bash +uv run inference-endpoint benchmark from-config \ + -c examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_accuracy.yaml \ + --timeout 3600 +``` + +| Argument / env | Default | Description | +| ----------------------- | ------------ | -------------------------------------------- | +| `HF_TOKEN` | _(required)_ | HuggingFace token for GPQA download | +| `TIMEOUT` | `86400` | Benchmark timeout (seconds) | +| `DOCKER_LOG_STORAGE_GB` | `64` | Container writable layer size when supported | + +Accuracy config uses `max_new_tokens: 320000`, `num_workers: 32`, `num_repeats: 4` per dataset, +and phase order AIME25 → GPQA → LiveCodeBench. + +### Docker log storage + +Server and LCB containers mount `results/docker_logs//` on the host at `/workspace`. +Scripts also pass `--storage-opt size=16G` on `docker run` when the daemon uses overlay2 +(override with `DOCKER_LOG_STORAGE_GB`). SGLang server stdout is written to +`results/docker_logs/sglang/server.log`. + +### Config notes + +- **Performance dataset**: `text_input` → `prompt`. Server applies DeepSeek-V4 chat template. +- **Accuracy datasets**: `::deepseek_v4` presets (same prompt formatting as GPT-OSS). +- **Reasoning output**: server streams reasoning separately; client accumulates `reasoning_content` + and final `content` for scoring. +- **`model_params.name`**: use the HuggingFace id (`deepseek-ai/DeepSeek-V4-Pro`). Set + `TOKENIZER_MODEL_PATH` to the host weights path for ISL/OSL/TPOT. + +--- + +## LiveCodeBench Setup + +LiveCodeBench has dependency conflicts with the main package. Two options: + +### Option A: containerized scorer (recommended when `dhi.io` access is available) + +Follow the [LiveCodeBench README](../../src/inference_endpoint/evaluation/livecodebench/README.md#running-the-container). +Requires `docker login dhi.io`, then: + +```bash +./examples/10_DeepSeekV4Pro_Example/start_lcb_service.sh +curl http://127.0.0.1:13835/info +``` + +### Option B: local subprocess scorer (no `docker login dhi.io`) + +Same fallback as the GPT-OSS example — runs `lcb_serve` as a subprocess on the host: + +```bash +export ALLOW_LCB_LOCAL_EVAL=true +./examples/10_DeepSeekV4Pro_Example/run_sglang_accuracy_benchmark.sh +``` + +The accuracy helper skips the `:13835` preflight when this is set (default `true`). +WebSocket scoring is attempted first if `lcb-service` is up; otherwise scoring falls +back to the subprocess path automatically. + +--- + +## Troubleshooting + +**Cannot connect to SGLang server** + +- Verify: `curl http://localhost:30000/health` +- Confirm `SGLANG_PORT` matches `endpoint_config.endpoints` in the SGLang YAML +- For ROCm, use the DSv4 image: `rocm/sgl-dev:rocm720-mi35x-f96ac98-20260526-DSv4` + +**SGLang `address already in use` on `--port`** + +- Do not export `SGLANG_PORT` into the SGLang process (upstream reserves it for ZMQ). + Use `start_sglang_server.sh`, which passes `--port` on the CLI and runs with + `env -u SGLANG_PORT`. + +**SGLang fails to load model / unknown architecture** + +- Run the `config.json` patch (`model_type`: `deepseek_v4` → `deepseek_v3`) before launch +- Confirm `SGLANG_DSV4_FP4_EXPERTS=True` for the original HF checkpoint with MXFP4 experts + +**LiveCodeBench scoring fails / Connection refused on port 13835** + +- Start `lcb-service` (`docker login dhi.io` + `start_lcb_service.sh`), or +- `export ALLOW_LCB_LOCAL_EVAL=true` and re-run (see LiveCodeBench Setup above) + +**Out of memory** + +- Increase `TP` / `--tensor-parallel-size` +- Lower `--mem-fraction-static` + +**Model not found in container** + +- Confirm the host path exists: `ls "${MODEL_PATH}"` +- Mount into the container at the path used by `--model-path` + +**docker build fails with `dhi.io ... unauthorized`** + +- Run `docker login dhi.io` with your Docker Hub credentials (PAT with read access to hardened images) diff --git a/examples/10_DeepSeekV4Pro_Example/docker_common.sh b/examples/10_DeepSeekV4Pro_Example/docker_common.sh new file mode 100644 index 000000000..5fc04ea4b --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/docker_common.sh @@ -0,0 +1,50 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Shared Docker helpers for DeepSeek-V4-Pro example scripts. + +# Writable log directory on the host (mounted into containers at /workspace). +# Accuracy + server logs can grow large; default leaves headroom on the host. +DOCKER_LOG_STORAGE_GB="${DOCKER_LOG_STORAGE_GB:-16}" + +ensure_docker_log_dir() { + local subdir="${1:-misc}" + LOG_DIR="${LOG_DIR:-${ENDPOINTS_DIR:-.}/results/docker_logs/${subdir}}" + mkdir -p "${LOG_DIR}" + export LOG_DIR +} + +# Extra docker run args for a larger container writable layer (opt-in only). +# Logs are written to the mounted host LOG_DIR; most hosts use overlay2 without xfs +# pquota and reject --storage-opt. +docker_storage_args() { + if [[ "${DOCKER_USE_LOG_STORAGE_OPT:-false}" == "true" ]]; then + # shellcheck disable=SC2207 + echo --storage-opt "size=${DOCKER_LOG_STORAGE_GB}G" + fi +} + +# Wait for an OpenAI-compatible or SGLang HTTP server (example script preflight). +# Tries GET /health (SGLang native), then GET /v1/models (OpenAI compatibility). +# Args: base_url [max_wait_seconds] +wait_openai_compatible_server() { + local base="${1%/}" + local max_wait="${2:-0}" + local start + start=$(date +%s) + while true; do + for path in /health /v1/models; do + if curl --output /dev/null --silent --fail --max-time 5 "${base}${path}"; then + echo "Inference server ready (${base}${path})" + return 0 + fi + done + if (( max_wait <= 0 )); then + return 1 + fi + if (( $(date +%s) - start >= max_wait )); then + return 1 + fi + sleep 2 + done +} diff --git a/examples/10_DeepSeekV4Pro_Example/launch_accuracy_background.sh b/examples/10_DeepSeekV4Pro_Example/launch_accuracy_background.sh new file mode 100755 index 000000000..349becf8a --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/launch_accuracy_background.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +# Start run_sglang_accuracy_benchmark.sh under nohup with a stable log path. +# (Avoids `VAR=... && nohup ... &` where bash backgrounds the whole `&&` chain so +# follow-up commands in the parent shell lose VAR.) +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +# shellcheck source=docker_common.sh +source "${SCRIPT_DIR}/docker_common.sh" + +ensure_docker_log_dir "accuracy" +LOGF="${LOG_DIR}/nohup_accuracy_$(date +%Y%m%d_%H%M%S).log" +export WAIT_FOR_SGLANG_S="${WAIT_FOR_SGLANG_S:-120}" +export PYTHONUNBUFFERED="${PYTHONUNBUFFERED:-1}" + +nohup "${SCRIPT_DIR}/run_sglang_accuracy_benchmark.sh" >"${LOGF}" 2>&1 & +echo "Started accuracy benchmark PID=$!" +echo "Wrapper log: ${LOGF}" +echo "Tee log (inside run): ${LOG_DIR}/accuracy_from_config.log" diff --git a/examples/10_DeepSeekV4Pro_Example/monitor_accuracy_run.sh b/examples/10_DeepSeekV4Pro_Example/monitor_accuracy_run.sh new file mode 100755 index 000000000..3dfb1ac56 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/monitor_accuracy_run.sh @@ -0,0 +1,81 @@ +#!/usr/bin/env bash +# Monitor the running SGLang DeepSeek-V4-Pro accuracy benchmark. +# Emits sentinel lines (ALERT_HANG / RUN_FAILED / RUN_FINISHED_OK) so an external +# watcher can be notified, and logs periodic status to a monitor log. +set -uo pipefail + +PROC_PAT="${PROC_PAT:-from-config.*sglang_deepseek_v4_pro_accuracy}" +WRAPPER_LOG="${WRAPPER_LOG:?set WRAPPER_LOG to the nohup wrapper log path}" +REPORT_DIR="${REPORT_DIR:-results/sglang_deepseek_v4_pro_accuracy}" +INTERVAL_S="${INTERVAL_S:-300}" +# Consecutive idle (no new completes + GPU idle) checks before declaring a hang. +HANG_STRIKES="${HANG_STRIKES:-3}" +GPU_IDLE_PCT="${GPU_IDLE_PCT:-5}" + +ts() { date '+%Y-%m-%d %H:%M:%S'; } + +live_events_file() { + # Derive the event logger's tmpfs events.jsonl from its --log-dir arg. + local logdir + logdir=$(pgrep -af "services.event_logger" \ + | grep -oE -- "--log-dir [^ ]+" | head -1 | awk '{print $2}') + [[ -n "${logdir}" ]] && echo "${logdir}/events.jsonl" +} + +gpu_busy_pct() { + rocm-smi --showuse 2>/dev/null \ + | grep -oE "GPU use \(%\): [0-9]+" | grep -oE "[0-9]+$" \ + | sort -rn | head -1 +} + +completed_count() { + local f="$1" c + if [[ -f "${f}" ]]; then + # grep -c always prints a count (0 when no match); capture it so a non-zero + # exit status on 0 matches does not also trigger a fallback echo. + c=$(grep -c '"event_type":"sample.complete"' "${f}" 2>/dev/null) + echo "${c:-0}" + else + echo 0 + fi +} + +prev_completed=-1 +strikes=0 + +echo "[$(ts)] monitor start: pat='${PROC_PAT}' interval=${INTERVAL_S}s wrapper=${WRAPPER_LOG}" + +while true; do + if ! pgrep -f "${PROC_PAT}" >/dev/null 2>&1; then + # Process gone — classify success vs failure from the wrapper log. + if grep -q "Score for livecodebench" "${WRAPPER_LOG}" 2>/dev/null \ + || grep -q "Saved: ${REPORT_DIR}/results.json" "${WRAPPER_LOG}" 2>/dev/null; then + echo "[$(ts)] RUN_FINISHED_OK" + else + echo "[$(ts)] RUN_FAILED (process exited without final score)" + echo "---- wrapper tail ----" + tr '\r' '\n' < "${WRAPPER_LOG}" 2>/dev/null | tail -25 + fi + exit 0 + fi + + ev=$(live_events_file) + done_n=$(completed_count "${ev:-/nonexistent}") + gpu=$(gpu_busy_pct); gpu="${gpu:-0}" + echo "[$(ts)] alive completed=${done_n} gpu_busy=${gpu}% events=${ev:-none}" + + # Hang heuristic: no new completions AND GPUs idle for HANG_STRIKES intervals. + if [[ "${done_n}" == "${prev_completed}" && "${gpu}" -lt "${GPU_IDLE_PCT}" ]]; then + strikes=$((strikes + 1)) + echo "[$(ts)] no-progress strike ${strikes}/${HANG_STRIKES} (completed unchanged at ${done_n}, gpu ${gpu}%)" + if [[ "${strikes}" -ge "${HANG_STRIKES}" ]]; then + echo "[$(ts)] ALERT_HANG completed stuck at ${done_n}, gpu ${gpu}% for ${strikes} checks" + echo "---- wrapper tail ----" + tr '\r' '\n' < "${WRAPPER_LOG}" 2>/dev/null | tail -15 + fi + else + strikes=0 + fi + prev_completed="${done_n}" + sleep "${INTERVAL_S}" +done diff --git a/examples/10_DeepSeekV4Pro_Example/run_benchmark.sh b/examples/10_DeepSeekV4Pro_Example/run_benchmark.sh new file mode 100755 index 000000000..5e0032570 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/run_benchmark.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +# Run DeepSeek-V4-Pro benchmark the same way as examples/04_GPTOSS120B_Example: +# 1. SGLang server on :30000 +# 2. lcb-service container on :13835 (for LiveCodeBench scoring) +# 3. inference-endpoint benchmark from-config +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ENDPOINTS_DIR="${ENDPOINTS_DIR:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +# shellcheck source=docker_common.sh +source "${SCRIPT_DIR}/docker_common.sh" +CONFIG="${CONFIG:-${ENDPOINTS_DIR}/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_example.yaml}" +SGLANG_PORT="${SGLANG_PORT:-30000}" +LCB_PORT="${LCB_PORT:-13835}" +TIMEOUT="${TIMEOUT:-60}" +MODE="${MODE:-}" # empty = default (perf for online); set to 'both' for perf + accuracy collection + +cd "${ENDPOINTS_DIR}" + +if [[ -z "${TOKENIZER_MODEL_PATH:-}" ]]; then + export TOKENIZER_MODEL_PATH="${MODEL_PATH:-/data/workloads-inference/models/deepseek-ai/DeepSeek-V4-Pro}" +fi + +echo "=== Pre-flight checks ===" + +SGLANG_BASE_URL="${SGLANG_BASE_URL:-http://127.0.0.1:${SGLANG_PORT}}" +WAIT_FOR_SGLANG_S="${WAIT_FOR_SGLANG_S:-0}" + +if ! wait_openai_compatible_server "${SGLANG_BASE_URL}" "${WAIT_FOR_SGLANG_S}"; then + echo "ERROR: Inference server not reachable at ${SGLANG_BASE_URL} (tried GET /health and GET /v1/models)." >&2 + echo " Start the SGLang server first (see examples/10_DeepSeekV4Pro_Example/README.md)." >&2 + echo " Smoke test: uv run python -m inference_endpoint.testing.echo_server --host 127.0.0.1 --port ${SGLANG_PORT}" >&2 + echo " If the server is slow to bind: export WAIT_FOR_SGLANG_S=120" >&2 + exit 1 +fi + +if ! curl --output /dev/null --silent --fail "http://127.0.0.1:${LCB_PORT}/info"; then + echo "ERROR: lcb-service is not running on port ${LCB_PORT}." + echo "" + echo "Build and start it per:" + echo " src/inference_endpoint/evaluation/livecodebench/README.md#running-the-container" + echo "" + echo "Quick start (after 'docker login dhi.io'):" + echo " docker build -f src/inference_endpoint/evaluation/livecodebench/lcb_serve.dockerfile \\" + echo " --secret id=HF_TOKEN,env=HF_TOKEN -t lcb-service \\" + echo " src/inference_endpoint/evaluation/livecodebench" + echo " docker run --name lcb-service --rm -p 127.0.0.1:${LCB_PORT}:13835 lcb-service:latest" + exit 1 +fi +echo "lcb-service OK on port ${LCB_PORT}" + +echo "" +echo "=== Running benchmark ===" +CMD=(uv run inference-endpoint benchmark from-config -c "${CONFIG}" --timeout "${TIMEOUT}") +if [[ -n "${MODE}" ]]; then + CMD+=(--mode "${MODE}") +fi +echo "${CMD[*]}" +"${CMD[@]}" diff --git a/examples/10_DeepSeekV4Pro_Example/run_sglang_accuracy_benchmark.sh b/examples/10_DeepSeekV4Pro_Example/run_sglang_accuracy_benchmark.sh new file mode 100755 index 000000000..40213f596 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/run_sglang_accuracy_benchmark.sh @@ -0,0 +1,111 @@ +#!/usr/bin/env bash +# Accuracy-only DeepSeek-V4-Pro benchmark against SGLang (GPQA + AIME25 + LCB). +# Workflow mirrors examples/04_GPTOSS120B_Example/run.py + from-config accuracy YAML. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ENDPOINTS_DIR="${ENDPOINTS_DIR:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +# shellcheck source=docker_common.sh +source "${SCRIPT_DIR}/docker_common.sh" + +CONFIG="${CONFIG:-${SCRIPT_DIR}/sglang_deepseek_v4_pro_accuracy.yaml}" +SGLANG_PORT="${SGLANG_PORT:-30000}" +LCB_PORT="${LCB_PORT:-13835}" +# CLI wall-clock for from-config (multi-hour accuracy + LCB); override if needed. +TIMEOUT="${TIMEOUT:-86400}" +# Host log volume for docker --storage-opt when DOCKER_USE_LOG_STORAGE_OPT=true. +export DOCKER_LOG_STORAGE_GB="${DOCKER_LOG_STORAGE_GB:-64}" +# Accuracy phases drain with no timeout by default (settings.drain.accuracy_timeout_s). +export ALLOW_LCB_LOCAL_EVAL="${ALLOW_LCB_LOCAL_EVAL:-true}" + +cd "${ENDPOINTS_DIR}" + +if [[ -z "${HF_TOKEN:-}" && -f "${HF_HOME:-${HOME}/.cache/huggingface}/token" ]]; then + HF_TOKEN="$(cat "${HF_HOME:-${HOME}/.cache/huggingface}/token")" + export HF_TOKEN +fi +if [[ -z "${HF_TOKEN:-}" ]]; then + echo "ERROR: HF_TOKEN is required (GPQA is a gated HuggingFace dataset)." + echo " export HF_TOKEN=" + exit 1 +fi +export HUGGING_FACE_HUB_TOKEN="${HF_TOKEN}" +export HF_HOME="${HF_HOME:-${HOME}/.cache/huggingface}" + +if [[ -z "${TOKENIZER_MODEL_PATH:-}" ]]; then + export TOKENIZER_MODEL_PATH="${MODEL_PATH:-/data/workloads-inference/models/DeepSeek-V4-Pro}" +fi + +ensure_docker_log_dir "accuracy" + +export LCB_DATASETS_DIR="${LCB_DATASETS_DIR:-${ENDPOINTS_DIR}/dataset_cache/livecodebench/release_v6}" + +echo "=== Pre-flight checks ===" + +SGLANG_BASE_URL="${SGLANG_BASE_URL:-http://127.0.0.1:${SGLANG_PORT}}" +WAIT_FOR_SGLANG_S="${WAIT_FOR_SGLANG_S:-0}" + +if ! wait_openai_compatible_server "${SGLANG_BASE_URL}" "${WAIT_FOR_SGLANG_S}"; then + echo "ERROR: Inference server not reachable at ${SGLANG_BASE_URL} (tried GET /health and GET /v1/models)." >&2 + echo " Start SGLang: ${SCRIPT_DIR}/start_sglang_server.sh" >&2 + echo " Smoke test: uv run python -m inference_endpoint.testing.echo_server --host 127.0.0.1 --port ${SGLANG_PORT}" >&2 + echo " If the server is slow to bind: export WAIT_FOR_SGLANG_S=120" >&2 + exit 1 +fi + +_allow_lcb_local=false +case "${ALLOW_LCB_LOCAL_EVAL:-}" in + true | 1 | yes | TRUE | YES) _allow_lcb_local=true ;; +esac + +if [[ "${_allow_lcb_local}" == "true" ]]; then + export ALLOW_LCB_LOCAL_EVAL=true + echo "ALLOW_LCB_LOCAL_EVAL=true — LiveCodeBench will use subprocess scoring (no lcb-service)" + if [[ ! -d "${LCB_DATASETS_DIR}/test_cases" ]]; then + echo "LiveCodeBench test_cases missing — regenerating dataset cache (required for local scoring)..." + uv run python -c " +from pathlib import Path +from inference_endpoint.dataset_manager.predefined.livecodebench import LiveCodeBench + +LiveCodeBench.generate( + Path('${ENDPOINTS_DIR}/dataset_cache'), + variant='release_v6', + force=True, + save_test_cases=True, +) +" + fi +elif ! curl --output /dev/null --silent --fail "http://127.0.0.1:${LCB_PORT}/info"; then + echo "ERROR: lcb-service is not running on port ${LCB_PORT}." + echo "Either start it: ${SCRIPT_DIR}/start_lcb_service.sh (requires 'docker login dhi.io')" + echo "Or run without the container: export ALLOW_LCB_LOCAL_EVAL=true" + exit 1 +else + echo "lcb-service OK on port ${LCB_PORT}" +fi + +echo "Log directory (host): ${LOG_DIR}" +echo "Accuracy phase drain: unlimited (settings.drain.accuracy_timeout_s default)" +echo "" + +rm -rf "${ENDPOINTS_DIR}/results/sglang_deepseek_v4_pro_accuracy" +echo "Cleared prior results: results/sglang_deepseek_v4_pro_accuracy/" +echo "" + +echo "=== Running accuracy benchmark (from-config) ===" +BENCHMARK_LOG="${LOG_DIR}/accuracy_from_config.log" +CMD=( + uv run inference-endpoint benchmark from-config + -c "${CONFIG}" + --timeout "${TIMEOUT}" + --mode both +) +echo "${CMD[*]}" +set +e +"${CMD[@]}" 2>&1 | tee "${BENCHMARK_LOG}" +bench_rc=${PIPESTATUS[0]} +set -e + +echo "" +echo "Benchmark log: ${BENCHMARK_LOG}" +exit "${bench_rc}" diff --git a/examples/10_DeepSeekV4Pro_Example/run_sglang_benchmark.sh b/examples/10_DeepSeekV4Pro_Example/run_sglang_benchmark.sh new file mode 100755 index 000000000..aef66d392 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/run_sglang_benchmark.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# Run DeepSeek-V4-Pro benchmark against SGLang: +# 1. SGLang server on :30000 (see start_sglang_server.sh) +# 2. lcb-service container on :13835 (for LiveCodeBench scoring) +# 3. inference-endpoint benchmark from-config +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ENDPOINTS_DIR="${ENDPOINTS_DIR:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +# shellcheck source=docker_common.sh +source "${SCRIPT_DIR}/docker_common.sh" +CONFIG="${CONFIG:-${ENDPOINTS_DIR}/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_example.yaml}" +SGLANG_PORT="${SGLANG_PORT:-30000}" +LCB_PORT="${LCB_PORT:-13835}" +TIMEOUT="${TIMEOUT:-60}" +MODE="${MODE:-}" + +cd "${ENDPOINTS_DIR}" + +if [[ -z "${TOKENIZER_MODEL_PATH:-}" ]]; then + export TOKENIZER_MODEL_PATH="${MODEL_PATH:-/data/workloads-inference/models/deepseek-ai/DeepSeek-V4-Pro}" +fi + +echo "=== Pre-flight checks ===" + +SGLANG_BASE_URL="${SGLANG_BASE_URL:-http://127.0.0.1:${SGLANG_PORT}}" +WAIT_FOR_SGLANG_S="${WAIT_FOR_SGLANG_S:-0}" + +if ! wait_openai_compatible_server "${SGLANG_BASE_URL}" "${WAIT_FOR_SGLANG_S}"; then + echo "ERROR: Inference server not reachable at ${SGLANG_BASE_URL} (tried GET /health and GET /v1/models)." >&2 + echo " Start SGLang first (see examples/10_DeepSeekV4Pro_Example/README.md)." >&2 + echo " Smoke test: uv run python -m inference_endpoint.testing.echo_server --host 127.0.0.1 --port ${SGLANG_PORT}" >&2 + echo " If the server is slow to bind: export WAIT_FOR_SGLANG_S=120" >&2 + exit 1 +fi + +if ! curl --output /dev/null --silent --fail "http://127.0.0.1:${LCB_PORT}/info"; then + echo "ERROR: lcb-service is not running on port ${LCB_PORT}." + echo "" + echo "Quick start (after 'docker login dhi.io'):" + echo " ${ENDPOINTS_DIR}/examples/10_DeepSeekV4Pro_Example/start_lcb_service.sh" + exit 1 +fi +echo "lcb-service OK on port ${LCB_PORT}" + +echo "" +echo "=== Running benchmark ===" +CMD=(uv run inference-endpoint benchmark from-config -c "${CONFIG}" --timeout "${TIMEOUT}") +if [[ -n "${MODE}" ]]; then + CMD+=(--mode "${MODE}") +fi +echo "${CMD[*]}" +"${CMD[@]}" diff --git a/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_accuracy.yaml b/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_accuracy.yaml new file mode 100644 index 000000000..7872e6279 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_accuracy.yaml @@ -0,0 +1,80 @@ +name: "deepseek-v4-pro-sglang-accuracy" +version: "1.0" +type: "offline" +# Wall-clock CLI budget (seconds) for the full multi-phase run (AIME + GPQA + LCB). +# Set large: max_new_tokens=320000 with 4 repeats makes generations very long (multi-day total). +timeout: 1209600 + +model_params: + # Must match the model id served by SGLang (see GET /v1/models). + name: "/data/workloads-inference/models/DeepSeek-V4-Pro" + temperature: 1.0 + max_new_tokens: 320000 + top_k: -1 + top_p: 1.0 + streaming: "on" + +datasets: + - name: "perf-stub" + type: "performance" + path: "examples/10_DeepSeekV4Pro_Example/data/perf_stub.jsonl" + samples: 1 + parser: + prompt: "text_input" + - name: "aime25::deepseek_v4" + type: "accuracy" + accuracy_config: + eval_method: "pass_at_1" + ground_truth: "answer" + extractor: "boxed_math_extractor" + num_repeats: 4 + - name: "gpqa::deepseek_v4" + type: "accuracy" + accuracy_config: + eval_method: "pass_at_1" + extractor: "abcd_extractor" + ground_truth: "ground_truth" + num_repeats: 4 + - name: "livecodebench::deepseek_v4" + type: "accuracy" + accuracy_config: + eval_method: "code_bench_scorer" + extractor: "python_code_extractor" + num_repeats: 4 + +settings: + runtime: + min_duration_ms: 100 + max_duration_ms: 0 + scheduler_random_seed: 42 + dataloader_random_seed: 42 + + load_pattern: + type: "max_throughput" + + client: + # Concurrency is bounded by the SGLang KV pool, NOT batch size: with + # max_new_tokens=320000 the dsv4 scheduler ABORTS (SIGABRT) if the + # ~8M-token KV pool saturates to 100% (no graceful preempt). Sequence + # length varies by phase: AIME/GPQA reasoning averages ~81k tokens, but + # LiveCodeBench averages ~167k, so it saturates the pool at far lower + # concurrency (48 crashed during LCB; 96 and 141 crashed on GPQA). + # 32 keeps every phase (incl. LCB) near/below ~75% KV with margin. + num_workers: 32 + # Large worker counts + cold model can exceed the default 60s init budget. + worker_initialization_timeout: 600.0 + + warmup: + enabled: false + + drain: + metrics_drain_timeout_s: 900.0 + metrics_tokenizer_workers: 8 + +endpoint_config: + endpoints: + - "http://127.0.0.1:30000" + api_key: null + api_type: "openai" + +report_dir: "results/sglang_deepseek_v4_pro_accuracy/" diff --git a/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_example.yaml b/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_example.yaml new file mode 100644 index 000000000..f7649e095 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_example.yaml @@ -0,0 +1,61 @@ +name: "deepseek-v4-pro-sglang-benchmark" +version: "1.0" +type: "online" +timeout: 60 + +model_params: + name: "deepseek-ai/DeepSeek-V4-Pro" + temperature: 1.0 + max_new_tokens: 50000 + top_k: -1 + top_p: 1.0 + streaming: "on" + +datasets: + - name: "perf-test" + type: "performance" + path: "examples/04_GPTOSS120B_Example/data/perf_eval_ref.parquet" + parser: + prompt: "text_input" + - name: "livecodebench::deepseek_v4" + type: "accuracy" + accuracy_config: + eval_method: "code_bench_scorer" + extractor: "python_code_extractor" + num_repeats: 3 + - name: "aime25::deepseek_v4" + type: "accuracy" + accuracy_config: + eval_method: "pass_at_1" + ground_truth: "answer" + extractor: "boxed_math_extractor" + num_repeats: 8 + - name: "gpqa::deepseek_v4" + type: "accuracy" + accuracy_config: + eval_method: "pass_at_1" + extractor: "abcd_extractor" + ground_truth: "ground_truth" + num_repeats: 5 + +settings: + runtime: + min_duration_ms: 3000 + max_duration_ms: 60000 + scheduler_random_seed: 42 + dataloader_random_seed: 42 + + load_pattern: + type: "concurrency" + target_concurrency: 512 + + client: + num_workers: 8 + +endpoint_config: + endpoints: + - "http://localhost:30000" + api_key: null + api_type: "openai" + +report_dir: "results/sglang_deepseek_v4_pro_benchmark_full/" diff --git a/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_perf.yaml b/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_perf.yaml new file mode 100644 index 000000000..49adbd5e5 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/sglang_deepseek_v4_pro_perf.yaml @@ -0,0 +1,41 @@ +name: "deepseek-v4-pro-sglang-benchmark" +version: "1.0" +type: "online" +timeout: 60 + +model_params: + name: "deepseek-ai/DeepSeek-V4-Pro" + temperature: 1.0 + max_new_tokens: 50000 + top_k: -1 + top_p: 1.0 + streaming: "on" + +datasets: + - name: "perf-test" + type: "performance" + path: "examples/04_GPTOSS120B_Example/data/perf_eval_ref.parquet" + parser: + prompt: "text_input" + +settings: + runtime: + min_duration_ms: 3000 + max_duration_ms: 60000 + scheduler_random_seed: 42 + dataloader_random_seed: 42 + + load_pattern: + type: "concurrency" + target_concurrency: 512 + + client: + num_workers: 8 + +endpoint_config: + endpoints: + - "http://localhost:30000" + api_key: null + api_type: "openai" + +report_dir: "results/sglang_deepseek_v4_pro_benchmark_perf/" diff --git a/examples/10_DeepSeekV4Pro_Example/start_lcb_service.sh b/examples/10_DeepSeekV4Pro_Example/start_lcb_service.sh new file mode 100755 index 000000000..85975df0d --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/start_lcb_service.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +# Build and run the LiveCodeBench evaluation container (repo-standard workflow). +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ENDPOINTS_DIR="${ENDPOINTS_DIR:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +# shellcheck source=docker_common.sh +source "${SCRIPT_DIR}/docker_common.sh" + +IMAGE="${LCB_IMAGE:-lcb-service:latest}" +CONTAINER_NAME="${LCB_CONTAINER_NAME:-lcb-service}" +PORT="${LCB_PORT:-13835}" + +cd "${ENDPOINTS_DIR}" +ensure_docker_log_dir "lcb" + +if curl --output /dev/null --silent --fail "http://127.0.0.1:${PORT}/info" 2>/dev/null; then + echo "lcb-service already responding on port ${PORT}" + exit 0 +fi + +if ! docker image inspect "${IMAGE}" >/dev/null 2>&1; then + if [[ -z "${HF_TOKEN:-}" ]]; then + echo "ERROR: HF_TOKEN is required to build lcb-service (dataset download at build time)." + exit 1 + fi + echo "Building ${IMAGE} (requires 'docker login dhi.io')..." + docker build \ + -f src/inference_endpoint/evaluation/livecodebench/lcb_serve.dockerfile \ + --secret id=HF_TOKEN,env=HF_TOKEN \ + -t "${IMAGE%%:*}" \ + src/inference_endpoint/evaluation/livecodebench \ + 2>&1 | tee "${LOG_DIR}/lcb_build.log" +fi + +docker rm -f "${CONTAINER_NAME}" 2>/dev/null || true + +echo "Starting ${CONTAINER_NAME} on port ${PORT}..." +echo "LCB logs: ${LOG_DIR} (container /workspace, ${DOCKER_LOG_STORAGE_GB}G storage-opt when supported)" +# shellcheck disable=SC2207 +RUN_STORAGE_OPTS=($(docker_storage_args)) +docker run -d \ + "${RUN_STORAGE_OPTS[@]}" \ + --name "${CONTAINER_NAME}" \ + --rm \ + -p "127.0.0.1:${PORT}:13835" \ + -v "${LOG_DIR}:/workspace:rw" \ + "${IMAGE}" + +for _ in $(seq 1 120); do + if curl --output /dev/null --silent --fail "http://127.0.0.1:${PORT}/info"; then + echo "lcb-service is ready at http://127.0.0.1:${PORT}/info" + exit 0 + fi + sleep 5 +done + +echo "ERROR: lcb-service did not become ready. Check: docker logs ${CONTAINER_NAME}" +exit 1 diff --git a/examples/10_DeepSeekV4Pro_Example/start_sglang_server.sh b/examples/10_DeepSeekV4Pro_Example/start_sglang_server.sh new file mode 100755 index 000000000..79d984f30 --- /dev/null +++ b/examples/10_DeepSeekV4Pro_Example/start_sglang_server.sh @@ -0,0 +1,211 @@ +#!/usr/bin/env bash +# Launch SGLang for DeepSeek-V4-Pro on ROCm (MI35x / DSv4 image). +# Mirrors the FP4-experts env block from SGLang amd/deepseek_v4 run_dsv4.sh. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ENDPOINTS_DIR="${ENDPOINTS_DIR:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +# shellcheck source=docker_common.sh +source "${SCRIPT_DIR}/docker_common.sh" + +MODEL="${MODEL:-${MODEL_PATH:-/data/workloads-inference/models/deepseek-ai/DeepSeek-V4-Pro}}" +MODEL_REPO="${MODEL_REPO:-deepseek-ai/DeepSeek-V4-Pro}" +# HTTP listen port for our scripts/YAML. Do NOT export SGLANG_PORT to the SGLang +# process: upstream uses that env var for internal ZMQ ports (get_open_port), which +# collides with --port and breaks uvicorn bind. +PORT="${HTTP_PORT:-${SGLANG_PORT:-30000}}" +TP="${TP:-8}" +CONC="${CONC:-512}" +MEM_FRACTION_STATIC="${MEM_FRACTION_STATIC:-0.90}" +# Must be >= max_new_tokens (320k) + input so 320k-token generations aren't +# capped by context length. Model supports up to 1M positions. +MAX_MODEL_LEN="${MAX_MODEL_LEN:-327680}" +DP_ATTENTION="${DP_ATTENTION:-false}" +EP_SIZE="${EP_SIZE:-1}" +# DSv4 tokenizer has no chat_template in tokenizer_config.json; SGLang needs --chat-template +# for /v1/chat/completions. Default to the v3.2 tool template shipped in the DSv4 image. +CHAT_TEMPLATE="${CHAT_TEMPLATE:-/sgl-workspace/sglang/examples/chat_template/tool_chat_template_deepseekv32.jinja}" +if [[ ! -f "${CHAT_TEMPLATE}" ]]; then + CHAT_TEMPLATE="" +fi +SGLANG_IMAGE="${SGLANG_IMAGE:-lmsysorg/sglang-rocm:v0.5.14-rocm720-mi35x-20260706}" +RUN_MODE="${RUN_MODE:-host}" # host | docker + +patch_model_config() { + local model_ref="$1" + python3 < deepseek_v3") +else: + print(f"No patch needed: model_type is {config.get('model_type')!r}") +PYEOF +} + +export_sglang_env() { + # Triton/AITER fallback env block for the stock lmsysorg sglang-rocm image, + # which does not ship deep_gemm/tilelang kernels. Routes the DSv4 attention + # backend around deep_gemm (unified_kv_triton flashmla, triton fused compress, + # torch paged MQA logits, aiter indexer instead of tilelang). + export SGLANG_DEFAULT_THINKING=1 + export SGLANG_DSV4_REASONING_EFFORT=max + export SGLANG_OPT_DEEPGEMM_HC_PRENORM=false + export SGLANG_USE_AITER=1 + export SGLANG_USE_ROCM700A=0 + export SGLANG_DP_USE_GATHERV=1 + export SGLANG_OPT_USE_FUSED_COMPRESS=true + export SGLANG_HACK_FLASHMLA_BACKEND=unified_kv_triton + export SGLANG_OPT_FP8_WO_A_GEMM=false + export SGLANG_OPT_USE_JIT_INDEXER_METADATA=false + export SGLANG_OPT_USE_TOPK_V2=false + export SGLANG_OPT_USE_AITER_INDEXER=true + export SGLANG_OPT_USE_TILELANG_INDEXER=false + export SGLANG_OPT_USE_TILELANG_MHC_PRE=false + export SGLANG_OPT_USE_TILELANG_MHC_POST=false + export SGLANG_FP8_PAGED_MQA_LOGITS_TORCH=1 + export SGLANG_OPT_USE_FUSED_COMPRESS_TRITON=true + export AITER_BF16_FP8_MOE_BOUND=0 + export SGLANG_EAGER_INPUT_NO_COPY=true + export SGLANG_OPT_USE_MULTI_STREAM_OVERLAP=false + export SGLANG_ROCM_USE_MULTI_STREAM=false +} + +launch_sglang_server() { + local model_path="$1" + local eval_context_args=() + if [[ "${EVAL_ONLY:-false}" == "true" ]]; then + eval_context_args=(--context-length "${EVAL_MAX_MODEL_LEN:-${MAX_MODEL_LEN}}") + fi + + local parallel_args=(--tensor-parallel-size "${TP}") + if [[ "${DP_ATTENTION}" == "true" ]]; then + export SGLANG_SHARED_EXPERT_TP1=1 + export SGLANG_DP_SHARED_EXPERT_LOCAL=1 + export SGLANG_DP_USE_GATHERV=1 + export SGLANG_DP_USE_REDUCE_SCATTER=1 + export GPU_MAX_HW_QUEUES=5 + parallel_args+=(--dp "${TP}" --enable-dp-attention --enable-prefill-delayer --enable-two-batch-overlap) + fi + if [[ "${EP_SIZE:-1}" -gt 1 ]]; then + parallel_args+=(--ep-size "${EP_SIZE}") + fi + + local chat_template_args=() + if [[ -n "${CHAT_TEMPLATE}" ]]; then + chat_template_args=(--chat-template "${CHAT_TEMPLATE}") + fi + + # SGLANG_PORT must stay unset: see PORT comment above. + env -u SGLANG_PORT python3 -m sglang.launch_server \ + --model-path "${model_path}" \ + --host=0.0.0.0 \ + --port "${PORT}" \ + "${parallel_args[@]}" \ + --trust-remote-code \ + --disable-radix-cache \ + --attention-backend dsv4 \ + --cuda-graph-max-bs "${CONC}" \ + --max-running-requests "${CONC}" \ + --mem-fraction-static "${MEM_FRACTION_STATIC}" \ + --swa-full-tokens-ratio 0.15 \ + --page-size 256 \ + --kv-cache-dtype fp8_e4m3 \ + --context-length "${MAX_MODEL_LEN}" \ + --chunked-prefill-size 8192 \ + --disable-shared-experts-fusion \ + --tool-call-parser deepseekv4 \ + --reasoning-parser deepseek-v4 \ + "${chat_template_args[@]}" \ + --watchdog-timeout 1800 \ + "${eval_context_args[@]}" +} + +if [[ ! -d "${MODEL}" && "${MODEL}" == *"/"* ]]; then + echo "NOTE: local model path ${MODEL} not found; patching HF cache for ${MODEL_REPO}" + patch_model_config "${MODEL_REPO}" + MODEL="${MODEL_REPO}" +elif [[ -f "${MODEL}/config.json" ]]; then + python3 < deepseek_v3") +else: + print(f"No patch needed: model_type is {config.get('model_type')!r}") +PYEOF +else + patch_model_config "${MODEL_REPO}" +fi + +export_sglang_env + +ensure_docker_log_dir "sglang" +SERVER_LOG="${SERVER_LOG:-${LOG_DIR}/server.log}" + +if [[ "${RUN_MODE}" == "docker" && ! -f /.dockerenv ]]; then + if [[ ! -d "${MODEL}" ]]; then + echo "ERROR: RUN_MODE=docker requires a local model directory at MODEL=${MODEL}" + exit 1 + fi + # Writable layer budget for server logs under /workspace (opt-in --storage-opt). + DOCKER_LOG_STORAGE_GB="${DOCKER_LOG_STORAGE_GB:-64}" + export DOCKER_LOG_STORAGE_GB + HF_HOME="${HF_HOME:-${HOME}/.cache/huggingface}" + # shellcheck disable=SC2207 + STORAGE_OPTS=($(docker_storage_args)) + echo "Docker log mount: ${LOG_DIR} -> /workspace" + DOCKER_RUN_ARGS=(--rm -it) + if [[ -n "${DOCKER_NAME:-}" ]]; then + docker rm -f "${DOCKER_NAME}" 2>/dev/null || true + DOCKER_RUN_ARGS=(--name "${DOCKER_NAME}" -d) + fi + docker run "${DOCKER_RUN_ARGS[@]}" \ + "${STORAGE_OPTS[@]}" \ + --device=/dev/kfd \ + --device=/dev/dri \ + --group-add video \ + --ipc=host \ + -p "127.0.0.1:${PORT}:${PORT}" \ + -v "${MODEL}:${MODEL}:ro" \ + -v "${HF_HOME}:/root/.cache/huggingface" \ + -v "${LOG_DIR}:/workspace:rw" \ + -e HF_TOKEN="${HF_TOKEN:-}" \ + -e MODEL="${MODEL}" \ + -e MODEL_REPO="${MODEL_REPO}" \ + -e HTTP_PORT="${PORT}" \ + -e TP="${TP}" \ + -e CONC="${CONC}" \ + -e MEM_FRACTION_STATIC="${MEM_FRACTION_STATIC}" \ + -e MAX_MODEL_LEN="${MAX_MODEL_LEN}" \ + -e DP_ATTENTION="${DP_ATTENTION}" \ + -e EP_SIZE="${EP_SIZE}" \ + -e CHAT_TEMPLATE="${CHAT_TEMPLATE}" \ + -e EVAL_ONLY="${EVAL_ONLY:-false}" \ + -e EVAL_MAX_MODEL_LEN="${EVAL_MAX_MODEL_LEN:-}" \ + -e RUN_MODE=host \ + -e SERVER_LOG=/workspace/server.log \ + -e LOG_DIR=/workspace \ + -v "${SCRIPT_DIR}/start_sglang_server.sh:/start_sglang_server.sh:ro" \ + -v "${SCRIPT_DIR}/docker_common.sh:/docker_common.sh:ro" \ + "${SGLANG_IMAGE}" \ + bash -c 'source /docker_common.sh && bash /start_sglang_server.sh' + exit 0 +fi + +echo "Starting SGLang on port ${PORT} with model ${MODEL} (TP=${TP}, CONC=${CONC})" +echo "Server log: ${SERVER_LOG}" +launch_sglang_server "${MODEL}" 2>&1 | tee -a "${SERVER_LOG}" diff --git a/examples/README.md b/examples/README.md index bb3e43670..caf12d181 100644 --- a/examples/README.md +++ b/examples/README.md @@ -33,6 +33,12 @@ Sample yaml configuration for online benchmarking of `meta-llama/Llama-2-70b-cha Sample yaml configuration to benchmark the multimodal `Qwen/Qwen3-VL-235B-A22B` model on a visual reasoning workload. +### [10_DeepSeekV4Pro_Example](10_DeepSeekV4Pro_Example/) + +End-to-end example for benchmarking `deepseek-ai/DeepSeek-V4-Pro` with SGLang (ROCm), +using the same datasets as the GPT-OSS-120B example (performance parquet + AIME25 + GPQA + +LiveCodeBench). + ## Getting Help - For general usage: See main [README](../README.md) diff --git a/src/inference_endpoint/async_utils/event_publisher.py b/src/inference_endpoint/async_utils/event_publisher.py index bfc2f92ac..13c156884 100644 --- a/src/inference_endpoint/async_utils/event_publisher.py +++ b/src/inference_endpoint/async_utils/event_publisher.py @@ -35,6 +35,7 @@ def __init__( extra_eager: bool = False, isolated_event_loop: bool = False, send_threshold: int = 1000, + max_batch_latency_s: float | None = 1.0, ): """Creates a new EventPublisherService. @@ -46,6 +47,11 @@ def __init__( isolated_event_loop: If True, runs on a separate event loop thread. send_threshold: Minimum number of buffered records before an automatic flush is triggered. See ZmqMessagePublisher. + max_batch_latency_s: Upper bound (seconds) on how long a buffered + record waits before being sent, even below send_threshold. This + keeps low-rate streams (e.g. slow accuracy completions) landing + on disk promptly so a crash loses at most this window; high-rate + streams still flush by count first. See ZmqMessagePublisher. """ if extra_eager: loop = None @@ -60,4 +66,5 @@ def __init__( managed_zmq_context, loop=loop, send_threshold=send_threshold, + max_batch_latency_s=max_batch_latency_s, ) diff --git a/src/inference_endpoint/async_utils/services/event_logger/__main__.py b/src/inference_endpoint/async_utils/services/event_logger/__main__.py index f57d51a0e..801d453e5 100644 --- a/src/inference_endpoint/async_utils/services/event_logger/__main__.py +++ b/src/inference_endpoint/async_utils/services/event_logger/__main__.py @@ -67,6 +67,7 @@ def __init__( *args, writer_classes: tuple[type[RecordWriter], ...] = (JSONLWriter,), flush_interval: int | None = 100, + max_flush_latency_s: float | None = 1.0, shutdown_event: asyncio.Event | None = None, **kwargs, ): @@ -86,7 +87,11 @@ def __init__( self.writers: list[RecordWriter] = [] for writer_class in writer_classes: self.writers.append( - writer_class(log_dir / "events", flush_interval=flush_interval) + writer_class( + log_dir / "events", + flush_interval=flush_interval, + max_flush_latency_s=max_flush_latency_s, + ) ) def _write_record_to_writers(self, record: EventRecord) -> None: diff --git a/src/inference_endpoint/async_utils/services/event_logger/file_writer.py b/src/inference_endpoint/async_utils/services/event_logger/file_writer.py index 56af539d5..a47b01c10 100644 --- a/src/inference_endpoint/async_utils/services/event_logger/file_writer.py +++ b/src/inference_endpoint/async_utils/services/event_logger/file_writer.py @@ -39,9 +39,13 @@ def __init__( file_path: Path, mode: str = "w", flush_interval: int | None = None, + max_flush_latency_s: float | None = None, **kwargs: object, ): - super().__init__(flush_interval=flush_interval) + super().__init__( + flush_interval=flush_interval, + max_flush_latency_s=max_flush_latency_s, + ) self.file_path = Path(file_path).with_suffix(self.extension) self.file_obj = self.file_path.open(mode=mode) # type: ignore[assignment] self.encoder = msgspec.json.Encoder(enc_hook=EventType.encode_hook) diff --git a/src/inference_endpoint/async_utils/services/event_logger/sql_writer.py b/src/inference_endpoint/async_utils/services/event_logger/sql_writer.py index 52c9b0108..63c893675 100644 --- a/src/inference_endpoint/async_utils/services/event_logger/sql_writer.py +++ b/src/inference_endpoint/async_utils/services/event_logger/sql_writer.py @@ -82,6 +82,7 @@ def __init__( path: Path, url: str | None = None, flush_interval: int | None = None, + max_flush_latency_s: float | None = None, **kwargs: object, ): """Initialize the SQL writer. @@ -90,8 +91,14 @@ def __init__( path: Base path for the database. For sqlite default, the file will be path.with_suffix(".db"). url: Optional SQLAlchemy database URL. If None, uses sqlite at path.with_suffix(".db"). flush_interval: If set, flush (commit) after every this many records. + max_flush_latency_s: If set, flush (commit) on the next write once this + many seconds have elapsed since the last flush, even below + flush_interval. Bounds on-disk staleness for low-rate streams. """ - super().__init__(flush_interval=flush_interval) + super().__init__( + flush_interval=flush_interval, + max_flush_latency_s=max_flush_latency_s, + ) if url is None: db_path = Path(path).with_suffix(".db") url = f"sqlite:///{db_path}" diff --git a/src/inference_endpoint/async_utils/services/event_logger/writer.py b/src/inference_endpoint/async_utils/services/event_logger/writer.py index 6dadabe4b..794653652 100644 --- a/src/inference_endpoint/async_utils/services/event_logger/writer.py +++ b/src/inference_endpoint/async_utils/services/event_logger/writer.py @@ -15,6 +15,7 @@ """Writer base class for event records.""" +import time from abc import ABC, abstractmethod from inference_endpoint.core.record import EventRecord @@ -23,29 +24,50 @@ class RecordWriter(ABC): """Abstract base class for writing event records. - Supports an optional flush interval: after every N records written via - write(), the writer is automatically flushed. + Supports two independent auto-flush triggers: a count-based interval + (after every N records) and a time-based latency bound (after this many + seconds since the last flush). The time-based trigger keeps low-rate + streams durable on disk without waiting to accumulate a full count batch. """ - def __init__(self, *args, flush_interval: int | None = None): + def __init__( + self, + *args, + flush_interval: int | None = None, + max_flush_latency_s: float | None = None, + ): """Initialize the writer. Args: flush_interval: If set, flush after every this many records written. - None means no automatic flushing. + None means no count-based flushing. + max_flush_latency_s: If set, flush on the next write() once this many + seconds have elapsed since the last flush, even if flush_interval + has not been reached. Bounds on-disk staleness for low-rate + streams. None means no time-based flushing. """ self._flush_interval = flush_interval + self._max_flush_latency_s = max_flush_latency_s self._n_since_last_flush = 0 + self._last_flush_monotonic = time.monotonic() def write(self, record: EventRecord) -> None: - """Write a record and optionally flush based on flush_interval.""" + """Write a record and optionally flush based on count or elapsed time.""" self._write_record(record) self._n_since_last_flush += 1 + if self._n_since_last_flush == 0: + return if ( self._flush_interval is not None and self._n_since_last_flush >= self._flush_interval ): self.flush() + elif ( + self._max_flush_latency_s is not None + and time.monotonic() - self._last_flush_monotonic + >= self._max_flush_latency_s + ): + self.flush() @abstractmethod def _write_record(self, record: EventRecord) -> None: @@ -60,7 +82,9 @@ def close(self) -> None: def flush(self) -> None: """Flush the writer to ensure all data is written to the underlying storage. - Also resets the flush-interval count so the next flush happens after - another N records (whether flush was triggered by the interval or manually). + Also resets the flush-interval count and the elapsed-time baseline so the + next auto-flush happens after another N records or another latency window + (whether flush was triggered by count, time, or manually). """ self._n_since_last_flush = 0 + self._last_flush_monotonic = time.monotonic() diff --git a/src/inference_endpoint/async_utils/services/metrics_aggregator/token_metrics.py b/src/inference_endpoint/async_utils/services/metrics_aggregator/token_metrics.py index 1f3636de2..241236260 100644 --- a/src/inference_endpoint/async_utils/services/metrics_aggregator/token_metrics.py +++ b/src/inference_endpoint/async_utils/services/metrics_aggregator/token_metrics.py @@ -44,7 +44,7 @@ from inference_endpoint.endpoint_client.cpu_affinity import ( cgroup_clamped_cpus, ) -from transformers import AutoTokenizer +from transformers import AutoTokenizer, PreTrainedTokenizerFast from transformers.utils import logging as transformers_logging # A single rayon pool peaks at ~8 cores for BPE (memory-bound; more threads @@ -239,9 +239,14 @@ def __init__( # -- setup -------------------------------------------------------------- def _load_tokenizer(self) -> None: - tok = AutoTokenizer.from_pretrained( - self._tokenizer_name, trust_remote_code=True - ) + try: + tok = AutoTokenizer.from_pretrained( + self._tokenizer_name, trust_remote_code=True + ) + except Exception: + # Some checkpoints (e.g. DeepSeek-V4) fail AutoTokenizer's config + # detection; fall back to loading the fast tokenizer directly. + tok = PreTrainedTokenizerFast.from_pretrained(self._tokenizer_name) self._tokenizer = tok # Baseline = tokens from a [user, empty-assistant] pair minus the [user] # prefix alone, so the assistant frame is subtracted from message counts. diff --git a/src/inference_endpoint/async_utils/transport/zmq/pubsub.py b/src/inference_endpoint/async_utils/transport/zmq/pubsub.py index 5f0ade89d..a29755741 100644 --- a/src/inference_endpoint/async_utils/transport/zmq/pubsub.py +++ b/src/inference_endpoint/async_utils/transport/zmq/pubsub.py @@ -16,6 +16,7 @@ import asyncio import logging import os +import time from collections import deque from typing import TypeVar from urllib.parse import urlparse @@ -75,6 +76,7 @@ def __init__( loop: asyncio.AbstractEventLoop | None = None, scheme: str = "ipc", send_threshold: int = 1000, + max_batch_latency_s: float | None = None, sndhwm: int = 0, linger: int = -1, ): @@ -96,6 +98,12 @@ def __init__( send_threshold: Minimum buffered records before automatic batch flush. Set to 1 to disable batching (e.g. one snapshot per tick, where batching adds latency). + max_batch_latency_s: If set, flush the buffer on the next send() + once this many seconds have elapsed since the last flush, even + if send_threshold has not been reached. Bounds the on-disk + staleness for low-rate streams (e.g. slow accuracy completions) + without changing high-rate behavior, where the count threshold + fires first. None (default) disables the time-based flush. sndhwm: ZMQ SNDHWM. 0 (default, unlimited) for delivery guarantees. A small value (e.g. 4) makes the writer drop instead of stall when subscribers are slow — appropriate @@ -117,6 +125,8 @@ def __init__( self._fd = self._socket.getsockopt(zmq.FD) self._send_threshold = send_threshold + self._max_batch_latency_s = max_batch_latency_s + self._last_batch_flush_monotonic = time.monotonic() self._batch_buffer: list[bytes] = [] self._last_topic: bytes = b"" self._pending: deque[bytes] = deque() @@ -146,6 +156,12 @@ def send(self, topic: bytes, payload: bytes) -> None: if len(self._batch_buffer) >= self._send_threshold: self._flush_batch() + elif ( + self._max_batch_latency_s is not None + and time.monotonic() - self._last_batch_flush_monotonic + >= self._max_batch_latency_s + ): + self._flush_batch() def flush(self) -> None: """Force-send any buffered records, regardless of threshold. @@ -164,6 +180,7 @@ def _flush_batch(self) -> None: buffer is restored so records are not lost. """ buf = self._batch_buffer + self._last_batch_flush_monotonic = time.monotonic() if len(buf) == 1: # Single record: send with its own topic (no batch overhead). diff --git a/src/inference_endpoint/commands/benchmark/cli.py b/src/inference_endpoint/commands/benchmark/cli.py index 0edb4da79..c95ecc961 100644 --- a/src/inference_endpoint/commands/benchmark/cli.py +++ b/src/inference_endpoint/commands/benchmark/cli.py @@ -31,6 +31,7 @@ ) from inference_endpoint.config.schema import ( BenchmarkConfig, + DatasetType, OfflineBenchmarkConfig, OnlineBenchmarkConfig, TestMode, @@ -177,7 +178,12 @@ def from_config( resolved = resolved.with_updates(timeout=timeout) if report_dir is not None: resolved = resolved.with_updates(report_dir=report_dir) - test_mode = mode or ( - TestMode.BOTH if resolved.type == TestType.SUBMISSION else TestMode.PERF - ) + if mode is not None: + test_mode = mode + elif resolved.type == TestType.SUBMISSION: + test_mode = TestMode.BOTH + elif any(ds.type == DatasetType.ACCURACY for ds in resolved.datasets): + test_mode = TestMode.BOTH + else: + test_mode = TestMode.PERF _run(resolved, [], test_mode, accuracy_only=accuracy_only) diff --git a/src/inference_endpoint/dataset_manager/predefined/aime25/__init__.py b/src/inference_endpoint/dataset_manager/predefined/aime25/__init__.py index 91699ae25..d5332b0aa 100644 --- a/src/inference_endpoint/dataset_manager/predefined/aime25/__init__.py +++ b/src/inference_endpoint/dataset_manager/predefined/aime25/__init__.py @@ -90,13 +90,13 @@ def generate( "opencompass/AIME2025", dataset_name="AIME2025-I", split="test", - cache_dir=datasets_dir / "hf_cache" / "aime25", + cache_dir=datasets_dir / "hf_cache" / "aime25" / "AIME2025-I", ) df_ii = load_from_huggingface( "opencompass/AIME2025", dataset_name="AIME2025-II", split="test", - cache_dir=datasets_dir / "hf_cache" / "aime25", + cache_dir=datasets_dir / "hf_cache" / "aime25" / "AIME2025-II", ) df = pd.concat([df_i, df_ii]) logger.info(f"Loaded {len(df)} samples from AIME25-I and AIME25-II") diff --git a/src/inference_endpoint/dataset_manager/predefined/aime25/presets.py b/src/inference_endpoint/dataset_manager/predefined/aime25/presets.py index eace7fafe..6f5cd4c5a 100644 --- a/src/inference_endpoint/dataset_manager/predefined/aime25/presets.py +++ b/src/inference_endpoint/dataset_manager/predefined/aime25/presets.py @@ -26,3 +26,7 @@ def gptoss() -> list[Transform]: return [UserPromptFormatter(user_prompt_format=_FORMAT)] + + +def deepseek_v4() -> list[Transform]: + return gptoss() diff --git a/src/inference_endpoint/dataset_manager/predefined/gpqa/presets.py b/src/inference_endpoint/dataset_manager/predefined/gpqa/presets.py index 25f4210a5..3b9f8f7d1 100644 --- a/src/inference_endpoint/dataset_manager/predefined/gpqa/presets.py +++ b/src/inference_endpoint/dataset_manager/predefined/gpqa/presets.py @@ -33,3 +33,7 @@ def gptoss() -> list[Transform]: return [UserPromptFormatter(user_prompt_format=_FORMAT)] + + +def deepseek_v4() -> list[Transform]: + return gptoss() diff --git a/src/inference_endpoint/dataset_manager/predefined/livecodebench/presets.py b/src/inference_endpoint/dataset_manager/predefined/livecodebench/presets.py index 092f76bd9..fa7a909ab 100644 --- a/src/inference_endpoint/dataset_manager/predefined/livecodebench/presets.py +++ b/src/inference_endpoint/dataset_manager/predefined/livecodebench/presets.py @@ -35,3 +35,7 @@ def gptoss() -> list[Transform]: return [UserPromptFormatter(user_prompt_format=_FORMAT)] + + +def deepseek_v4() -> list[Transform]: + return gptoss() diff --git a/src/inference_endpoint/evaluation/extractor.py b/src/inference_endpoint/evaluation/extractor.py index fcc116dd5..fe873034a 100644 --- a/src/inference_endpoint/evaluation/extractor.py +++ b/src/inference_endpoint/evaluation/extractor.py @@ -113,8 +113,89 @@ class ABCDExtractor(Extractor, extractor_id="abcd_extractor"): 'choice4' """ + CHOICE_MAP = { + "A": "choice1", + "B": "choice2", + "C": "choice3", + "D": "choice4", + } + + # Each pattern below captures the answer letter as group 1. + FINAL_ANSWER_PATTERNS = [ + # JSON-ish final responses. + # Examples: {"answer": "A"}, 'answer': 'C' + re.compile(r"""(?is)["']answer["']\s*:\s*["']?\s*([ABCD])\b"""), + # Explicit final-answer statements. + # Examples: "Final answer: B", "answer is (D)", "Answer = **C**" + re.compile( + r"""(?ix) + \b(?:final\s+answer|answer)\b + \s*(?:is|:|=)?\s* + (?:\\boxed\{\s*)? + (?:\*{1,2}|_{1,2})? + \(?\s*([ABCD])\b + """ + ), + # Explicit option/choice statements near the end of the response. + # Examples: "option C", "Choice: (A)", "the correct choice is **B**" + re.compile( + r"""(?ix) + \b(?:option|choice)\b + \s*(?:is|:|=)?\s* + (?:\*{1,2}|_{1,2})? + \(?\s*([ABCD])\b + """ + ), + # "the correct answer/choice/option is C", "correct answer: (B)" + re.compile( + r"""(?ix) + \bcorrect\s+(?:answer|choice|option)\b + \s*(?:is|:|=)?\s* + (?:\\boxed\{\s*)? + (?:\*{1,2}|_{1,2})? + \(?\s*([ABCD])\b + """ + ), + # "C is the correct answer", "**B** the correct choice", "(A) is correct" + re.compile( + r"""(?ix) + (?:\*{1,2}|_{1,2})? + \(?\s*([ABCD])\s*\)? + (?:\*{1,2}|_{1,2})? + \s+(?:is\s+)?(?:the\s+)correct + (?:\s+(?:answer|choice|option))? + \b + """ + ), + # Boxed answers. + # Examples: "\\boxed{D}", "\\boxed{\\text{A}}", "\\boxed{\\textbf{C}}" + re.compile(r"""(?is)\\boxed\{\s*(?:\\(?:text|textbf)\{\s*)?([ABCD])\b"""), + # A final standalone line such as "C", "**D**", or "(B)". + # Examples: final line "A", final line "**D**", final line "(B)" + re.compile( + r"""(?im)^\s* + (?:\*{1,2}|_{1,2})? + \(?\s*([ABCD])\s*\)? + (?:\*{1,2}|_{1,2})? + \s*[\.\)]?\s*$ + """ + ), + # A final answer line with the option text included, e.g. "(D) foo". + # Examples: "(D) all of the above", "**(B)** pressure increases" + re.compile( + r"""(?im)^\s* + (?:\*{1,2}|_{1,2})? + \(\s*([ABCD])\s*\) + (?:\*{1,2}|_{1,2})? + \s+\S + """ + ), + ] + + # Each fallback pattern below also captures the answer letter as group 1. PATTERNS = [ # 0) "**Answer:** A" or "*Answers* – B", i.e. markdown-wrapped "Answer(s)" with an unwrapped letter. + # Examples: "**Answer:** A", "*Answers* - B", "__Answer__ C" re.compile( r"""(?ix) # case-insensitive, ignore-space (?:\*{1,2}|_{1,2}) # leading *…* or _…_ @@ -127,6 +208,7 @@ class ABCDExtractor(Extractor, extractor_id="abcd_extractor"): re.X, ), # 0.1) Answer with optional markdown and colons + # Examples: "Answer: **D**", "**Answer:** C", "answer B" re.compile( r"""(?ix) # ignore case, allow verbose mode ^\s* # optional leading whitespace @@ -142,31 +224,41 @@ class ABCDExtractor(Extractor, extractor_id="abcd_extractor"): re.MULTILINE, ), # 1) Answer: (C) or Answers: (B) + # Examples: "Answer: (C)", "Answers - (B)" re.compile(r"(?ix)\bAnswer[s]?\b\s*[:\-–]?\s*\(\s*([ABCD])\s*\)"), # 2) Answer: C or Answers – D + # Examples: "Answer: C", "Answers - D" re.compile(r"(?ix)\bAnswer[s]?\b\s*[:\-–]?\s*([ABCD])\b"), # 3) Option B or Choice: C + # Examples: "Option B", "Choice: C" re.compile(r"(?ix)\b(?:Option|Choice)\b\s*[:\-–]?\s*([ABCD])\b"), # 7) LaTeX \boxed{...A...}, catches both \boxed{A} and # \boxed{\text{A } 2.08\times10^{-6}\,\mathrm{m}} etc. + # Examples: "\\boxed{A}", "\\boxed{the answer is C}" re.compile(r"(?x)\\boxed\{[^}]*?([ABCD])[^}]*\}", re.MULTILINE), # 7.5) LaTeX \boxed{\textbf{...C...}} + # Examples: "\\boxed{\\textbf{C}}", "\\boxed{\\textbf{choice D}}" re.compile( r"(?x)\\boxed\{[^}]*?\\textbf\{[^}]*?([ABCD])[^}]*\}[^}]*\}", re.MULTILINE ), # 7.51) LaTeX \boxed{\text{...C...}} + # Examples: "\\boxed{\\text{B}}", "\\boxed{\\text{Answer: A}}" re.compile( r"(?x)\\boxed\{[^}]*?\\text\{[^}]*?([ABCD])[^}]*\}[^}]*\}", re.MULTILINE ), # 4) bare singletons: (A) [B] + # Examples: "(A)", "[B]" re.compile(r"(?x)(? str | None: + if not text or not isinstance(text, str): + return default if default is not None else "" + + # Reasoning models often produce a long rationale followed by a concise + # final answer. Prefer explicit answer forms near the response tail before + # broad option-list fallbacks such as "(A)" or "(B)". + # + # Rightmost-match rule: scan only the last 6000 characters, collect every + # final-answer regex match as (start_offset, end_offset, letter), then + # select max(..., key=(start_offset, end_offset)). This guarantees that + # the latest regex match inside the scanned tail wins, e.g. an earlier + # "Choice: A" is overridden by a later "Final answer: D". This is still a + # regex-position guarantee, not a semantic guarantee: if the true answer + # is outside the tail, does not match these regexes, or is followed by a + # later false-positive answer-like string, the extractor can still choose + # the wrong letter. + tail = text.strip()[-6000:] + final_matches: list[tuple[int, int, str]] = [] + for pat in cls.FINAL_ANSWER_PATTERNS: + for m in pat.finditer(tail): + letter = m.group(1).upper() + if letter in cls.CHOICE_MAP: + final_matches.append((m.start(), m.end(), letter)) + if final_matches: + _, _, letter = max(final_matches, key=lambda item: (item[0], item[1])) + return cls.CHOICE_MAP[letter] + matches = [] for prio, pat in enumerate(cls.PATTERNS): - m = pat.search(text) - if m: - letter = m.group(1).upper() + sm = pat.search(text) + if sm: + letter = sm.group(1).upper() if letter in "ABCD": - matches.append((prio, m, letter)) + matches.append((prio, sm, letter)) # Sort by priority (lower is better) and then by match length (shorter is better) matches.sort(key=lambda triple: (triple[0], len(triple[1].group(0)))) - choice_map = { - "A": "choice1", - "B": "choice2", - "C": "choice3", - "D": "choice4", - } - # Return the best match for _, _, letter in matches: - return choice_map[letter] + return cls.CHOICE_MAP[letter] # Final fallback from OpenAI: take first character after stripping markdown # This is a last resort if no patterns matched stripped = text.removeprefix("**") if stripped and stripped[0].upper() in "ABCD": abcd_choice = stripped[0].upper() - return choice_map[abcd_choice] + return cls.CHOICE_MAP[abcd_choice] return default if default is not None else "" @@ -279,6 +392,44 @@ class PythonCodeExtractor(Extractor, extractor_id="python_code_extractor"): '# FAILED' """ + @staticmethod + def _parses(code: str) -> bool: + try: + ast.parse(code) + return True + except (SyntaxError, ValueError): + return False + + @classmethod + def _select_solution_block(cls, blocks: list[str]) -> str: + """Choose the block most likely to be the complete final solution. + + Preference order (last-wins within each tier, since the final rewrite is + usually latest): + 1. Parseable blocks that define a function/class OR read input and + write output (a self-contained program), length >= 200. + 2. Any parseable block of length >= 200. + 3. Any parseable block. + 4. The last raw block (previous behaviour) as a last resort. + """ + parseable = [b for b in blocks if cls._parses(b)] + + def _complete(b: str) -> bool: + has_def = "def " in b or "class " in b + has_in = "input(" in b or "sys.stdin" in b + has_out = "print(" in b or "sys.stdout" in b + return has_def or (has_in and has_out) + + for b in reversed(parseable): + if len(b) >= 200 and _complete(b): + return b + substantial = [b for b in parseable if len(b) >= 200] + if substantial: + return substantial[-1] + if parseable: + return parseable[-1] + return blocks[-1] + @classmethod def extract(cls, text: str, default: str | None = None) -> str | None: if not text or not isinstance(text, str): @@ -288,19 +439,53 @@ def extract(cls, text: str, default: str | None = None) -> str | None: if not text: return default - # Try ```python blocks first (most specific) - python_matches = list(re.finditer(r"```python(.*?)```", text, re.DOTALL)) - if python_matches: - return python_matches[-1].group(1).strip() - - # Fall back to plain ``` blocks - plain_matches = list(re.finditer(r"```(.*?)```", text, re.DOTALL)) - if plain_matches: - # Get the last match - code = plain_matches[-1].group(1).strip() - # Remove language tag if present (e.g., ```python\n or ```py\n) - code = re.sub(r"^(?:python|py)\s*\n", "", code, flags=re.IGNORECASE) - return code + # Collapse repeated/empty language openers (e.g. a stray + # "```python\n\n```python\n") that otherwise make the middle + # fence look like a closing fence and yield an empty block. + text = re.sub( + r"(?:```[ \t]*(?:python|py)[ \t]*\r?\n\s*)+?(```[ \t]*(?:python|py)[ \t]*\r?\n)", + r"\1", + text, + flags=re.IGNORECASE, + ) + + # Try ```python blocks first (most specific). Reasoning models (e.g. + # DeepSeek-V4) stream their whole rationale into the answer and emit MANY + # ```python blocks — full-solution rewrites interleaved with tiny debug + # snippets ("idx = i + 1") and non-parseable pseudo-blocks. Blindly + # taking the last block often grabs a trailing fragment that references + # names defined in an earlier block (→ NameError) or garbage. Instead + # pick the last block that looks like a COMPLETE, self-contained solution. + python_blocks = [ + m.group(1).strip() + for m in re.finditer( + r"```[ \t]*python[ \t]*\r?\n(.*?)```", text, re.DOTALL | re.IGNORECASE + ) + ] + python_blocks = [c for c in python_blocks if c] + if python_blocks: + return cls._select_solution_block(python_blocks) + + # Fall back to plain ``` blocks (strip any lang tag). + plain_blocks = [] + for m in re.finditer(r"```(.*?)```", text, re.DOTALL): + code = re.sub( + r"^(?:python|py)\s*\r?\n", "", m.group(1).strip(), flags=re.IGNORECASE + ).strip() + if code: + plain_blocks.append(code) + if plain_blocks: + return cls._select_solution_block(plain_blocks) + + # Last resort: an unclosed final fence (truncated response). Take from + # the last opening fence to the end of text. + tail_m = re.search( + r"```[ \t]*(?:python|py)?[ \t]*\r?\n(.*)$", text, re.DOTALL | re.IGNORECASE + ) + if tail_m: + code = tail_m.group(1).strip().rstrip("`").strip() + if code: + return code return default diff --git a/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py b/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py index 71fc6bd2f..88a24b6f2 100644 --- a/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py +++ b/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py @@ -531,11 +531,29 @@ def evaluate_dataframe( datasets_dir=args.datasets_dir, ) + from collections import defaultdict as _dd + + df["extracted_code"] = df["extracted_code"].fillna("") + codes_dict = _dd(list) + for _, row in df.iterrows(): + codes_dict[row["question_id"]].append(row["extracted_code"]) + with tqdm(total=len(df)) as pbar: - results = lcb_serve.evaluate_dataframe( - df=df, + res = lcb_serve.evaluate( + codes_dict=codes_dict, timeout_sec=args.timeout, on_problem_complete=lambda x: pbar.update(len(x)), ) + total = sum(len(v) for v in res.values()) + passed = sum(sum(v) for v in res.values()) + nq = len(res) + results = { + "total_samples": total, + "passed_samples": passed, + "pass_at_1": (passed / total) if total else 0.0, + "n_problems": nq, + "pass_at_k_any": (sum(1 for v in res.values() if any(v)) / nq) if nq else 0.0, + "pass_all_k": (sum(1 for v in res.values() if all(v)) / nq) if nq else 0.0, + } print(json.dumps(results)) diff --git a/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py b/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py index 16798f6ff..a222edc18 100644 --- a/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py +++ b/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py @@ -29,7 +29,7 @@ import time from decimal import Decimal from enum import Enum -from io import StringIO +from io import BytesIO, StringIO # from pyext import RuntimeModule from types import ModuleType @@ -119,37 +119,58 @@ def __exit__(self, *args): sys.stdout = self._stdout -# Custom mock for sys.stdin that supports buffer attribute -class MockStdinWithBuffer: +class MockBuffer: + """Bytes-mode stand-in for ``sys.stdin.buffer`` backed by BytesIO so it + supports read/readline/readlines AND iteration (``for line in + sys.stdin.buffer``).""" + def __init__(self, inputs: str): - self.inputs = inputs - self._stringio = StringIO(inputs) - self.buffer = MockBuffer(inputs) + self._io = BytesIO(inputs.encode("utf-8")) def read(self, *args): - return self.inputs + return self._io.read(*args) def readline(self, *args): - return self._stringio.readline(*args) + return self._io.readline(*args) def readlines(self, *args): - return self.inputs.split("\n") + return self._io.readlines(*args) + + def __iter__(self): + return iter(self._io) + + def __next__(self): + return next(self._io) def __getattr__(self, name): - # Delegate other attributes to StringIO - return getattr(self._stringio, name) + return getattr(self._io, name) -class MockBuffer: +# Custom mock for sys.stdin that supports the buffer attribute AND iteration. +class MockStdinWithBuffer: def __init__(self, inputs: str): - self.inputs = inputs.encode("utf-8") # Convert to bytes + self.inputs = inputs + self._stringio = StringIO(inputs) + self.buffer = MockBuffer(inputs) def read(self, *args): - # Return as byte strings that can be split - return self.inputs + return self._stringio.read(*args) def readline(self, *args): - return self.inputs.split(b"\n")[0] + b"\n" + return self._stringio.readline(*args) + + def readlines(self, *args): + return self._stringio.readlines(*args) + + def __iter__(self): + return iter(self._stringio) + + def __next__(self): + return next(self._stringio) + + def __getattr__(self, name): + # Delegate other attributes (e.g. seek, tell) to the backing StringIO + return getattr(self._stringio, name) def clean_if_name(code: str) -> str: @@ -364,21 +385,36 @@ def grade_stdio( ## runtime doesn't interact well with __name__ == '__main__' code = clean_if_name(code) - ## we wrap the given code inside another function - code = make_function(code) - - compiled_sol = compile_code(code, timeout) - if compiled_sol is None: - return - - method = get_function(compiled_sol, "wrapped_function") - - if method is None: - return + # Execute the solution at MODULE scope (re-exec per test input) instead of + # wrapping it in a function. Wrapping in a function turns module-level names + # into function locals, which breaks `global` declarations and comprehensions + # / nested functions that reference module-level state (the source of the + # spurious NameError / UnboundLocalError "Runtime Error" failures). Re-exec + # per input mirrors how a real judge runs the program once per test file. + full_code = import_string + "\n" + code + try: + compiled = compile(full_code, "", "exec") + except Exception as e: + return [-4], { + "error": repr(e), + "error_code": -4, + "error_message": "Compilation Error", + } all_results = [] total_execution_time = 0.0 for gt_inp, gt_out in zip(all_inputs, all_outputs, strict=False): + if isinstance(gt_inp, list): + gt_inp = "\n".join(gt_inp) + + stdin_mock = MockStdinWithBuffer(gt_inp) + + def _mock_input(*_args, _stdin: MockStdinWithBuffer = stdin_mock) -> str: + line = _stdin._stringio.readline() + if line == "": + raise EOFError("EOF when reading a line") + return line.rstrip("\n") + signal.alarm(timeout) faulthandler.enable() @@ -386,7 +422,16 @@ def grade_stdio( with Capturing() as captured_output: try: start = time.time() - call_method(method, gt_inp) + with ( + patch("sys.stdin", stdin_mock), + patch("builtins.input", _mock_input), + patch("builtins.open", mock_open(read_data=gt_inp)), + ): + try: + exec(compiled, {"__name__": "__lcb_main__"}) + except SystemExit: + # Solutions may call exit()/sys.exit() after printing. + pass total_execution_time += time.time() - start # reset the alarm signal.alarm(0) diff --git a/src/inference_endpoint/evaluation/scoring.py b/src/inference_endpoint/evaluation/scoring.py index 8c4a49998..abec13102 100644 --- a/src/inference_endpoint/evaluation/scoring.py +++ b/src/inference_endpoint/evaluation/scoring.py @@ -58,6 +58,16 @@ logger = logging.getLogger(__name__) +def _join_output_parts(value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + if isinstance(value, tuple | list): + return "".join(str(part) for part in value) + return str(value) + + class Scorer(ABC): """Scorers will read in a dataset and outputs from a log and compute an accuracy score. An optional extractor can be provided to post-process the output to extract values that @@ -161,10 +171,27 @@ def get_outputs(self): continue record = decoder.decode(stripped) if record.event_type == SampleEventType.COMPLETE: - output_text = str(record.data) if record.data is not None else "" + if isinstance(record.data, TextModelOutput): + output_text = _join_output_parts(record.data.output) + # Reasoning-model fallback: some responses emit the final + # answer only inside the reasoning/thinking stream and leave + # the main content empty. When the main output is empty, + # fall back to the reasoning trace so the extractor can still + # recover the answer. + if ( + not output_text.strip() + and record.data.reasoning is not None + ): + output_text = _join_output_parts(record.data.reasoning) + else: + output_text = ( + str(record.data) if record.data is not None else "" + ) outputs.append( {"sample_uuid": record.sample_uuid, "output": output_text} ) + if not outputs: + return pd.DataFrame(columns=["sample_uuid", "output"]) return pd.DataFrame(outputs) def match_sample_index(self, row: pd.Series) -> pd.Series: @@ -184,10 +211,19 @@ def score(self) -> tuple[float | None, int]: Returns None as the score if evaluation fails. """ df = self.get_outputs() + if df.empty: + raise FileNotFoundError( + f"No COMPLETE events in {self.report_dir / 'events.jsonl'} " + f"for dataset {self.dataset_name}" + ) # Outputs are for all samples, not just the target dataset valid_uuids = self.sample_index_map.keys() df = df[df["sample_uuid"].isin(valid_uuids)] + if df.empty: + raise KeyError( + f"No COMPLETE events matched sample_idx_map for {self.dataset_name}" + ) # Match to sample index from dataset df = df.apply(self.match_sample_index, axis=1) @@ -1005,12 +1041,15 @@ def _evaluate_via_subprocess(self, df: pd.DataFrame) -> float | None: cmd = [ sys.executable, "-m", - "inference_endpoint.dataset_manager.predefined.livecodebench.lcb_serve", + "inference_endpoint.evaluation.livecodebench.lcb_serve", str(parquet_path), "--version-tag", self.lcb_version, "--datasets-dir", - f"datasets/livecodebench/{self.lcb_version}", + os.environ.get( + "LCB_DATASETS_DIR", + f"datasets/livecodebench/{self.lcb_version}", + ), "--timeout", str(self.timeout), ] diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index 0d991ab1f..4270137e0 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -355,6 +355,16 @@ def __init__( self._strategy_task: asyncio.Task | None = None self._drain_event = asyncio.Event() + def cancel_current_strategy(self) -> None: + """Cancel the running load strategy without ending the session. + + Used when a performance phase hits its duration limit but later accuracy + phases should still run. + """ + self._drain_event.set() + if self._strategy_task and not self._strategy_task.done(): + self._strategy_task.cancel() + def stop(self) -> None: """Signal early termination. Safe to call from signal handler. @@ -363,9 +373,7 @@ def stop(self) -> None: event to unblock _drain_inflight if it's waiting for responses. """ self._stop_requested = True - self._drain_event.set() - if self._strategy_task and not self._strategy_task.done(): - self._strategy_task.cancel() + self.cancel_current_strategy() def stop_current_phase(self) -> None: """End the in-progress phase without aborting the session. @@ -391,6 +399,8 @@ async def run( self, phases: list[PhaseConfig], on_phase_start: Callable[[PhaseConfig], None] | None = None, + on_phase_complete: Callable[[PhaseConfig, PhaseResult | None], None] + | None = None, ) -> SessionResult: """Run all benchmark phases sequentially. @@ -411,6 +421,8 @@ async def run( result = await self._run_phase(phase) if result is not None: phase_results.append(result) + if on_phase_complete is not None: + on_phase_complete(phase, result) finally: self._done = True if self._recv_task and not self._recv_task.done(): @@ -515,7 +527,12 @@ async def _drain_inflight( or self._current_phase_stopped ): return - logger.info("Draining %d in-flight responses...", phase_issuer.inflight) + timeout_label = "unlimited" if timeout is None else f"{timeout:.0f} s" + logger.info( + "Draining %d in-flight responses (timeout=%s)...", + phase_issuer.inflight, + timeout_label, + ) self._drain_event.clear() # Re-check after clear: a completion (or a per-phase cap firing) may have # set the event between the initial inflight check and clear(), which diff --git a/src/inference_endpoint/testing/echo_server.py b/src/inference_endpoint/testing/echo_server.py index 9cb7847b0..a188b8824 100644 --- a/src/inference_endpoint/testing/echo_server.py +++ b/src/inference_endpoint/testing/echo_server.py @@ -36,6 +36,11 @@ RequestHandler = Callable[[web.Request], web.Response | Awaitable[web.Response]] +async def _echo_server_health(_request: web.Request) -> web.Response: + """Liveness probe compatible with SGLang / vLLM-style ``GET /health`` checks.""" + return web.json_response({"status": "ok"}) + + class HTTPServer: @property @abstractmethod @@ -343,6 +348,7 @@ def _register_routes(self, app: "web.Application") -> None: Subclasses can override to swap out the OpenAI-shaped routes for a different wire contract while reusing the lifecycle plumbing. """ + app.router.add_get("/health", _echo_server_health) app.router.add_post( "/v1/chat/completions", self._handle_echo_chat_completions_request ) diff --git a/tests/unit/commands/test_benchmark.py b/tests/unit/commands/test_benchmark.py index 491dc72b6..a41d5a45f 100644 --- a/tests/unit/commands/test_benchmark.py +++ b/tests/unit/commands/test_benchmark.py @@ -392,8 +392,31 @@ def test_from_config_submission_defaults_to_both(self, mock_run, tmp_path): _, called_mode = mock_run.call_args[0] assert called_mode == TestMode.BOTH + @pytest.mark.unit + @patch("inference_endpoint.commands.benchmark.cli.run_benchmark") + def test_from_config_offline_accuracy_defaults_to_both(self, mock_run, tmp_path): + yaml_content = """ +type: "offline" +model_params: + name: "test-model" +endpoint_config: + endpoints: ["http://test:8000"] +datasets: + - path: "tests/assets/datasets/dummy_1k.jsonl" + type: "performance" + - name: "gpqa::test" + type: "accuracy" + accuracy_config: + eval_method: "pass_at_1" + ground_truth: "answer" + num_repeats: 1 +""" + config_file = tmp_path / "acc.yaml" + config_file.write_text(yaml_content) + from_config(config=config_file) + _, called_mode = mock_run.call_args[0] + assert called_mode == TestMode.BOTH -class TestBenchmarkValidation: """Test BenchmarkConfig validation paths.""" @pytest.mark.unit diff --git a/tests/unit/test_http_mock_fixtures.py b/tests/unit/test_http_mock_fixtures.py index 786f28ddf..0d60a53bb 100644 --- a/tests/unit/test_http_mock_fixtures.py +++ b/tests/unit/test_http_mock_fixtures.py @@ -57,6 +57,14 @@ async def test_http_echo_server_post_request(self, mock_http_echo_server): assert response_data["request"]["endpoint"] == "/echo" assert response_data["request"]["json_payload"] == payload + @pytest.mark.unit + @pytest.mark.asyncio + async def test_mock_http_echo_server_health(self, mock_http_echo_server): + async with aiohttp.ClientSession() as session: + async with session.get(f"{mock_http_echo_server.url}/health") as response: + assert response.status == 200 + assert await response.json() == {"status": "ok"} + @pytest.mark.asyncio async def test_mock_http_echo_server_chat_completions(self, mock_http_echo_server): """Test basic echo functionality of the real HTTP server."""