diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index bf1f5a5..f917c8c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -35,7 +35,7 @@ jobs: - name: Install dependencies run: | pip install torch --index-url https://download.pytorch.org/whl/cpu - pip install -e ".[dev,webrtc]" + pip install -e ".[dev]" - name: Run unit tests run: | @@ -75,7 +75,7 @@ jobs: - name: Install dependencies run: | pip install torch --index-url https://download.pytorch.org/whl/cpu - pip install -e ".[dev,webrtc]" + pip install -e ".[dev]" - name: Run server tests run: | diff --git a/CLAUDE.md b/CLAUDE.md index 3700178..2fee9f9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -16,6 +16,7 @@ pre-commit run --all-files # Linting checks pytest tests/ # Run tests bash scripts/run_ci_tests.sh # Full CI suite telefuser serve /path/to/pipeline.py --port 8000 # Start API server +telefuser stream-serve /path/to/pipeline.py --port 8088 # LiveKit-backed streaming ``` ## Troubleshooting @@ -65,7 +66,7 @@ telefuser/ ├── orchestrator/ # Request orchestration and actor-based streaming scheduler ├── worker/ # Distributed worker management ├── entrypoints/ # CLI entry points -├── service/ # FastAPI service +├── service/ # FastAPI request-response and LiveKit-backed streaming └── client/ # Python SDK ``` @@ -94,6 +95,14 @@ telefuser/ and memory without retaining duration-sized tensor lists. - Interpret chunk period as output cadence: real-time operation requires p95 to stay below the media duration represented by one chunk, with margin for transport and encoding. +- Treat `telefuser stream-serve` as the only streaming entrypoint. It must accept both + `ServerPushService` and `BidirectionalService` while preserving native frame/audio payloads. +- LiveKit workers own one room and one pipeline session. Browser reconnects must not mutate + pipeline caches, and controller messages must remain on the reliable `tf.control` topic; + status uses reliable `tf.status`, while bounded telemetry uses unreliable `tf.metrics`. +- Keep transport metrics distinct from model facts: output cadence measures adjacent emitted + chunks, pipeline residence measures actor admission to output, and applied-control latency + is bound to the control snapshot consumed by that output chunk. ### LingBot-Video Single-Process Runtime diff --git a/README.md b/README.md index 596f267..c6d3e35 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,8 @@ TeleFuser is a high-performance runtime for world model inference and multimodal ## News 📰 +- ✨ **2026-07-27**: Unified streaming on LiveKit with room sessions, worker admission, reconnect-friendly browser + transport, and support for both server-push and bidirectional pipeline contracts. - ✨ **2026-07-22**: **NEW** Added [**LingBot-Video**](examples/lingbot_video/README.md) support for Dense and MoE T2I/T2V/TI2V generation, native four-GPU CFG/SP execution, and in-memory MoE refinement. - ✨ **2026-07-15**: Added [**LingBot-World v2**](https://github.com/Robbyant/lingbot-world-v2) support for offline generation, interactive WebRTC streaming, and multi-GPU inference. @@ -39,10 +41,12 @@ The project treats a world model as more than a function that returns a single c - **World-model-oriented runtime**: Support for continuous video generation, interactive sessions, and bidirectional control loops. - **ADF (AI Dev First)**: Repository layers, pipeline contracts, examples, and docs are structured for coding agents to discover capabilities, follow project conventions, and extend pipelines efficiently. - **Streaming pipeline scheduler**: Actor-owned stateful stages, bounded artifact edges, per-session ordering, backpressure, lifecycle cleanup, and explicit resource groups. -- **Streaming transport**: WebRTC-based streaming with media tracks plus DataChannel control for real-time inference. +- **Streaming transport**: LiveKit-backed WebRTC for server-push media and resilient bidirectional sessions, with + room lifecycle, reconnect handling, participant roles, and reliable controls. - **Scalable GPU runtime**: Multi-GPU execution with tensor parallelism, sequence parallelism, optional Ray workers, and distributed service replicas. - **Inference optimization stack**: Triton kernels, optimized attention backends, quantization, offload, feature caching, and CacheSeek latent cache integration. -- **Unified serving**: Local Python API, `telefuser serve` for task APIs, and `telefuser stream-serve` for continuous streaming services. +- **Unified serving**: Local Python API, `telefuser serve` for task APIs, and `telefuser stream-serve` for LiveKit + rooms and media. ## Quick Start @@ -62,7 +66,8 @@ TeleFuser does not require `tf-kernel` to run. No prebuilt tf-kernel package is extension is built from source with the Makefile under `tf-kernel/`. See the [tf-kernel README](tf-kernel/README.md) and [installation and usage guide](docs/en/tf_kernel.md) for build, verification, and compatibility details. -WebRTC streaming support is included in the default installation through `aiortc`. +The base installation includes the LiveKit Python SDKs used by `telefuser stream-serve`. A LiveKit Cloud project or +self-hosted LiveKit Server is operated separately. ### 1. Batch Video Inference @@ -84,65 +89,76 @@ video = pipe( ) ``` -### 2. Real-Time World Model Demo +### 2. Real-Time World Model WebRTC Demo -TeleFuser includes a bidirectional WebRTC demo for `LingBot-World v2`. -LingBot-World v2 uses camera control and its v2 PPL defaults; its streaming example caps a session at two minutes. +TeleFuser streams `LingBot-World v2` through LiveKit. LingBot-World v2 uses camera control and its v2 PPL defaults; +its streaming example caps a session at two minutes. LingBot streaming uses the actor-based scheduler for both offline and service execution. Encode, DiT, and decode may overlap even on the same GPU; move stages only when memory placement requires it. See the [streaming scheduler guide](docs/en/stream_scheduler.md). -For a laptop browser connected through VS Code Remote SSH, coturn is the only additional system package required; -no extra Python package is needed. On Debian or Ubuntu, install it with: +The checked-in browser page forces a TCP TURN relay so the same setup works through VS Code Remote SSH. The complete +local development stack therefore has four processes: coturn, LiveKit Server, TeleFuser, and the browser page. +Install the LiveKit Server and your platform's `coturn` package once: ```bash +# Debian/Ubuntu; use the equivalent coturn package on other platforms. sudo apt-get update sudo apt-get install -y coturn + +# Install LiveKit Server once. +curl -sSL https://get.livekit.io | bash +``` + +Then run each command below in a separate terminal from the repository root. + +Terminal 1 — start the development-only TCP TURN relay: + +```bash +turnserver -n -m 1 \ + --listening-ip=127.0.0.1 --relay-ip=127.0.0.1 \ + --listening-port=3478 --min-port=49160 --max-port=49200 \ + --user=livekit-demo:livekit-demo-password \ + --realm=livekit.local --fingerprint --lt-cred-mech \ + --no-tls --no-dtls --no-cli --allow-loopback-peers +``` + +Terminal 2 — start LiveKit with its development credentials (`devkey` / `secret`): + +```bash +livekit-server --dev ``` -The package provides both `turnserver` and the `turnutils_uclient` verification tool. Skip this step when both -commands already exist, or when the browser and GPU service run on the same physical machine. +Terminal 3 — load the four-GPU LingBot-World v2 service: ```bash TF_MODEL_ZOO_PATH=/path/to/model_zoo \ CUDA_VISIBLE_DEVICES=0,1,2,3 \ -TELEFUSER_TURN_SERVER='turn:127.0.0.1:3478?transport=tcp' \ -TELEFUSER_TURN_USERNAME=telefuser \ -TELEFUSER_TURN_CREDENTIAL=telefuser-turn \ telefuser stream-serve examples/lingbot/lingbot_world_v2_image_to_video_h100.py \ - --gpu-num 4 -p 8088 --host 0.0.0.0 --skip-validation - -python examples/stream_server/webrtc_bidirectional_demo.py \ - --server-url http://127.0.0.1:8088 \ - --port 8091 \ - --image-path examples/data/lingbot_world_fast/image.jpg \ - --turn-url 'turn:localhost:3478?transport=tcp' \ - --turn-username telefuser --turn-credential telefuser-turn \ - --force-turn-relay --ice-gather-timeout-ms 30000 --no-open + --livekit-url ws://127.0.0.1:7880 \ + --livekit-api-key devkey --livekit-api-secret secret \ + --num-workers 1 --worker-gpu-map 0,1,2,3 \ + --port 8088 --skip-validation ``` -This starts a continuous session where the client sends control messages over a WebRTC DataChannel and receives -generated video frames over media tracks. When the browser runs on a laptop through VS Code Remote SSH, configure -TURN over TCP and forward ports `8091` and `3478`; port `8088` does not need forwarding because the demo proxies -signaling requests. Keep local port `3478` equal to remote port `3478`; the forwarded 8091 port may use any available -local port. Without VS Code, run the equivalent tunnel from a terminal on the laptop: +Terminal 4 — serve the browser controller and proxy its session API: ```bash -ssh -N -o ExitOnForwardFailure=yes -o ServerAliveInterval=30 \ - -L 8091:127.0.0.1:8091 \ - -L 3478:127.0.0.1:3478 \ - USER@SERVER_HOST +python examples/stream_server/livekit_bidirectional_demo.py \ + --server-url http://127.0.0.1:8088 --port 8092 --no-open ``` -Then open `http://localhost:8091`. The TURN command and credentials above are development examples. See the -[stream server guide](docs/en/stream_server.md) and the -[LingBot example README](examples/lingbot/README.md) for coturn startup and the tested four-H100 setup. +For VS Code Remote SSH, forward remote TCP ports `8092`, `7880`, and `3478` to the same local ports; `8088` does not +need forwarding because the page proxies the TeleFuser API. Open `http://127.0.0.1:8092`, select an initial image, +click **Start**, and use the on-page controls or `W/A/S/D` and arrow keys. A successful connection shows a video +track plus `control_state`, generation-stage, and chunk status messages. -If the browser runs on the same physical machine as TeleFuser, no SSH tunnel or TURN server is needed. Unset all -`TELEFUSER_TURN_*` variables, start the service on `127.0.0.1:8088`, run the demo without any `--turn-*` or -`--force-turn-relay` arguments, and open `http://localhost:8091`. This does not apply when only the shell is on the -server through SSH but the browser still runs on a laptop. +Check the server independently with `curl http://127.0.0.1:8088/v1/service/health`. To stop the stack, stop the +browser session or close the page first, then press Ctrl+C in terminals 4, 3, 2, and 1. These loopback addresses, +static credentials, disabled TURN TLS, and `--allow-loopback-peers` are for trusted development only. See the +[stream server guide](docs/en/stream_server.md) for LiveKit Cloud, production networking, session APIs, and +troubleshooting. ### 3. Batch Service Mode @@ -162,7 +178,7 @@ See [docs/en/service.md](docs/en/service.md) for full API details. TeleFuser uses a layered runtime architecture that maps cleanly to the repository structure: -1. **Access layer**: FastAPI task APIs and WebRTC streaming entrypoints. +1. **Access layer**: FastAPI task APIs and LiveKit-backed stream room/session entrypoints. 2. **Service layer**: request routing, task management, stream sessions, replica pools, and integration with pipeline execution. 3. **Pipeline abstraction layer**: model-specific `BasePipeline` / `BaseStage` components, with an actor-based streaming orchestrator for bounded dataflow, session ordering, metrics, and cleanup. 4. **Model and optimization layer**: model loading, attention selection, quantization, offload, LoRA, and cache integration. @@ -172,7 +188,7 @@ Relevant directories: ```text telefuser/ -├── service/ # REST APIs, streaming APIs, WebRTC integration +├── service/ # REST APIs and LiveKit-backed streaming ├── orchestrator/ # Request orchestration and actor-based streaming scheduler ├── pipelines/ # Model-specific pipelines ├── distributed/ # TP / SP / FSDP / Ray utilities @@ -188,7 +204,7 @@ telefuser/ | Pipeline | Task | Notes | |----------|------|-------| -| `LingBot-World v2` | Bidirectional world-model streaming | Interactive WebRTC control loop via [examples/lingbot/lingbot_world_v2_image_to_video_h100.py](examples/lingbot/lingbot_world_v2_image_to_video_h100.py) | +| `LingBot-World v2` | Bidirectional world-model streaming | LiveKit control loop via [examples/lingbot/lingbot_world_v2_image_to_video_h100.py](examples/lingbot/lingbot_world_v2_image_to_video_h100.py) | | `LiveAct` | S2V | Speech-driven talking head generation via [examples/liveact/liveact_s2v_h100.py](examples/liveact/liveact_s2v_h100.py) | | `FlashVSR` | VSR | Streaming video super-resolution via [examples/flashvsr/README.md](examples/flashvsr/README.md) | @@ -215,7 +231,7 @@ See [examples/README.md](examples/README.md) for the example runner and baseline ## Documentation - [docs/en/service.md](docs/en/service.md): REST serving, task APIs, OpenAI-compatible APIs -- [docs/en/stream_server.md](docs/en/stream_server.md): continuous streaming and WebRTC protocols +- [docs/en/stream_server.md](docs/en/stream_server.md): LiveKit streaming, session APIs, data topics, and deployment - [docs/en/stream_scheduler.md](docs/en/stream_scheduler.md): actor-based stage scheduling, backpressure, lifecycle, metrics, and LingBot placement - [docs/en/parallel.md](docs/en/parallel.md): distributed inference architecture - [docs/en/latent_cache.md](docs/en/latent_cache.md): CacheSeek latent cache integration diff --git a/README_zh.md b/README_zh.md index 9f2fd1d..2dfa87a 100644 --- a/README_zh.md +++ b/README_zh.md @@ -17,6 +17,8 @@ TeleFuser 是一个面向世界模型推理与多模态生成的高性能运行 ## News 📰 +- ✨ **2026-07-27**:统一使用 LiveKit 流式后端,支持 room 会话、worker 准入、浏览器自动重连,以及 + server-push 和 bidirectional 两种 pipeline contract。 - ✨ **2026-07-22**:**NEW** 新增 [**LingBot-Video**](examples/lingbot_video/README.md) 支持,覆盖 Dense/MoE T2I、T2V、TI2V、原生四卡 CFG/SP 推理与内存直传 MoE refiner。 - ✨ **2026-07-15**:新增 [**LingBot-World v2**](https://github.com/Robbyant/lingbot-world-v2) 支持,支持离线生成、交互式 WebRTC 流和多卡推理。 @@ -39,10 +41,12 @@ TeleFuser 是一个面向世界模型推理与多模态生成的高性能运行 - **面向世界模型的运行时**:支持连续视频生成、交互式会话和双向控制闭环。 - **ADF (AI Dev First)**:通过清晰的仓库分层、Pipeline Contract、示例和文档约束,让 AI Agent 能理解能力边界、遵循项目开发流程,并高效扩展 Pipeline。 - **流式 Pipeline 调度器**:基于 actor 管理有状态 Stage,提供有界 artifact edge、session 顺序、backpressure、生命周期清理和显式 resource group。 -- **流式传输能力**:基于 WebRTC 的媒体流传输,并结合 DataChannel 实现实时控制。 +- **流式传输能力**:LiveKit-backed WebRTC 同时支持 server-push 媒体和稳定的双向会话,提供 room 生命周期、 + 重连、参与者角色和可靠控制消息。 - **可扩展 GPU 运行时**:支持多 GPU、张量并行、序列并行、Ray 部署和分布式工作节点编排。 - **推理优化栈**:包含 Triton Kernel、优化注意力后端、量化、卸载、特征缓存和 CacheSeek latent cache 集成。 -- **统一服务方式**:既支持本地 Python 调用,也支持 `telefuser serve` 和 `telefuser stream-serve` 两种服务模式。 +- **统一服务方式**:支持本地 Python 调用、任务 API `telefuser serve`,以及基于 LiveKit room/media 的 + `telefuser stream-serve`。 ## 快速开始 @@ -62,7 +66,8 @@ TeleFuser 不依赖 `tf-kernel` 也能运行。目前没有发布 tf-kernel 预 `tf-kernel/` 下的 Makefile 从源码构建。编译方法见 [tf-kernel README](tf-kernel/README_zh.md),安装验证、 支持配置和常见问题见 [tf-kernel 安装与使用指南](docs/zh/tf_kernel.md)。 -默认安装已通过 `aiortc` 包含 WebRTC 流式服务能力。 +基础安装已包含 `telefuser stream-serve` 使用的 LiveKit Python SDK;LiveKit Cloud 项目或自托管 LiveKit +Server 需要单独运行。 ### 1. 批量视频推理 @@ -84,64 +89,74 @@ video = pipe( ) ``` -### 2. 实时世界模型 Demo +### 2. 实时世界模型 WebRTC Demo -TeleFuser 当前提供了 `LingBot-World v2` 的双向 WebRTC Demo。 -LingBot-World v2 使用相机控制和 v2 PPL 默认值;其流式示例将单个会话上限设为两分钟。 +TeleFuser 通过 LiveKit 传输 `LingBot-World v2`。LingBot-World v2 使用相机控制和 v2 PPL 默认值;其流式 +示例将单个会话上限设为两分钟。 LingBot 的离线与服务执行共用 actor scheduler。即使位于同一张 GPU,encode、DiT 和 decode 也可以重叠; 仅在显存放置需要时移动 Stage。详见[流式调度器指南](docs/zh/stream_scheduler.md)。 -通过 VS Code Remote SSH 从笔记本浏览器访问时,coturn 是唯一需要额外安装的系统软件,不需要增加 -Python 包。在 Debian 或 Ubuntu 上执行: +仓库内浏览器页面强制使用 TCP TURN relay,以便同一套配置可通过 VS Code Remote SSH 工作。因此完整的本地 +开发环境包含四个进程:coturn、LiveKit Server、TeleFuser 和浏览器页面。先安装一次 LiveKit Server,并 +通过操作系统的包管理器安装 `coturn`: ```bash +# Debian/Ubuntu;其他平台请安装对应的 coturn 软件包。 sudo apt-get update sudo apt-get install -y coturn + +# LiveKit Server 只需安装一次。 +curl -sSL https://get.livekit.io | bash +``` + +然后从仓库根目录在四个独立终端中依次运行以下命令。 + +终端 1——启动仅供开发使用的 TCP TURN relay: + +```bash +turnserver -n -m 1 \ + --listening-ip=127.0.0.1 --relay-ip=127.0.0.1 \ + --listening-port=3478 --min-port=49160 --max-port=49200 \ + --user=livekit-demo:livekit-demo-password \ + --realm=livekit.local --fingerprint --lt-cred-mech \ + --no-tls --no-dtls --no-cli --allow-loopback-peers +``` + +终端 2——使用开发凭据(`devkey` / `secret`)启动 LiveKit: + +```bash +livekit-server --dev ``` -该软件包同时提供 `turnserver` 和用于验证的 `turnutils_uclient`。如果这两个命令已经存在,或者浏览器 -和 GPU 服务运行在同一台物理机器上,则可以跳过安装。 +终端 3——加载四卡 LingBot-World v2 服务: ```bash TF_MODEL_ZOO_PATH=/path/to/model_zoo \ CUDA_VISIBLE_DEVICES=0,1,2,3 \ -TELEFUSER_TURN_SERVER='turn:127.0.0.1:3478?transport=tcp' \ -TELEFUSER_TURN_USERNAME=telefuser \ -TELEFUSER_TURN_CREDENTIAL=telefuser-turn \ telefuser stream-serve examples/lingbot/lingbot_world_v2_image_to_video_h100.py \ - --gpu-num 4 -p 8088 --host 0.0.0.0 --skip-validation - -python examples/stream_server/webrtc_bidirectional_demo.py \ - --server-url http://127.0.0.1:8088 \ - --port 8091 \ - --image-path examples/data/lingbot_world_fast/image.jpg \ - --turn-url 'turn:localhost:3478?transport=tcp' \ - --turn-username telefuser --turn-credential telefuser-turn \ - --force-turn-relay --ice-gather-timeout-ms 30000 --no-open + --livekit-url ws://127.0.0.1:7880 \ + --livekit-api-key devkey --livekit-api-secret secret \ + --num-workers 1 --worker-gpu-map 0,1,2,3 \ + --port 8088 --skip-validation ``` -该流程会启动一个持续运行的会话:客户端通过 WebRTC DataChannel 发送控制消息,服务端通过媒体轨道 -持续回传生成视频。当浏览器运行在笔记本上,并通过 VS Code Remote SSH 访问远端服务器时,需要配置 -TCP TURN,并转发 `8091` 和 `3478` 端口。由于 demo 会代理信令请求,因此不需要转发 `8088`。本地 -`3478` 应保持映射到远端 `3478`;8091 可以映射到任意可用的本地端口。不使用 VS Code 时,可以在 -笔记本终端中建立等效的 OpenSSH 隧道: +终端 4——启动浏览器控制页面及其 session API 代理: ```bash -ssh -N -o ExitOnForwardFailure=yes -o ServerAliveInterval=30 \ - -L 8091:127.0.0.1:8091 \ - -L 3478:127.0.0.1:3478 \ - USER@SERVER_HOST +python examples/stream_server/livekit_bidirectional_demo.py \ + --server-url http://127.0.0.1:8088 --port 8092 --no-open ``` -然后打开 `http://localhost:8091`。上面的 TURN 账号密码仅作为开发配置示例。coturn 启动方式、生产环境 -注意事项及四张 H100 的实测配置见 [流服务文档](docs/zh/stream_server.md) -和 [LingBot example README](examples/lingbot/README.md)。 +使用 VS Code Remote SSH 时,把远端 TCP `8092`、`7880` 和 `3478` 映射到相同本地端口;页面会代理 +TeleFuser API,因此无需映射 `8088`。打开 `http://127.0.0.1:8092`,选择初始图片,点击 **Start**, +再使用页面按钮或 `W/A/S/D` 和方向键控制相机。成功连接后会显示视频轨道以及 `control_state`、生成 Stage +和 chunk 状态消息。 -如果浏览器和 TeleFuser 服务运行在同一台物理机器上,则不需要 SSH 隧道或 TURN 服务。清除所有 -`TELEFUSER_TURN_*` 环境变量,让服务监听 `127.0.0.1:8088`,启动 demo 时不要传入任何 `--turn-*` -或 `--force-turn-relay` 参数,然后打开 `http://localhost:8091`。如果只是通过 SSH 登录服务器、浏览器 -仍然运行在笔记本上,则不属于本机访问,仍需使用上述端口转发和 TURN 配置。 +可用 `curl http://127.0.0.1:8088/v1/service/health` 独立检查服务。停止时先结束浏览器 session 或关闭页面, +再按终端 4、3、2、1 的顺序按 Ctrl+C。Loopback 地址、静态凭据、禁用 TURN TLS 和 +`--allow-loopback-peers` 仅适用于可信开发环境。LiveKit Cloud、生产网络、session API 和故障排查见 +[流服务文档](docs/zh/stream_server.md)。 ### 3. 批处理服务模式 @@ -161,7 +176,7 @@ TeleFuser 对外提供: TeleFuser 采用分层运行时架构,并与仓库目录结构保持一致: -1. **接入层**:FastAPI 任务接口与 WebRTC 流式入口。 +1. **接入层**:FastAPI 任务接口和 LiveKit-backed stream room/session 入口。 2. **服务层**:请求路由、任务管理、流式会话、副本池,以及与 Pipeline 执行过程的集成。 3. **Pipeline 抽象层**:模型相关的 `BasePipeline` / `BaseStage` 组件;actor-based streaming orchestrator 提供有界数据流、session 顺序、指标和清理。 4. **模型与优化层**:模型加载、注意力选择、量化、offload、LoRA、cache 集成。 @@ -171,7 +186,7 @@ TeleFuser 采用分层运行时架构,并与仓库目录结构保持一致: ```text telefuser/ -├── service/ # REST API、流式 API、WebRTC 集成 +├── service/ # REST API 和 LiveKit-backed 流服务 ├── orchestrator/ # 请求编排与基于 actor 的流式调度 ├── pipelines/ # 模型级 Pipeline 实现 ├── distributed/ # TP / SP / FSDP / Ray 等并行能力 @@ -187,7 +202,7 @@ telefuser/ | Pipeline | 任务 | 说明 | |----------|------|------| -| `LingBot-World v2` | 双向世界模型流式推理 | 交互式 WebRTC 控制闭环,见 [examples/lingbot/lingbot_world_v2_image_to_video_h100.py](examples/lingbot/lingbot_world_v2_image_to_video_h100.py) | +| `LingBot-World v2` | 双向世界模型流式推理 | LiveKit 控制闭环,见 [examples/lingbot/lingbot_world_v2_image_to_video_h100.py](examples/lingbot/lingbot_world_v2_image_to_video_h100.py) | | `LiveAct` | S2V | 语音驱动数字人视频生成,见 [examples/liveact/liveact_s2v_h100.py](examples/liveact/liveact_s2v_h100.py) | | `FlashVSR` | VSR | 流式视频超分,见 [examples/flashvsr/README.md](examples/flashvsr/README.md) | @@ -214,7 +229,7 @@ telefuser/ ## 文档 - [docs/zh/service.md](docs/zh/service.md):REST 服务、任务 API、OpenAI 兼容接口 -- [docs/zh/stream_server.md](docs/zh/stream_server.md):连续流式推理与 WebRTC 协议 +- [docs/zh/stream_server.md](docs/zh/stream_server.md):LiveKit 流服务、session API、data topic 和部署 - [docs/zh/stream_scheduler.md](docs/zh/stream_scheduler.md):基于 actor 的 Stage 调度、backpressure、生命周期、指标和 LingBot 卡位 - [docs/zh/parallel.md](docs/zh/parallel.md):分布式推理架构 - [docs/zh/latent_cache.md](docs/zh/latent_cache.md):CacheSeek latent cache 集成 diff --git a/benchmarks/baseline/sglang_lingbot_stream/README.md b/benchmarks/baseline/sglang_lingbot_stream/README.md deleted file mode 100644 index 633d3d9..0000000 --- a/benchmarks/baseline/sglang_lingbot_stream/README.md +++ /dev/null @@ -1,53 +0,0 @@ -# SGLang-Diffusion LingBot Stream Baseline - -This target compares TeleFuser LingBot streaming with the diffusion runtime in -`sgl-project/sglang` (`sglang.multimodal_gen`). It uses WebSocket + MessagePack while -TeleFuser uses WebRTC + DataChannel; AIPerf normalizes both into the same session and -control timeline. - -## Requirements - -- a version-pinned SGLang checkout that provides `LingBotWorldCausalDMDPipeline`; -- `robbyant/lingbot-world-fast-diffusers` or an equivalent local model path; -- the AIPerf checkout prepared by `scripts/setup_aiperf_repo.sh`. - -The launcher does not monkeypatch SGLang internals. Missing dependencies or incompatible -CUDA kernels must be fixed in the SGLang environment or recorded as a failed -qualification, not hidden behind an unversioned shim. - -## Start the target - -```bash -bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh -``` - -Common overrides: - -```bash -SGLANG_PYTHON=/path/to/venv/bin/python \ -SGLANG_LINGBOT_MODEL_PATH=/path/to/model \ -SGLANG_LINGBOT_NUM_GPUS=1 \ -SGLANG_LINGBOT_ULYSSES_DEGREE=1 \ - bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh -``` - -The default address is `http://127.0.0.1:30000`; readiness is checked at `/health`. - -## Run AIPerf - -```bash -bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh - -bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh \ - benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json -``` - -The baseline reuses -`benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json`. AIPerf maps the shared -directional controls to SGLang camera actions and keeps implementation-specific fields -as raw evidence. - -For a valid performance comparison, use the same accelerator count, prompt, first -frame, FPS, session window, control trace, dtype, attention/cache geometry, and offload -policy. Record the exact SGLang and model revisions. Mock, native fallback, CPU offload, -and layerwise offload runs require separate qualifications. diff --git a/benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml b/benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml deleted file mode 100644 index 488c60b..0000000 --- a/benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml +++ /dev/null @@ -1,104 +0,0 @@ -# Benchmark contract example for a WebSocket stream-world baseline. -# Enum comments use "one-of". List comments use "one-or-more". -contract_version: v1 # one-of: v1. Bump only when the contract schema changes. -name: sglang_lingbot_world_stream # Stable benchmark target id used in reports and automation. -mode: stream_world # one-of: batch_video, stream_world. -implementation: sglang_diffusion # Example values: telefuser, diffusers, sglang_diffusion. -model_family: lingbot_world_fast # Example values: wan_video, lingbot_world_fast, hunyuan_video, ltx_video. -model: robbyant/lingbot-world-fast-diffusers # Concrete model or service profile under test. -supported_tasks: # one-or-more: t2v, i2v, ti2v, bidirectional. - - bidirectional -transport: websocket # one-of: http, http_polling, websocket, sse, webrtc. -adapter: sglang_websocket # Built-in AIPerf adapter; target repository owns no transport implementation. -endpoint: - health_path: /health # Service readiness path. - metadata_path: /v1/models # Optional SGLang model and pipeline identity snapshot. - websocket_path: /v1/realtime_video/generate # WebSocket path for realtime video sessions. - models_path: /v1/models # Optional model metadata path. -request_encoding: - message_format: msgpack # one-of for this baseline: msgpack. - init_required_fields: # one-or-more. Required fields in the first WebSocket message. - - type - - prompt - - first_frame - - size - - fps - - num_frames - init_parameters: # Mapping from benchmark parameter names to SGLang init payload fields. - model: model - prompt: prompt - image_path: first_frame - size: size - fps: fps - num_frames: num_frames - control_channel: - transport: websocket_message # one-of for WebSocket: websocket_message. - message_type: event # Runtime control messages use type=event. - kind: camera_actions # one-of for LingBot realtime controls: camera_actions, prompt. - payload_mode: state # one-of for camera_actions: script, state. - action_tokens: # one-or-more. SGLang LingBot camera action tokens. - - w - - a - - s - - d - - i - - j - - k - - l - key_mapping: - ArrowUp: w - ArrowDown: s - ArrowLeft: a - ArrowRight: d -result_delivery: - media: websocket_frame_batch # one-of for this baseline: websocket_frame_batch. - metadata: websocket_chunk_stats # one-of for this baseline: websocket_chunk_stats. - session_log: sessions.jsonl # Per-session result records. - event_log: events/{phase}_{logical_session_index}_{session_id}.jsonl # Per-session event trace template. -workload: - mode: bidirectional # one-of: server_push, bidirectional. - task: bidirectional # one-of for this service: bidirectional. - size: 832x480 - fps: 16 - session_count: 1 - warmup_sessions: 1 - session_duration_s: 90.0 - control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json # Timed control-event workload. - request_extra: # Service-specific request config passed through to SGLang. - num_frames: 9 - num_inference_steps: 4 - guidance_scale: 1.0 - realtime_causal_sink_size: 6 - realtime_causal_kv_cache_num_frames: 9 - realtime_output_format: webp # one-of: webp, jpeg, raw. - output_compression: 95 - max_chunks: 8 -metrics: # one-or-more. Choose all metrics emitted by this benchmark mode. - - connected_latency_ms - - first_frame_latency_ms - - first_metadata_latency_ms - - stream_fps - - session_runtime_s - - frames_received - - control_ack_latency_ms - - control_to_next_frame_latency_ms - - chunk_request_prepare_seconds - - chunk_compute_seconds - - chunk_encode_seconds - - chunk_output_pacing_seconds - - chunk_output_header_write_seconds - - chunk_output_payload_write_seconds - - chunk_output_write_seconds - - chunk_total_seconds - - chunk_compute_fps - - chunk_raw_output_bytes - - chunk_wire_output_bytes - - chunk_output_batches - - chunk_peak_reserved_bytes - - success_rate -limits: - active_sessions: 1 # Current SGLang realtime endpoint accepts one active session for this target. -artifacts: # Paths consumed by automation and documentation. - config: benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json - control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json - runner: aiperf profile --stream-config diff --git a/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json b/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json deleted file mode 100644 index 76597d4..0000000 --- a/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "contract": "benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml", - "server_url": "http://127.0.0.1:30000", - "mode": "bidirectional", - "task": "bidirectional", - "prompt": "walk forward through the scene", - "image_path": "examples/data/1.png", - "fps": 16, - "session_count": 1, - "warmup_sessions": 1, - "warmup_chunks": 1, - "session_duration_s": 90.0, - "stagger_s": 0.0, - "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", - "request_extra": { - "num_frames": 9, - "num_inference_steps": 4, - "guidance_scale": 1.0, - "realtime_causal_sink_size": 6, - "realtime_causal_kv_cache_num_frames": 9, - "realtime_output_format": "webp", - "output_compression": 95, - "max_chunks": 8, - "realtime_output_pacing": false - }, - "transport": { - "connect_timeout_s": 60.0, - "message_timeout_s": 180.0 - }, - "server_metrics": { - "enabled": false - }, - "artifacts_dir": "artifacts/sglang_lingbot_stream/stream_lingbot_compare" -} diff --git a/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json b/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json deleted file mode 100644 index 6ad373c..0000000 --- a/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json +++ /dev/null @@ -1,33 +0,0 @@ -{ - "contract": "benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml", - "server_url": "http://127.0.0.1:30000", - "mode": "bidirectional", - "task": "bidirectional", - "prompt": "walk forward through the scene", - "image_path": "examples/data/1.png", - "fps": 16, - "session_count": 1, - "warmup_sessions": 0, - "session_duration_s": 12.0, - "stagger_s": 0.0, - "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", - "request_extra": { - "num_frames": 9, - "num_inference_steps": 4, - "guidance_scale": 1.0, - "realtime_causal_sink_size": 6, - "realtime_causal_kv_cache_num_frames": 9, - "realtime_output_format": "webp", - "output_compression": 95, - "max_chunks": 8, - "realtime_output_pacing": false - }, - "transport": { - "connect_timeout_s": 30.0, - "message_timeout_s": 120.0 - }, - "server_metrics": { - "enabled": false - }, - "artifacts_dir": "artifacts/sglang_lingbot_stream/stream_lingbot_quick" -} diff --git a/benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh b/benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh deleted file mode 100755 index fe34921..0000000 --- a/benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd)" -cd "${ROOT_DIR}" - -if [[ -z "${CUDA_HOME:-}" && -d /usr/local/cuda ]]; then - export CUDA_HOME=/usr/local/cuda -fi - -if [[ -n "${SGLANG_EXTRA_PYTHONPATH:-}" ]]; then - export PYTHONPATH="${SGLANG_EXTRA_PYTHONPATH}${PYTHONPATH:+:${PYTHONPATH}}" -fi - -SGLANG_BIN="${SGLANG_BIN:-sglang}" -SGLANG_PYTHON="${SGLANG_PYTHON:-}" -SERVICE_PORT="${SGLANG_LINGBOT_PORT:-30000}" -MODEL_PATH="${SGLANG_LINGBOT_MODEL_PATH:-robbyant/lingbot-world-fast-diffusers}" -MODEL_ID="${SGLANG_LINGBOT_MODEL_ID:-lingbot-world-fast-diffusers}" -MODEL_TYPE="${SGLANG_LINGBOT_MODEL_TYPE:-diffusion}" -PIPELINE_CLASS="${SGLANG_LINGBOT_PIPELINE_CLASS:-LingBotWorldCausalDMDPipeline}" -PERFORMANCE_MODE="${SGLANG_LINGBOT_PERFORMANCE_MODE:-speed}" -ATTENTION_BACKEND_CONFIG="${SGLANG_LINGBOT_ATTENTION_BACKEND_CONFIG:-VSA_sparsity=0.0}" -NUM_GPUS="${SGLANG_LINGBOT_NUM_GPUS:-1}" -ULYSSES_DEGREE="${SGLANG_LINGBOT_ULYSSES_DEGREE:-1}" -DIT_CPU_OFFLOAD="${SGLANG_LINGBOT_DIT_CPU_OFFLOAD:-false}" -TEXT_ENCODER_CPU_OFFLOAD="${SGLANG_LINGBOT_TEXT_ENCODER_CPU_OFFLOAD:-false}" - -if [[ -n "${SGLANG_PYTHON}" ]]; then - SGLANG_CMD=("${SGLANG_PYTHON}" -c "from sglang.cli.main import main; main()") -else - read -r -a SGLANG_CMD <<< "${SGLANG_BIN}" -fi - -exec "${SGLANG_CMD[@]}" serve \ - --model-type "${MODEL_TYPE}" \ - --model-path "${MODEL_PATH}" \ - --model-id "${MODEL_ID}" \ - --pipeline-class-name "${PIPELINE_CLASS}" \ - --performance-mode "${PERFORMANCE_MODE}" \ - --attention-backend-config "${ATTENTION_BACKEND_CONFIG}" \ - --port "${SERVICE_PORT}" \ - --num-gpus "${NUM_GPUS}" \ - --ulysses-degree "${ULYSSES_DEGREE}" \ - --dit-cpu-offload "${DIT_CPU_OFFLOAD}" \ - --text-encoder-cpu-offload "${TEXT_ENCODER_CPU_OFFLOAD}" \ - "$@" diff --git a/benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh b/benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh deleted file mode 100755 index 1385d59..0000000 --- a/benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh +++ /dev/null @@ -1,49 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd)" -cd "${ROOT_DIR}" - -AIPERF_DIR="${ROOT_DIR}/benchmarks/aiperf" -UV_BIN="${AIPERF_UV_BIN:-uv}" -CONFIG_PATH="${1:-benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json}" -if [[ $# -gt 0 ]]; then - shift -fi - -if [[ ! -f "${AIPERF_DIR}/pyproject.toml" ]]; then - echo "AIPerf checkout not found. Run: bash scripts/setup_aiperf_repo.sh" >&2 - exit 1 -fi -if ! command -v "${UV_BIN}" >/dev/null 2>&1; then - echo "uv is required: https://docs.astral.sh/uv/getting-started/installation/" >&2 - exit 1 -fi - -SERVER_URL="${SGLANG_STREAM_BENCH_URL:-http://127.0.0.1:30000}" -SERVER_ARGS=(--stream-server-url "${SERVER_URL}") -for argument in "$@"; do - if [[ "${argument}" == "--stream-server-url" || "${argument}" == --stream-server-url=* ]]; then - SERVER_ARGS=() - break - fi -done - -RESOURCE_ARGS=() -RESOURCE_HISTORY_URL="${AIPERF_HISTORY_URL:-}" -RESOURCE_TARGET_PID="${AIPERF_RESOURCE_TARGET_PID:-${SGLANG_STREAM_BENCH_PID:-}}" -if [[ -n "${RESOURCE_HISTORY_URL}" || -n "${RESOURCE_TARGET_PID}" ]]; then - if [[ -z "${RESOURCE_HISTORY_URL}" || -z "${RESOURCE_TARGET_PID}" ]]; then - echo "AIPERF_HISTORY_URL and AIPERF_RESOURCE_TARGET_PID must be set together" >&2 - exit 2 - fi - RESOURCE_ARGS+=(--stream-resource-history-url "${RESOURCE_HISTORY_URL}") - RESOURCE_ARGS+=(--stream-resource-target-pid "${RESOURCE_TARGET_PID}") -fi - -exec "${UV_BIN}" run --frozen --no-dev --project "${AIPERF_DIR}" \ - aiperf profile \ - --stream-config "${CONFIG_PATH}" \ - "${SERVER_ARGS[@]}" \ - "${RESOURCE_ARGS[@]}" \ - "$@" diff --git a/benchmarks/telefuser_aiperf/README.md b/benchmarks/telefuser_aiperf/README.md new file mode 100644 index 0000000..47e8130 --- /dev/null +++ b/benchmarks/telefuser_aiperf/README.md @@ -0,0 +1,227 @@ +# TeleFuser AIPerf Integration + +This directory contains the TeleFuser-owned benchmark assets for AIPerf. AIPerf itself is not +vendored here, and the LiveKit adapter is loaded directly from this source tree instead of being built or published as +a separate Python distribution. + +Run every command in this guide from the TeleFuser repository root. The user-facing metric definitions and the +latest validated result are documented in +[`docs/en/benchmark_aiperf.md`](../../docs/en/benchmark_aiperf.md). This README is the canonical installation and +operation guide. + +## Prerequisites + +- A working TeleFuser installation with the LingBot-World v2 checkpoints. +- Four CUDA GPUs visible to TeleFuser. The validated configuration used four H100 80 GB GPUs. +- Python 3.10 or newer, Git, curl, and the `livekit-server` executable. + +Install the local LiveKit development server if it is not already available: + +```bash +curl -sSL https://get.livekit.io | bash +``` + +The development server and its default `devkey` / `secret` credentials are only for trusted local testing. + +## Install AIPerf + +The official AIPerf 0.11.0 wheel does not contain the streaming runner. Install the pinned `teleai` source commit +from GitHub. The recommended helper creates an isolated `.venv-aiperf` so AIPerf dependencies do not change the +TeleFuser runtime environment: + +```bash +bash scripts/setup_aiperf.sh +``` + +Successful setup ends by printing the AIPerf version, the pinned VCS commit, and the TeleFuser adapter path. The +default commit is `e977ffbb1648510acec431b2a3fbd1a0f7bb8a35`. + +To install into an existing benchmark environment instead, run: + +```bash +python -m pip uninstall -y aiperf +python -m pip install \ + 'aiperf @ git+https://github.com/ActivePeter/aiperf.git@e977ffbb1648510acec431b2a3fbd1a0f7bb8a35' \ + 'livekit>=1.1.13,<2' +``` + +The uninstall is intentional: the official wheel and the streaming source currently both report version `0.11.0`, +so pip may otherwise keep the non-streaming wheel. + +Pip uses a temporary Git checkout while building the package; no AIPerf checkout is retained in this repository. No +adapter `pyproject.toml`, wheel, or editable install is required. The stream launcher adds this directory to +`PYTHONPATH` before registering the `telefuser_livekit` adapter. + +Use `AIPERF_ENV_DIR` to select another helper-created environment. If AIPerf was installed manually, set +`TELEFUSER_AIPERF_PYTHON` to its Python executable when running the stream launcher, for example: + +```bash +TELEFUSER_AIPERF_PYTHON=/path/to/benchmark-env/bin/python \ + bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` + +## Batch Video + +Start a compatible TeleFuser service, for example: + +```bash +telefuser serve \ + examples/wan_video/wan21_14b_image_to_video_480p_service.py \ + --port 8000 \ + --task i2v +``` + +Run the smoke workload or the fixed Wan2.1 comparison: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh + +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml +``` + +The launcher checks `/v1/service/health` before profiling. Its common overrides are `TELEFUSER_AIPERF_URL`, +`TELEFUSER_AIPERF_HEALTH_URL`, `TELEFUSER_BENCH_NOFILE_LIMIT`, and `AIPERF_BIN`. + +Available batch configs: + +| Config | Purpose | +|---|---| +| `configs/video_generation_quick.yaml` | Connectivity and latency smoke test | +| `configs/video_generation_e2e.yaml` | Warmup, trace, records, and target metrics | +| `configs/video_generation_rate.yaml` | Poisson-arrival load | +| `configs/video_generation_wan21_i2v_480p_compare.yaml` | Fixed Wan2.1 I2V comparison | + +## LingBot-World v2 Streaming + +The v2 pipeline expects the following files below `TF_MODEL_ZOO_PATH`: + +```text +Wan2.2-I2V-A14B/Wan2.1_VAE.pth +Wan2.2-I2V-A14B/models_t5_umt5-xxl-enc-bf16.pth +lingbot/lingbot-world-v2-14b-causal-fast/transformers/model-00001-of-00008.safetensors +... +lingbot/lingbot-world-v2-14b-causal-fast/transformers/model-00008-of-00008.safetensors +``` + +Use three terminals for the local benchmark. In terminal 1, start LiveKit: + +```bash +livekit-server --dev --bind 127.0.0.1 +``` + +Leave this process running. In terminal 2, start the four-GPU LingBot-World v2 service. Replace +`/path/to/model_zoo`; local LiveKit connections must not use host proxy variables: + +```bash +env -u http_proxy -u https_proxy -u HTTP_PROXY -u HTTPS_PROXY -u all_proxy -u ALL_PROXY \ + TF_MODEL_ZOO_PATH=/path/to/model_zoo \ + CUDA_VISIBLE_DEVICES=0,1,2,3 \ + telefuser stream-serve examples/lingbot/lingbot_world_v2_image_to_video_h100.py \ + --livekit-url ws://127.0.0.1:7880 \ + --livekit-api-key devkey \ + --livekit-api-secret secret \ + --num-workers 1 \ + --worker-gpu-map 0,1,2,3 \ + --port 8088 \ + --skip-validation +``` + +Initial model loading and pipeline warmup can take several minutes. Do not start AIPerf until the service reports +`"ready":true`, `"workers_idle":1`, and `"workers_failed":0`: + +```bash +curl --noproxy '*' --fail --silent --show-error \ + http://127.0.0.1:8088/v1/service/health +``` + +`"livekit_connected":false` is normal while no session is active; it does not mean the service is unhealthy. + +In terminal 3, run the one-minute LingBot-World v2 workload: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` + +The v2 workload is the launcher default and the only checked-in stream workload. It requests 59.75 seconds of media +using the model's fixed attention window. Its AIPerf active window is 240 seconds, so a successful command normally +takes about four minutes rather than one minute. Do not terminate it after media generation becomes quiet. + +A successful run prints `Stream profile sessions: 1/1 succeeded`, an artifact directory, and an HTML report path. +Results are written below: + +```text +artifacts/telefuser_aiperf/stream_lingbot_v2_1min// +``` + +The service command clears common proxy variables for the local LiveKit connection. The adapter also bypasses proxy +variables when the target returns a loopback LiveKit URL; remote LiveKit deployments are unchanged. + +The contract records `transport: webrtc` and `transport_provider: livekit`. The adapter creates the TeleFuser +session, joins its room, receives native video tracks, sends reliable controls on `tf.control`, and consumes +`tf.status` and bounded `tf.metrics` messages. It produces AIPerf's standard session results without requiring any +LiveKit-specific changes in AIPerf. + +## Troubleshooting + +- `The pinned streaming-capable AIPerf or LiveKit is not installed`: rerun `bash scripts/setup_aiperf.sh`. +- Connection refused on port 8088: the TeleFuser process is still warming up or has exited; inspect terminal 2. +- `0/1 succeeded` with zero received frames: confirm LiveKit is still running, restart the TeleFuser service, wait for + one idle worker, and rerun the benchmark. +- LiveKit connection errors on localhost: start TeleFuser with the proxy variables removed exactly as shown above. +- `aiperf_commit` is `unknown` inside a source-install artifact: use the commit printed by `setup_aiperf.sh`; AIPerf + only embeds `_build_info.py` in its CI-built wheels. + +## History And Resources + +AIPerf history and active resource collection require GreptimeDB. Start persistent storage, then the AIPerf history +service: + +```bash +docker volume create aiperf-greptime-data +docker run -d --name aiperf-greptime --restart unless-stopped \ + -p 127.0.0.1:4000:4000 \ + -v aiperf-greptime-data:/greptimedb_data \ + greptime/greptimedb:latest \ + standalone start \ + --http-addr 0.0.0.0:4000 \ + --data-home /greptimedb_data + +.venv-aiperf/bin/aiperf history serve \ + --greptime-url http://127.0.0.1:4000 \ + --greptime-database public \ + --artifact-root artifacts \ + --host 127.0.0.1 \ + --port 8095 +``` + +Set `AIPERF_HISTORY_URL` and `AIPERF_RESOURCE_TARGET_PID` before the batch launcher to collect the target process +tree. History failures do not silently fall back to an in-memory or file-only database. + +## Layout And Verification + +```text +configs/ Reproducible batch and streaming workloads +data/ Prompt and control inputs +scripts/ Batch and streaming launchers +telefuser_aiperf/ Source-loaded LiveKit adapter +tests/ Adapter tests +*_contract.yaml Target and transport capability contracts +``` + +Runtime use does not require pytest. To run the optional adapter checks, install the test-only dependencies into the +AIPerf environment first, then run the checks from the repository root: + +```bash +.venv-aiperf/bin/python -m pip install \ + 'pytest>=7' \ + 'pytest-asyncio>=0.21' + +PYTHONPATH=benchmarks/telefuser_aiperf \ + .venv-aiperf/bin/python -m pytest \ + benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py + +bash -n \ + scripts/setup_aiperf.sh \ + benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` diff --git a/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json deleted file mode 100644 index 31ff159..0000000 --- a/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json +++ /dev/null @@ -1,42 +0,0 @@ -{ - "contract": "benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml", - "server_url": "http://127.0.0.1:8088", - "mode": "bidirectional", - "task": "bidirectional", - "prompt": "walk forward through the scene", - "image_path": "examples/data/1.png", - "fps": 16, - "session_count": 1, - "warmup_sessions": 0, - "warmup_chunks": 1, - "session_duration_s": 30.0, - "stagger_s": 0.0, - "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", - "request_extra": { - "chunk_size": 3, - "frame_num": 81, - "sample_shift": 5.0, - "control_mode": "cam", - "show_control_hud": false, - "benchmark_metrics": true - }, - "transport": { - "connect_timeout_s": 30.0, - "frame_timeout_s": 60.0, - "ice_gather_timeout_s": 5.0, - "shutdown_timeout_s": 5.0, - "receive_audio": false - }, - "server_metrics": { - "enabled": true, - "urls": [ - "http://127.0.0.1:8088/v1/service/metrics" - ], - "collection_interval_s": 1.0, - "export_raw_jsonl": true - }, - "observability": { - "mapping": "builtin:telefuser" - }, - "artifacts_dir": "artifacts/telefuser_aiperf/stream_lingbot_quick" -} diff --git a/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_v2_1min.json similarity index 57% rename from benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json rename to benchmarks/telefuser_aiperf/configs/stream_lingbot_world_v2_1min.json index e922f1f..0548e90 100644 --- a/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json +++ b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_v2_1min.json @@ -1,42 +1,40 @@ { - "contract": "benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml", + "contract": "benchmarks/telefuser_aiperf/stream_lingbot_world_v2_benchmark_contract.yaml", "server_url": "http://127.0.0.1:8088", "mode": "bidirectional", "task": "bidirectional", "prompt": "walk forward through the scene", - "image_path": "examples/data/1.png", + "image_path": "examples/data/lingbot_world_fast/image.jpg", "fps": 16, "session_count": 1, - "warmup_sessions": 1, + "warmup_sessions": 0, "warmup_chunks": 1, - "session_duration_s": 90.0, + "session_duration_s": 240.0, "stagger_s": 0.0, "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", "request_extra": { - "chunk_size": 3, - "frame_num": 81, - "sample_shift": 5.0, + "chunk_size": 4, + "frame_num": 957, + "max_duration_seconds": 60.0, + "sample_shift": 10.0, "control_mode": "cam", "show_control_hud": false, "benchmark_metrics": true }, "transport": { "connect_timeout_s": 60.0, - "frame_timeout_s": 180.0, - "ice_gather_timeout_s": 5.0, - "shutdown_timeout_s": 5.0, + "frame_timeout_s": 300.0, + "shutdown_timeout_s": 10.0, "receive_audio": false }, "server_metrics": { "enabled": true, - "urls": [ - "http://127.0.0.1:8088/v1/service/metrics" - ], + "urls": ["http://127.0.0.1:8088/v1/service/metrics"], "collection_interval_s": 1.0, "export_raw_jsonl": true }, "observability": { "mapping": "builtin:telefuser" }, - "artifacts_dir": "artifacts/telefuser_aiperf/stream_lingbot_compare" + "artifacts_dir": "artifacts/telefuser_aiperf/stream_lingbot_v2_1min" } diff --git a/benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json b/benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json index d70b5dd..46094ab 100644 --- a/benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json +++ b/benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json @@ -2,59 +2,31 @@ "events": [ { "delay_s": 1.0, - "message": { - "type": "control", - "key": "ArrowUp", - "action": "press" - } + "message": {"type": "control", "key": "ArrowUp", "action": "press"} }, { "delay_s": 1.8, - "message": { - "type": "control", - "key": "ArrowUp", - "action": "release" - } + "message": {"type": "control", "key": "ArrowUp", "action": "release"} }, { "delay_s": 2.8, - "message": { - "type": "control", - "key": "ArrowLeft", - "action": "press" - } + "message": {"type": "control", "key": "ArrowLeft", "action": "press"} }, { "delay_s": 3.6, - "message": { - "type": "control", - "key": "ArrowLeft", - "action": "release" - } + "message": {"type": "control", "key": "ArrowLeft", "action": "release"} }, { "delay_s": 4.6, - "message": { - "type": "control", - "key": "ArrowRight", - "action": "press" - } + "message": {"type": "control", "key": "ArrowRight", "action": "press"} }, { "delay_s": 5.4, - "message": { - "type": "control", - "key": "ArrowRight", - "action": "release" - } + "message": {"type": "control", "key": "ArrowRight", "action": "release"} }, { "delay_s": 6.4, - "message": { - "type": "control", - "key": "ArrowUp", - "action": "press" - } + "message": {"type": "control", "key": "ArrowUp", "action": "press"} } ] } diff --git a/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh index 92e9947..4dda0af 100755 --- a/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +++ b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh @@ -4,63 +4,26 @@ set -euo pipefail ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" cd "${ROOT_DIR}" -AIPERF_DIR="${ROOT_DIR}/benchmarks/aiperf" -UV_BIN="${AIPERF_UV_BIN:-uv}" -CONFIG_PATH="${1:-benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json}" +CONFIG_PATH="${1:-benchmarks/telefuser_aiperf/configs/stream_lingbot_world_v2_1min.json}" if [[ $# -gt 0 ]]; then shift fi -if [[ ! -f "${AIPERF_DIR}/pyproject.toml" ]]; then - echo "AIPerf checkout not found. Run: bash scripts/setup_aiperf_repo.sh" >&2 - exit 1 +DEFAULT_PYTHON="${ROOT_DIR}/.venv-aiperf/bin/python" +ADAPTER_ROOT="${ROOT_DIR}/benchmarks/telefuser_aiperf" +if [[ -n "${TELEFUSER_AIPERF_PYTHON:-}" ]]; then + AIPERF_PYTHON="${TELEFUSER_AIPERF_PYTHON}" +elif [[ -x "${DEFAULT_PYTHON}" ]]; then + AIPERF_PYTHON="${DEFAULT_PYTHON}" +else + AIPERF_PYTHON="$(command -v python || true)" fi -if ! command -v "${UV_BIN}" >/dev/null 2>&1; then - echo "uv is required: https://docs.astral.sh/uv/getting-started/installation/" >&2 +if [[ -z "${AIPERF_PYTHON}" ]] || ! command -v "${AIPERF_PYTHON}" >/dev/null 2>&1 \ + || ! PYTHONPATH="${ADAPTER_ROOT}${PYTHONPATH:+:${PYTHONPATH}}" \ + "${AIPERF_PYTHON}" -c 'import livekit, telefuser_aiperf' >/dev/null 2>&1; then + echo "The pinned streaming-capable AIPerf or LiveKit is not installed. Run: bash scripts/setup_aiperf.sh" >&2 exit 1 fi -SERVER_URL="${TELEFUSER_STREAM_BENCH_URL:-http://127.0.0.1:8088}" -SERVER_ARGS=(--stream-server-url "${SERVER_URL}") -for argument in "$@"; do - if [[ "${argument}" == "--stream-server-url" || "${argument}" == --stream-server-url=* ]]; then - SERVER_ARGS=() - break - fi -done -ICE_HOST_IPS="${TELEFUSER_STREAM_BENCH_ICE_HOST_IPS:-}" -ICE_HOST_ARGS=() -if [[ -n "${ICE_HOST_IPS}" ]]; then - IFS=',' read -r -a _ICE_HOST_IP_ARRAY <<< "${ICE_HOST_IPS}" - for ice_host_ip in "${_ICE_HOST_IP_ARRAY[@]}"; do - if [[ -n "${ice_host_ip}" ]]; then - ICE_HOST_ARGS+=(--stream-ice-host-ip "${ice_host_ip}") - fi - done -fi - -METRICS_ARGS=() -if [[ -n "${TELEFUSER_STREAM_BENCH_METRICS_URL:-}" ]]; then - METRICS_ARGS+=(--stream-server-metrics-url "${TELEFUSER_STREAM_BENCH_METRICS_URL}") -fi - -RESOURCE_ARGS=() -RESOURCE_HISTORY_URL="${AIPERF_HISTORY_URL:-}" -RESOURCE_TARGET_PID="${AIPERF_RESOURCE_TARGET_PID:-${TELEFUSER_STREAM_BENCH_PID:-}}" -if [[ -n "${RESOURCE_HISTORY_URL}" || -n "${RESOURCE_TARGET_PID}" ]]; then - if [[ -z "${RESOURCE_HISTORY_URL}" || -z "${RESOURCE_TARGET_PID}" ]]; then - echo "AIPERF_HISTORY_URL and AIPERF_RESOURCE_TARGET_PID must be set together" >&2 - exit 2 - fi - RESOURCE_ARGS+=(--stream-resource-history-url "${RESOURCE_HISTORY_URL}") - RESOURCE_ARGS+=(--stream-resource-target-pid "${RESOURCE_TARGET_PID}") -fi - -exec "${UV_BIN}" run --frozen --no-dev --project "${AIPERF_DIR}" --extra streaming-webrtc \ - aiperf profile \ - --stream-config "${CONFIG_PATH}" \ - "${SERVER_ARGS[@]}" \ - "${ICE_HOST_ARGS[@]}" \ - "${METRICS_ARGS[@]}" \ - "${RESOURCE_ARGS[@]}" \ - "$@" +export PYTHONPATH="${ADAPTER_ROOT}${PYTHONPATH:+:${PYTHONPATH}}" +exec "${AIPERF_PYTHON}" -m telefuser_aiperf.cli profile --stream-config "${CONFIG_PATH}" "$@" diff --git a/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh b/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh index 2f3869b..92150e6 100755 --- a/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh +++ b/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh @@ -7,20 +7,19 @@ cd "${ROOT_DIR}" CONFIG_PATH="${1:-benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml}" SERVER_URL="${TELEFUSER_AIPERF_URL:-http://127.0.0.1:8000}" HEALTH_URL="${TELEFUSER_AIPERF_HEALTH_URL:-${SERVER_URL}/v1/service/health}" -AIPERF_DIR="${ROOT_DIR}/benchmarks/aiperf" -UV_BIN="${AIPERF_UV_BIN:-uv}" NOFILE_LIMIT="${TELEFUSER_BENCH_NOFILE_LIMIT:-8192}" if ! ulimit -n "${NOFILE_LIMIT}" >/dev/null 2>&1; then echo "Warning: failed to raise open-file limit to ${NOFILE_LIMIT}" >&2 fi -if [[ ! -f "${AIPERF_DIR}/pyproject.toml" ]]; then - echo "AIPerf checkout not found. Run: bash scripts/setup_aiperf_repo.sh" >&2 - exit 1 +DEFAULT_BIN="${ROOT_DIR}/.venv-aiperf/bin/aiperf" +AIPERF_BIN="${AIPERF_BIN:-${DEFAULT_BIN}}" +if [[ ! -x "${AIPERF_BIN}" ]]; then + AIPERF_BIN="$(command -v aiperf || true)" fi -if ! command -v "${UV_BIN}" >/dev/null 2>&1; then - echo "uv is required: https://docs.astral.sh/uv/getting-started/installation/" >&2 +if [[ -z "${AIPERF_BIN}" ]]; then + echo "aiperf is not installed. Run: bash scripts/setup_aiperf.sh" >&2 exit 1 fi @@ -30,4 +29,4 @@ if command -v curl >/dev/null 2>&1; then fi echo "Running AIPerf with config: ${CONFIG_PATH}" -exec "${UV_BIN}" run --frozen --no-dev --project "${AIPERF_DIR}" aiperf profile --config "${CONFIG_PATH}" +exec "${AIPERF_BIN}" profile --config "${CONFIG_PATH}" diff --git a/benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml b/benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml deleted file mode 100644 index be088d2..0000000 --- a/benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml +++ /dev/null @@ -1,79 +0,0 @@ -# Benchmark contract example for a WebRTC stream-world target. -# Enum comments use "one-of". List comments use "one-or-more". -contract_version: v1 # one-of: v1. Bump only when the contract schema changes. -name: telefuser_lingbot_world_fast_stream # Stable benchmark target id used in reports and automation. -mode: stream_world # one-of: batch_video, stream_world. -implementation: telefuser # Example values: telefuser, diffusers, sglang_diffusion. -model_family: lingbot_world_fast # Example values: wan_video, lingbot_world_fast, hunyuan_video, ltx_video. -model: LingBot-World-Fast # Concrete model or service profile under test. -supported_tasks: # one-or-more: t2v, i2v, ti2v, bidirectional. - - bidirectional -transport: webrtc # one-of: http, http_polling, websocket, sse, webrtc. -adapter: telefuser_webrtc # Built-in AIPerf adapter; target repository owns no transport implementation. -endpoint: - health_path: /v1/service/health # Service readiness path. - metadata_path: /v1/service/metadata # Optional target environment and startup phase facts. - offer_path: /v1/stream/webrtc/offer # WebRTC SDP offer/answer path. - delete_path_template: /v1/stream/webrtc/{session_id} # Session cleanup path template. -request_encoding: - offer_content_type: application/json # one-of: application/json. - offer_required_fields: # one-or-more. Required JSON fields in the offer payload. - - sdp - - type - - task - offer_parameters: # Mapping from benchmark parameter names to wire payload fields. - session_id: session_id - prompt: prompt - fps: fps - image_path: image_path - config: config - control_channel: - transport: datachannel # one-of for WebRTC: datachannel. - label: telefuser # DataChannel label expected by the service. - message_types: # one-or-more. Message types understood by the stream benchmark semantics. - - control - - status - - chunk - - done - - error -result_delivery: - media: rtp_video_track # one-of for WebRTC: rtp_video_track. - metadata: datachannel # one-of for WebRTC: datachannel. - session_log: sessions.jsonl # Per-session result records. - event_log: events/{phase}_{logical_session_index}_{session_id}.jsonl # Per-session event trace template. -workload: - mode: bidirectional # one-of: server_push, bidirectional. - task: bidirectional # one-of for this service: bidirectional. - fps: 16 - session_count: 1 - warmup_sessions: 1 - session_duration_s: 90.0 - control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json # Timed control-event workload. - request_extra: # Service-specific request config passed through to the stream service. - chunk_size: 3 - frame_num: 81 - sample_shift: 5.0 - control_mode: cam # one-of for LingBotWorldFast: cam. - show_control_hud: false - benchmark_metrics: true # Emit synchronized runtime/chunk facts for AIPerf aggregation. -metrics: # one-or-more. Choose all metrics emitted by this benchmark mode. - - offer_rtt_ms - - connected_latency_ms - - first_frame_latency_ms - - first_metadata_latency_ms - - stream_fps - - session_runtime_s - - frames_received - - control_ack_latency_ms - - control_to_next_frame_latency_ms - - pipeline_init_seconds - - runtime_creation_seconds - - chunk_compute_seconds - - chunk_compute_fps - - success_rate -limits: - active_sessions: 1 # Service limit for this target. The harness still exposes session_count. -artifacts: # Paths consumed by automation and documentation. - config: benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json - control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json - runner: aiperf profile --stream-config diff --git a/benchmarks/telefuser_aiperf/stream_lingbot_world_v2_benchmark_contract.yaml b/benchmarks/telefuser_aiperf/stream_lingbot_world_v2_benchmark_contract.yaml new file mode 100644 index 0000000..fa06d97 --- /dev/null +++ b/benchmarks/telefuser_aiperf/stream_lingbot_world_v2_benchmark_contract.yaml @@ -0,0 +1,68 @@ +# Benchmark contract for the LiveKit-backed LingBot-World v2 stream target. +contract_version: v1 +name: telefuser_lingbot_world_v2_stream +mode: stream_world +implementation: telefuser +model_family: lingbot_world_v2 +model: LingBot-World-v2-14B-Causal-Fast +supported_tasks: + - bidirectional +transport: webrtc +adapter: telefuser_livekit +transport_provider: livekit +endpoint: + health_path: /v1/service/health + metadata_path: /v1/service/metadata + offer_path: /v1/stream/sessions + delete_path_template: /v1/stream/sessions/{session_id} +request_encoding: + content_type: application/json + create_required_fields: + - identity + - role + - prompt + - config + control_topic: tf.control + control_reliable: true +result_delivery: + media: livekit_video_track + status_topic: tf.status + metrics_topic: tf.metrics + session_log: sessions.jsonl + event_log: events/{phase}_{logical_session_index}_{session_id}.jsonl +workload: + mode: bidirectional + task: bidirectional + fps: 16 + session_count: 1 + warmup_sessions: 0 + session_duration_s: 240.0 + control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json + request_extra: + chunk_size: 4 + frame_num: 957 + max_duration_seconds: 60.0 + sample_shift: 10.0 + control_mode: cam + show_control_hud: false + benchmark_metrics: true +metrics: + - offer_rtt_ms + - connected_latency_ms + - first_frame_latency_ms + - first_metadata_latency_ms + - stream_fps + - session_runtime_s + - frames_received + - control_ack_latency_ms + - control_to_next_frame_latency_ms + - runtime_creation_seconds + - chunk_compute_seconds + - chunk_compute_fps + - success_rate +limits: + active_sessions: 1 +artifacts: + config: benchmarks/telefuser_aiperf/configs/stream_lingbot_world_v2_1min.json + control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json + runner: benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh diff --git a/benchmarks/telefuser_aiperf/telefuser_aiperf/__init__.py b/benchmarks/telefuser_aiperf/telefuser_aiperf/__init__.py new file mode 100644 index 0000000..c2fec9c --- /dev/null +++ b/benchmarks/telefuser_aiperf/telefuser_aiperf/__init__.py @@ -0,0 +1,20 @@ +"""TeleFuser-owned AIPerf integration.""" + +from __future__ import annotations + +from aiperf.streaming.adapters import register_stream_adapter + +from telefuser_aiperf.adapter import TeleFuserLiveKitAdapter + + +def register_adapters(*, replace: bool = False) -> None: + """Register TeleFuser adapters in the current AIPerf process.""" + + register_stream_adapter( + "telefuser_livekit", + TeleFuserLiveKitAdapter, + replace=replace, + ) + + +__all__ = ["TeleFuserLiveKitAdapter", "register_adapters"] diff --git a/benchmarks/telefuser_aiperf/telefuser_aiperf/adapter.py b/benchmarks/telefuser_aiperf/telefuser_aiperf/adapter.py new file mode 100644 index 0000000..debf387 --- /dev/null +++ b/benchmarks/telefuser_aiperf/telefuser_aiperf/adapter.py @@ -0,0 +1,396 @@ +"""AIPerf adapter for TeleFuser LiveKit streaming sessions.""" + +from __future__ import annotations + +import asyncio +import time +from collections import deque +from collections.abc import Callable, Mapping +from pathlib import Path +from typing import Any + +import orjson +from aiperf.common.redact import redact_string +from aiperf.streaming.adapters.common import ( + StreamHttpClient, + StreamTargetMetadataMixin, +) +from aiperf.streaming.adapters.target_measurements import record_target_measurement +from aiperf.streaming.config import StreamProfileConfig +from aiperf.streaming.contracts import BenchmarkContract +from aiperf.streaming.events import StreamEventRecorder +from aiperf.streaming.models import ( + ControlEventResult, + SessionResult, + StreamSessionPlan, +) + +from telefuser_aiperf.livekit_room import ( + LiveKitRoomClient, + LiveKitRoomClientProtocol, +) +from telefuser_aiperf.payload import build_telefuser_livekit_session_body + +_CONTROL_TOPIC = "tf.control" +_STATUS_TOPIC = "tf.status" +_METRICS_TOPIC = "tf.metrics" + + +class _LiveKitSession: + def __init__( + self, + *, + adapter: TeleFuserLiveKitAdapter, + plan: StreamSessionPlan, + room: LiveKitRoomClientProtocol, + ) -> None: + self.adapter = adapter + self.plan = plan + self.room = room + self.session_id = plan.planned_session_id + self.result = SessionResult( + logical_session_index=plan.logical_session_index, + phase=plan.phase, + mode=plan.mode, + planned_session_id=plan.planned_session_id, + session_id=plan.planned_session_id, + ) + self.events = StreamEventRecorder( + artifacts_dir=adapter.artifacts_dir, + phase=plan.phase, + logical_session_index=plan.logical_session_index, + planned_session_id=plan.planned_session_id, + print_events=adapter.config.print_events, + ) + self.started_at = 0.0 + self.active_started_at: float | None = None + self.first_frame_at: float | None = None + self.last_frame_at: float | None = None + self.first_metadata_at: float | None = None + self.connected = False + self.target_ready = False + self.active_event = asyncio.Event() + self.first_frame_event = asyncio.Event() + self.done_event = asyncio.Event() + self.control_task: asyncio.Task[None] | None = None + self.pending_control_acks: deque[int] = deque() + self.pending_control_frames: deque[int] = deque() + self.control_sent_at: dict[int, float] = {} + + def _active_start(self) -> float: + if self.active_started_at is None: + raise RuntimeError("LiveKit active window has not started") + return self.active_started_at + + def _try_start_active_window(self) -> None: + if self.active_started_at is not None or not self.connected: + return + if not self.target_ready: + return + self.active_started_at = time.perf_counter() + self.active_event.set() + self.events.record("active_window_start") + if self.control_task is None and self.plan.control_trace: + self.control_task = asyncio.create_task(self._send_control_trace()) + + def _handle_video_frame(self) -> None: + now = time.perf_counter() + self.target_ready = True + self.result.frames_received += 1 + if self.first_frame_at is None: + self.first_frame_at = now + self.result.first_frame_latency_ms = (now - self.started_at) * 1000.0 + self.first_frame_event.set() + self.events.record("first_frame") + self.last_frame_at = now + self._mark_control_frame(now) + self._try_start_active_window() + + def _mark_control_frame(self, now: float) -> None: + if not self.pending_control_frames: + return + index = self.pending_control_frames.popleft() + control = self.result.control_events[index] + if control.next_frame_latency_ms is None: + control.next_frame_latency_ms = max( + (now - self.control_sent_at[index]) * 1000.0, + 0.0, + ) + + def _handle_room_event(self, event: str, payload: Mapping[str, Any]) -> None: + self.events.record(event, **dict(payload)) + if event == "disconnected": + self.done_event.set() + + def _handle_data_message( + self, + raw_message: bytes | str, + topic: str, + sender_identity: str, + ) -> None: + if topic not in {_STATUS_TOPIC, _METRICS_TOPIC}: + self.events.record("livekit_data_ignored", topic=topic) + return + now = time.perf_counter() + self.result.metadata_messages += 1 + if self.first_metadata_at is None: + self.first_metadata_at = now + self.result.first_metadata_latency_ms = (now - self.started_at) * 1000.0 + try: + payload = orjson.loads(raw_message) + except orjson.JSONDecodeError: + self.events.record("livekit_data_invalid", topic=topic) + return + if not isinstance(payload, dict): + self.events.record("livekit_data_ignored", topic=topic, reason="not_mapping") + return + if payload.get("type") == "done": + self.result.done_received = True + self.done_event.set() + self.events.record("done_message", topic=topic) + return + data = payload.get("data") if isinstance(payload.get("data"), dict) else payload + if data.get("type") == "error" or payload.get("error"): + error = data.get("error") or payload.get("error") + self.result.error = redact_string(str(error)) + self.done_event.set() + stage = data.get("stage") + if stage is not None: + self._handle_status_stage(str(stage), data, now) + self.events.record( + "livekit_data", + topic=topic, + message_type=payload.get("type"), + sender_identity=sender_identity, + source_timestamp=payload.get("timestamp"), + ) + + def _handle_status_stage( + self, + stage: str, + data: Mapping[str, Any], + now: float, + ) -> None: + self.result.status_messages += 1 + self.result.last_status_stage = stage + if stage in {"worker_running", "runtime_ready"}: + self.target_ready = True + self._try_start_active_window() + if stage in {"runtime_ready", "chunk_sent"}: + record_target_measurement( + result=self.result, + events=self.events, + stage=stage, + data=data, + ) + if not self.pending_control_acks: + return + if stage not in {"control_state", "applying_direction_control"}: + return + index = self.pending_control_acks.popleft() + control = self.result.control_events[index] + if control.ack_latency_ms is None: + control.ack_latency_ms = max( + (now - self.control_sent_at[index]) * 1000.0, + 0.0, + ) + + async def _send_control_trace(self) -> None: + active_started_at = self._active_start() + for event_index, entry in enumerate(self.plan.control_trace): + deadline = active_started_at + float(entry["delay_s"]) + await asyncio.sleep(max(deadline - time.perf_counter(), 0.0)) + message = dict(entry["message"]) + sent_at = time.perf_counter() + self.result.control_events.append( + ControlEventResult( + index=event_index, + scheduled_delay_s=float(entry["delay_s"]), + message=message, + sent_offset_s=sent_at - active_started_at, + ) + ) + self.control_sent_at[event_index] = sent_at + self.pending_control_acks.append(event_index) + self.pending_control_frames.append(event_index) + await self.room.publish_data( + message, + topic=_CONTROL_TOPIC, + reliable=True, + ) + self.events.record("control_sent", event_id=event_index) + + async def _create_and_connect(self) -> None: + create_started_at = time.perf_counter() + response = await self.adapter.http.request_json( + f"{self.plan.server_url}{self.plan.endpoints.offer_path}", + method="POST", + timeout_s=float(self.adapter.options.connect_timeout_s), + payload=build_telefuser_livekit_session_body(self.plan), + ) + self.result.offer_rtt_ms = (time.perf_counter() - create_started_at) * 1000.0 + required = ("session_id", "livekit_url", "token") + missing = [name for name in required if not isinstance(response.get(name), str)] + if missing: + raise ValueError("LiveKit session response requires string fields: " + ", ".join(missing)) + self.session_id = response["session_id"] + self.result.session_id = self.session_id + self.events.set_session_id(self.session_id) + self.events.record( + "session_created", + room=response.get("room"), + status=response.get("status"), + queue_position=response.get("queue_position"), + ) + await self.room.connect( + response["livekit_url"], + response["token"], + timeout_s=float(self.adapter.options.connect_timeout_s), + on_data=self._handle_data_message, + on_video_frame=self._handle_video_frame, + on_event=self._handle_room_event, + ) + self.connected = True + self.result.connected_latency_ms = (time.perf_counter() - self.started_at) * 1000.0 + self.events.record("connected") + self._try_start_active_window() + + async def _wait_for_media(self) -> None: + await asyncio.wait_for( + self.active_event.wait(), + timeout=float(self.adapter.options.connect_timeout_s), + ) + first_frame = asyncio.create_task(self.first_frame_event.wait()) + done = asyncio.create_task(self.done_event.wait()) + completed, pending = await asyncio.wait( + {first_frame, done}, + timeout=float(self.adapter.options.frame_timeout_s), + return_when=asyncio.FIRST_COMPLETED, + ) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + if not completed or not self.first_frame_event.is_set(): + raise TimeoutError("No LiveKit video frame received before the frame timeout") + remaining = max( + self._active_start() + float(self.plan.session_duration_s) - time.perf_counter(), + 0.0, + ) + try: + await asyncio.wait_for(self.done_event.wait(), timeout=remaining) + except asyncio.TimeoutError: + self.events.record("session_duration_elapsed") + + async def run(self) -> SessionResult: + self.started_at = time.perf_counter() + self.events.record( + "session_start", + transport="webrtc", + transport_provider="livekit", + mode=self.plan.mode, + ) + completed = False + try: + await self._create_and_connect() + await self._wait_for_media() + completed = True + except Exception as exc: # noqa: BLE001 - transport failures become results + self.result.error = redact_string(f"{type(exc).__name__}: {exc}") + self.events.record("session_error", error=self.result.error) + finally: + await self._shutdown() + if completed: + self._finalize_success() + self.result.session_runtime_s = time.perf_counter() - self.started_at + event_path = await self.events.export() + self.result.artifacts_event_file = str(event_path) + return self.result + + def _finalize_success(self) -> None: + self.result.success = self.first_frame_at is not None and self.result.error is None + if self.first_frame_at is None: + self.result.error = self.result.error or "No LiveKit video frame received" + return + if self.last_frame_at is not None and self.last_frame_at > self.first_frame_at: + self.result.stream_fps = (self.result.frames_received - 1) / (self.last_frame_at - self.first_frame_at) + + async def _shutdown(self) -> None: + if self.connected: + try: + await self.room.publish_data( + {"type": "stop"}, + topic=_CONTROL_TOPIC, + reliable=True, + ) + self.events.record("stop_sent") + except Exception as exc: # noqa: BLE001 - cleanup is best effort + self.events.record_error("stop_send_failed", exc) + await self._delete_target_session() + if self.control_task is not None and not self.control_task.done(): + self.control_task.cancel() + await asyncio.gather(self.control_task, return_exceptions=True) + try: + await self.room.disconnect() + except Exception as exc: # noqa: BLE001 - cleanup is best effort + self.events.record_error("livekit_disconnect_failed", exc) + + async def _delete_target_session(self) -> None: + template = self.plan.endpoints.delete_path_template + if template is None or self.session_id == self.plan.planned_session_id: + return + delete_path = template.format(session_id=self.session_id) + try: + await self.adapter.http.request_json( + f"{self.plan.server_url}{delete_path}", + method="DELETE", + timeout_s=float(self.adapter.options.shutdown_timeout_s), + accepted_error_statuses=(404,), + ) + self.events.record("session_delete") + except Exception as exc: # noqa: BLE001 - cleanup is best effort + self.events.record_error("session_delete_failed", exc) + + +class TeleFuserLiveKitAdapter(StreamTargetMetadataMixin): + """AIPerf adapter for WebRTC media delivered through TeleFuser LiveKit rooms.""" + + transport = "webrtc" + + def __init__( + self, + *, + contract: BenchmarkContract, + config: StreamProfileConfig, + artifacts_dir: str | Path, + room_client_factory: Callable[[], LiveKitRoomClientProtocol] = LiveKitRoomClient, + http_client: StreamHttpClient | None = None, + ) -> None: + self.contract = contract + self.config = config + self.options = config.transport + self.artifacts_dir = Path(artifacts_dir) + self.room_client_factory = room_client_factory + self.http = http_client or StreamHttpClient() + + async def check_health(self) -> None: + """Check the contract-declared target health endpoint.""" + + health_path = str(self.contract.endpoint.get("health_path", "/v1/service/health")) + await self.http.check_health( + f"{self.config.server_url}{health_path}", + timeout_s=float(self.options.connect_timeout_s), + ) + + async def run_session(self, plan: StreamSessionPlan) -> SessionResult: + """Execute one normalized plan through a LiveKit room.""" + + return await _LiveKitSession( + adapter=self, + plan=plan, + room=self.room_client_factory(), + ).run() + + async def aclose(self) -> None: + """Close adapter-owned HTTP resources.""" + + await self.http.aclose() diff --git a/benchmarks/telefuser_aiperf/telefuser_aiperf/cli.py b/benchmarks/telefuser_aiperf/telefuser_aiperf/cli.py new file mode 100644 index 0000000..3ad1f4b --- /dev/null +++ b/benchmarks/telefuser_aiperf/telefuser_aiperf/cli.py @@ -0,0 +1,19 @@ +"""AIPerf CLI bootstrap with TeleFuser adapters registered.""" + +from __future__ import annotations + + +def main() -> None: + """Register TeleFuser adapters and delegate to the AIPerf CLI.""" + + from telefuser_aiperf import register_adapters + + register_adapters() + + from aiperf.cli import app + + app() + + +if __name__ == "__main__": + main() diff --git a/benchmarks/telefuser_aiperf/telefuser_aiperf/livekit_room.py b/benchmarks/telefuser_aiperf/telefuser_aiperf/livekit_room.py new file mode 100644 index 0000000..e103782 --- /dev/null +++ b/benchmarks/telefuser_aiperf/telefuser_aiperf/livekit_room.py @@ -0,0 +1,202 @@ +"""Minimal LiveKit room client used by the TeleFuser AIPerf adapter.""" + +from __future__ import annotations + +import asyncio +import contextlib +import ipaddress +import os +from collections.abc import Callable, Mapping +from typing import Any, Protocol +from urllib.parse import urlsplit + +import orjson + +LiveKitDataHandler = Callable[[bytes | str, str, str], None] +LiveKitFrameHandler = Callable[[], None] +LiveKitEventHandler = Callable[[str, Mapping[str, Any]], None] + +_PROXY_ENV_NAMES = ( + "http_proxy", + "https_proxy", + "all_proxy", + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", +) + + +def _disable_proxy_for_loopback(url: str) -> bool: + host = urlsplit(url).hostname + if host is None: + return False + try: + is_loopback = host.lower() == "localhost" or ipaddress.ip_address(host).is_loopback + except ValueError: + is_loopback = False + if not is_loopback: + return False + for name in _PROXY_ENV_NAMES: + os.environ.pop(name, None) + return True + + +class LiveKitRoomClientProtocol(Protocol): + """Minimal LiveKit subscriber surface required by the stream adapter.""" + + async def connect( + self, + url: str, + token: str, + *, + timeout_s: float, + on_data: LiveKitDataHandler, + on_video_frame: LiveKitFrameHandler, + on_event: LiveKitEventHandler, + ) -> None: ... + + async def publish_data( + self, + payload: dict[str, Any], + *, + topic: str, + reliable: bool, + ) -> None: ... + + async def disconnect(self) -> None: ... + + +class LiveKitDependencyError(RuntimeError): + """Raised when the LiveKit SDK is unavailable.""" + + +class LiveKitRoomClient: + """SDK-backed LiveKit room subscriber used by the TeleFuser adapter.""" + + def __init__(self) -> None: + self._rtc = self._load_rtc() + self._room: Any | None = None + self._video_streams: list[Any] = [] + self._video_tasks: list[asyncio.Task[None]] = [] + self._on_video_frame: LiveKitFrameHandler | None = None + self._on_event: LiveKitEventHandler | None = None + + async def connect( + self, + url: str, + token: str, + *, + timeout_s: float, + on_data: LiveKitDataHandler, + on_video_frame: LiveKitFrameHandler, + on_event: LiveKitEventHandler, + ) -> None: + self._on_video_frame = on_video_frame + self._on_event = on_event + room = self._rtc.Room() + self._room = room + + @room.on("data_received") + def _on_data_received(packet: Any) -> None: + participant = getattr(packet, "participant", None) + identity = getattr(participant, "identity", "") if participant is not None else "" + on_data(packet.data, packet.topic or "", identity) + + @room.on("track_subscribed") + def _on_track_subscribed( + track: Any, + publication: Any, + participant: Any, + ) -> None: + if track.kind != self._rtc.TrackKind.KIND_VIDEO: + return + stream = self._rtc.VideoStream(track) + self._video_streams.append(stream) + self._video_tasks.append(asyncio.create_task(self._consume_video(stream))) + on_event( + "remote_track", + { + "kind": "video", + "participant_identity": getattr(participant, "identity", ""), + "track_sid": getattr(publication, "sid", ""), + }, + ) + + @room.on("reconnecting") + def _on_reconnecting() -> None: + on_event("reconnecting", {}) + + @room.on("reconnected") + def _on_reconnected() -> None: + on_event("reconnected", {}) + + @room.on("disconnected") + def _on_disconnected(reason: Any) -> None: + on_event("disconnected", {"reason": str(reason)}) + + options = self._rtc.RoomOptions( + auto_subscribe=True, + connect_timeout=timeout_s, + ) + _disable_proxy_for_loopback(url) + await room.connect(url, token, options) + + async def _consume_video(self, stream: Any) -> None: + try: + async for _ in stream: + if self._on_video_frame is not None: + self._on_video_frame() + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - remote media termination is data + if self._on_event is not None: + self._on_event( + "video_stream_ended", + {"error": f"{type(exc).__name__}: {exc}"}, + ) + + async def publish_data( + self, + payload: dict[str, Any], + *, + topic: str, + reliable: bool, + ) -> None: + room = self._require_room() + await room.local_participant.publish_data( + orjson.dumps(payload), + topic=topic, + reliable=reliable, + ) + + async def disconnect(self) -> None: + room = self._room + if room is None: + return + try: + for stream in self._video_streams: + with contextlib.suppress(Exception): + await asyncio.wait_for(stream.aclose(), timeout=2.0) + for task in self._video_tasks: + if not task.done(): + task.cancel() + if self._video_tasks: + await asyncio.gather(*self._video_tasks, return_exceptions=True) + await room.disconnect() + finally: + self._room = None + self._video_streams.clear() + self._video_tasks.clear() + + @staticmethod + def _load_rtc() -> Any: + try: + from livekit import rtc + except ModuleNotFoundError as exc: + raise LiveKitDependencyError("The TeleFuser AIPerf adapter requires the 'livekit' package") from exc + return rtc + + def _require_room(self) -> Any: + if self._room is None: + raise RuntimeError("LiveKit room is not connected") + return self._room diff --git a/benchmarks/telefuser_aiperf/telefuser_aiperf/payload.py b/benchmarks/telefuser_aiperf/telefuser_aiperf/payload.py new file mode 100644 index 0000000..02cf65f --- /dev/null +++ b/benchmarks/telefuser_aiperf/telefuser_aiperf/payload.py @@ -0,0 +1,29 @@ +"""TeleFuser LiveKit request payload construction.""" + +from __future__ import annotations + +from typing import Any + +from aiperf.streaming.models import StreamSessionPlan + + +def build_telefuser_livekit_session_body( + plan: StreamSessionPlan, +) -> dict[str, Any]: + """Build a TeleFuser LiveKit session request from a normalized plan.""" + + options = dict(plan.request_extra) + reserved = {"identity", "role", "prompt", "image_path", "config", "session_id"} + conflicts = sorted(reserved.intersection(options)) + if conflicts: + raise ValueError("LiveKit request_extra cannot override protocol fields: " + ", ".join(conflicts)) + config = {"task": plan.task, "fps": plan.fps, **options} + body: dict[str, Any] = { + "identity": f"aiperf-{plan.planned_session_id}", + "role": "controller", + "prompt": plan.prompt, + "config": config, + } + if plan.image_path: + body["image_path"] = plan.image_path + return body diff --git a/benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py b/benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py new file mode 100644 index 0000000..5766e93 --- /dev/null +++ b/benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py @@ -0,0 +1,354 @@ +from __future__ import annotations + +import os +from collections.abc import Callable, Mapping +from pathlib import Path +from typing import Any + +import orjson +import pytest +from aiperf.streaming.adapters import create_stream_adapter +from aiperf.streaming.config import StreamProfileConfig +from aiperf.streaming.contracts import BenchmarkContract +from aiperf.streaming.models import StreamEndpointPaths, StreamSessionPlan +from telefuser_aiperf import register_adapters +from telefuser_aiperf.adapter import TeleFuserLiveKitAdapter +from telefuser_aiperf.livekit_room import _disable_proxy_for_loopback +from telefuser_aiperf.payload import build_telefuser_livekit_session_body + + +class _FakeHttpClient: + def __init__(self) -> None: + self.health_urls: list[str] = [] + self.requests: list[tuple[str, str]] = [] + self.payloads: list[dict[str, Any] | None] = [] + + async def check_health(self, url: str, *, timeout_s: float) -> None: + self.health_urls.append(url) + + async def request_json( + self, + url: str, + *, + method: str, + timeout_s: float, + payload: dict[str, Any] | None = None, + accepted_error_statuses: tuple[int, ...] = (), + ) -> dict[str, Any]: + self.requests.append((method, url)) + self.payloads.append(payload) + if method == "POST": + return { + "session_id": "livekit-session", + "room": "tf-world-livekit-session", + "livekit_url": "ws://127.0.0.1:7880", + "token": "test-token", + "status": "assigned", + } + return {} + + async def aclose(self) -> None: + return None + + +class _FakeLiveKitRoom: + instances: list[_FakeLiveKitRoom] = [] + + def __init__(self) -> None: + self.connected: tuple[str, str] | None = None + self.published: list[tuple[dict[str, Any], str, bool]] = [] + self.disconnected = False + self.__class__.instances.append(self) + + async def connect( + self, + url: str, + token: str, + *, + timeout_s: float, + on_data: Callable[[bytes | str, str, str], None], + on_video_frame: Callable[[], None], + on_event: Callable[[str, Mapping[str, Any]], None], + ) -> None: + self.connected = (url, token) + on_event("remote_track", {"kind": "video"}) + on_data( + orjson.dumps( + { + "type": "chunk", + "data": {"type": "status", "stage": "worker_running"}, + } + ), + "tf.status", + "telefuser-worker-0", + ) + on_data( + orjson.dumps( + { + "type": "chunk", + "data": { + "stage": "runtime_ready", + "measurement": {"seconds": 2.0, "memory": []}, + "runtime": {"cache": "enabled"}, + }, + } + ), + "tf.metrics", + "telefuser-worker-0", + ) + on_data( + orjson.dumps( + { + "type": "chunk", + "data": { + "stage": "chunk_sent", + "measurement": { + "index": 0, + "frames": 3, + "compute_seconds": 0.5, + "encode_seconds": 0.1, + "memory": [], + }, + }, + } + ), + "tf.metrics", + "telefuser-worker-0", + ) + on_video_frame() + on_video_frame() + on_data( + orjson.dumps({"type": "done", "session_id": "livekit-session"}), + "tf.status", + "telefuser-worker-0", + ) + + async def publish_data( + self, + payload: dict[str, Any], + *, + topic: str, + reliable: bool, + ) -> None: + self.published.append((payload, topic, reliable)) + + async def disconnect(self) -> None: + self.disconnected = True + + +class _ControlLiveKitRoom(_FakeLiveKitRoom): + async def connect( + self, + url: str, + token: str, + *, + timeout_s: float, + on_data: Callable[[bytes | str, str, str], None], + on_video_frame: Callable[[], None], + on_event: Callable[[str, Mapping[str, Any]], None], + ) -> None: + self.connected = (url, token) + self._on_data = on_data + self._on_video_frame = on_video_frame + on_data( + orjson.dumps({"type": "chunk", "data": {"stage": "worker_running"}}), + "tf.status", + "telefuser-worker-0", + ) + on_video_frame() + + async def publish_data( + self, + payload: dict[str, Any], + *, + topic: str, + reliable: bool, + ) -> None: + await super().publish_data(payload, topic=topic, reliable=reliable) + if payload.get("type") == "stop": + return + self._on_data( + orjson.dumps({"type": "chunk", "data": {"stage": "control_state"}}), + "tf.status", + "telefuser-worker-0", + ) + self._on_video_frame() + self._on_data( + orjson.dumps({"type": "done", "session_id": "livekit-session"}), + "tf.status", + "telefuser-worker-0", + ) + + +def _contract() -> BenchmarkContract: + return BenchmarkContract( + contract_version="v1", + name="adapter-test", + mode="stream_world", + implementation="telefuser", + model_family="world", + model="world-model", + supported_tasks=["bidirectional"], + transport="webrtc", + adapter="telefuser_livekit", + transport_provider="livekit", + endpoint={ + "health_path": "/health", + "offer_path": "/stream", + "delete_path_template": "/sessions/{session_id}", + }, + request_encoding={"format": "json"}, + result_delivery={"media": "livekit_video_track"}, + workload={"size": "320x180"}, + metrics=["first_frame_latency_ms"], + artifacts={"config": "stream.json"}, + ) + + +def _config(tmp_path: Path) -> StreamProfileConfig: + return StreamProfileConfig.model_validate( + { + "contract": "contract.yaml", + "server_url": "http://127.0.0.1:30000", + "prompt": "walk forward", + "artifacts_dir": str(tmp_path), + "transport": { + "connect_timeout_s": 0.5, + "message_timeout_s": 0.5, + "frame_timeout_s": 0.5, + "ice_gather_timeout_s": 0.5, + "shutdown_timeout_s": 0.5, + }, + } + ) + + +def _plan() -> StreamSessionPlan: + return StreamSessionPlan( + logical_session_index=0, + phase="profiling", + planned_session_id="planned", + server_url="http://127.0.0.1:30000", + endpoints=StreamEndpointPaths( + health_path="/health", + offer_path="/stream", + delete_path_template="/sessions/{session_id}", + ), + mode="bidirectional", + task="bidirectional", + prompt="walk forward", + fps=16, + session_duration_s=0.03, + request_extra={"benchmark_metrics": True}, + ) + + +def test_registration_uses_unmodified_aiperf_registry(tmp_path: Path) -> None: + register_adapters(replace=True) + + adapter = create_stream_adapter( + contract=_contract(), + config=_config(tmp_path), + artifacts_dir=tmp_path, + ) + + assert isinstance(adapter, TeleFuserLiveKitAdapter) + assert adapter.transport == "webrtc" + + +def test_payload_rejects_protocol_field_override() -> None: + plan = _plan().model_copy(update={"request_extra": {"identity": "override"}}) + + with pytest.raises(ValueError, match="identity"): + build_telefuser_livekit_session_body(plan) + + +def test_proxy_bypass_is_limited_to_loopback_urls(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("http_proxy", "http://proxy.example:3128") + monkeypatch.setenv("HTTPS_PROXY", "http://proxy.example:3128") + + assert _disable_proxy_for_loopback("wss://livekit.example.test") is False + assert "http_proxy" in os.environ + assert _disable_proxy_for_loopback("ws://127.0.0.1:7880") is True + assert "http_proxy" not in os.environ + assert "HTTPS_PROXY" not in os.environ + + +@pytest.mark.asyncio +async def test_adapter_normalizes_room_events_and_deletes_target( + tmp_path: Path, +) -> None: + _FakeLiveKitRoom.instances.clear() + http = _FakeHttpClient() + adapter = TeleFuserLiveKitAdapter( + contract=_contract(), + config=_config(tmp_path), + artifacts_dir=tmp_path, + room_client_factory=_FakeLiveKitRoom, + http_client=http, + ) + + await adapter.check_health() + result = await adapter.run_session(_plan()) + + room = _FakeLiveKitRoom.instances[0] + assert result.success is True + assert result.session_id == "livekit-session" + assert result.frames_received == 2 + assert result.done_received is True + assert result.phase_measurements[0].name == "runtime_creation" + assert result.chunk_measurements[0].compute_seconds == 0.5 + assert result.runtime_metadata == {"cache": "enabled"} + assert room.connected == ("ws://127.0.0.1:7880", "test-token") + assert room.published[-1] == ({"type": "stop"}, "tf.control", True) + assert room.disconnected is True + assert http.payloads[0] == { + "identity": "aiperf-planned", + "role": "controller", + "prompt": "walk forward", + "config": { + "task": "bidirectional", + "fps": 16, + "benchmark_metrics": True, + }, + } + assert ( + "DELETE", + "http://127.0.0.1:30000/sessions/livekit-session", + ) in http.requests + assert Path(result.artifacts_event_file or "").is_file() + + +@pytest.mark.asyncio +async def test_adapter_maps_control_ack_and_next_frame(tmp_path: Path) -> None: + _ControlLiveKitRoom.instances.clear() + adapter = TeleFuserLiveKitAdapter( + contract=_contract(), + config=_config(tmp_path), + artifacts_dir=tmp_path, + room_client_factory=_ControlLiveKitRoom, + http_client=_FakeHttpClient(), + ) + plan = _plan().model_copy( + update={ + "control_trace": [ + { + "delay_s": 0.0, + "message": { + "type": "control", + "key": "ArrowUp", + "action": "press", + }, + } + ] + } + ) + + result = await adapter.run_session(plan) + + assert result.success is True + assert len(result.control_events) == 1 + assert result.control_events[0].ack_latency_ms is not None + assert result.control_events[0].next_frame_latency_ms is not None + room = _ControlLiveKitRoom.instances[0] + assert room.published[0][1:] == ("tf.control", True) diff --git a/docs/en/benchmark_aiperf.md b/docs/en/benchmark_aiperf.md index 9c9f1c9..b04d320 100644 --- a/docs/en/benchmark_aiperf.md +++ b/docs/en/benchmark_aiperf.md @@ -1,226 +1,127 @@ # TeleFuser and AIPerf -TeleFuser exposes raw target-side facts; AIPerf owns workload execution, aggregation, -resource collection, artifacts, GreptimeDB history, and visualization. This separation -keeps the same benchmark and dashboard reusable across TeleFuser, SGLang-Diffusion, and -future targets. +TeleFuser exposes raw target-side facts; AIPerf owns workload execution, aggregation, resource collection, artifacts, +GreptimeDB history, and visualization. The checked-in integration covers batch video generation through the +OpenAI-compatible `/v1/videos` API and LingBot streaming through LiveKit. -The included assets cover: +AIPerf's stream runner and result schema are transport-neutral. The LiveKit adapter is maintained by +TeleFuser, loads from source at process startup, and produces AIPerf's standard session results. The contract records WebRTC as +the media transport and LiveKit as its provider, preserving the SFU topology without adding LiveKit code to AIPerf. -- Wan2.1 image-to-video through the OpenAI-compatible `/v1/videos` API; -- LingBot-World-Fast sessions through WebRTC and DataChannel; -- a LingBot SGLang-Diffusion baseline through WebSocket and MessagePack. +For installation, workload configs, launch commands, history setup, and focused tests, use the canonical +[`benchmarks/telefuser_aiperf/README.md`](https://github.com/Tele-AI/TeleFuser/tree/main/benchmarks/telefuser_aiperf#readme). AIPerf is installed from a pinned +Git commit with `pip`; no retained AIPerf checkout or adapter `pyproject.toml` is required. -## Repository layout +## Quick start -```text -benchmarks/ -├── telefuser_aiperf/ # TeleFuser contracts, configs, data, launchers -├── baseline/sglang_lingbot_stream/ # Stream baseline -└── aiperf/ # Ignored external AIPerf checkout -``` - -The AIPerf implementation is not vendored into TeleFuser. The setup script always uses -`/benchmarks/aiperf`; neither the setup script nor the launchers accept a -checkout-path override. Install -[uv](https://docs.astral.sh/uv/getting-started/installation/), then run this once from -the TeleFuser repository root: - -```bash -bash scripts/setup_aiperf_repo.sh -``` - -The script clones AIPerf, creates its isolated runtime environment with WebRTC support, -and creates `/artifacts` for benchmark output and History imports. The -dashboard is bundled, so runtime users do not need Node.js or a separate frontend -process. Pin a commit for reproducible runs: - -```bash -AIPERF_REF= bash scripts/setup_aiperf_repo.sh -``` - -`AIPERF_REPO_URL`, `AIPERF_BRANCH`, and `AIPERF_REF` may select the source and revision, -but never change the checkout location. - -## Batch video - -Start the fixed Wan2.1 I2V target: - -```bash -telefuser serve \ - examples/wan_video/wan21_14b_image_to_video_480p_service.py \ - --port 8000 \ - --task i2v -``` - -Run a smoke profile or the fixed comparison workload: - -```bash -bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh - -bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ - benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml -``` - -The launcher checks `/v1/service/health` before profiling. Common overrides include -`TELEFUSER_AIPERF_URL`, `TELEFUSER_AIPERF_CONCURRENCY`, -`TELEFUSER_AIPERF_REQUESTS`, `TELEFUSER_AIPERF_SIZE`, and -`TELEFUSER_AIPERF_SECONDS`. - -## LingBot stream - -Start TeleFuser: - -```bash -telefuser stream-serve \ - examples/lingbot/lingbot_world_fast_image_to_video_h100.py \ - -p 8088 \ - --skip-validation -``` - -Then run: - -```bash -bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh - -bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \ - benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json -``` - -The stream config enables `benchmark_metrics`. TeleFuser then reports synchronized raw -facts for runtime creation, actor-graph chunk compute, cache geometry, and environment -identity. Allocator peaks are omitted because generation runs in child actors and the -service-process allocator cannot represent the complete graph; active AIPerf resource -telemetry supplies process-tree GPU-memory curves instead. The native WebRTC path does -not report a separate payload encoding duration because encoding happens after the target -chunk fact. AIPerf computes warmup-aware summaries and keeps client delivery separate -from target compute. - -The shared timed control trace is stored at -`benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json`. - -## SGLang-Diffusion baseline - -Use a compatible, version-pinned `sgl-project/sglang` environment. The TeleFuser tree -does not patch SGLang modules at import time. +From the TeleFuser repository root, install the streaming-capable AIPerf Git commit into its isolated environment: ```bash -bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh -bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh +bash scripts/setup_aiperf.sh ``` -The baseline uses the same prompt, first frame, FPS target, session window, and control -trace. The adapter translates only transport semantics. For a performance comparison, -record the exact SGLang commit and model revision, use GPU-resident speed mode, and keep -offload and fallback settings identical. An OOM is a result for that configuration; do -not replace it with a mock or offloaded result under the same label. - -Both documented launch commands default to one GPU. Override both targets explicitly -when comparing another accelerator count. - -## Configs - -| Config | Purpose | -|---|---| -| `video_generation_quick.yaml` | Batch connectivity and latency smoke test | -| `video_generation_e2e.yaml` | Batch warmup, trace, records, and server metrics | -| `video_generation_rate.yaml` | Poisson-arrival Batch load | -| `video_generation_wan21_i2v_480p_compare.yaml` | Fixed Wan I2V comparison | -| `stream_lingbot_world_fast_quick.json` | Bounded Stream smoke test | -| `stream_lingbot_world_fast_compare.json` | Fixed LingBot Stream comparison | - -SGLang equivalents are under `benchmarks/baseline/sglang_lingbot_stream/configs`. - -## Metric interpretation - -The most important distinction is scope: - -| Metric | Meaning | -|---|---| -| `stream_fps` | Frames received by the client divided by client session time | -| `chunk_compute_fps` | Frames divided by compute time for one target chunk | -| `chunk_compute_fps_weighted` | `sum(frames) / sum(compute_seconds)` after warmup exclusion | - -AIPerf presents metrics under five stable dimensions: delivery, latency, throughput, -target execution, and resources. Implementation-specific fields remain raw evidence and -map into these canonical leaves; they do not become separate top-level metrics. - -## Active resource history - -Docker provides the shortest persistent GreptimeDB setup: +Start a local LiveKit development server in terminal 1: ```bash -docker volume create aiperf-greptime-data -docker run -d --name aiperf-greptime --restart unless-stopped \ - -p 127.0.0.1:4000:4000 \ - -v aiperf-greptime-data:/greptimedb_data \ - greptime/greptimedb:latest \ - standalone start \ - --http-addr 0.0.0.0:4000 \ - --data-home /greptimedb_data +livekit-server --dev --bind 127.0.0.1 ``` -Pin the image tag or digest for production. The named volume keeps history across -container restarts. Then start the bundled AIPerf API and frontend from the TeleFuser -repository root; the `artifacts` root matches the benchmark launchers: +Start the four-GPU LingBot-World v2 target in terminal 2, replacing the model path: ```bash -uv run --frozen --no-dev --project benchmarks/aiperf aiperf history serve \ - --greptime-url http://127.0.0.1:4000 \ - --greptime-database public \ - --artifact-root artifacts \ - --host 127.0.0.1 \ - --port 8095 +env -u http_proxy -u https_proxy -u HTTP_PROXY -u HTTPS_PROXY -u all_proxy -u ALL_PROXY \ + TF_MODEL_ZOO_PATH=/path/to/model_zoo \ + CUDA_VISIBLE_DEVICES=0,1,2,3 \ + telefuser stream-serve examples/lingbot/lingbot_world_v2_image_to_video_h100.py \ + --livekit-url ws://127.0.0.1:7880 \ + --livekit-api-key devkey \ + --livekit-api-secret secret \ + --num-workers 1 \ + --worker-gpu-map 0,1,2,3 \ + --port 8088 \ + --skip-validation ``` -Verify the stack: +Wait for `"ready":true`, `"workers_idle":1`, and `"workers_failed":0`: ```bash -curl --fail http://127.0.0.1:8095/api/v1/history/health -curl --fail -X POST 'http://127.0.0.1:4000/v1/sql?db=public' \ - --data-urlencode 'sql=SELECT 1 AS ready' +curl --noproxy '*' --fail --silent --show-error \ + http://127.0.0.1:8088/v1/service/health ``` -Enable active collection on the target host: +An idle service reports `"livekit_connected":false`; that is expected. Run the benchmark in terminal 3: ```bash -export AIPERF_HISTORY_URL=http://:8095 -export AIPERF_RESOURCE_TARGET_PID= - bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh ``` -The agent recursively observes the target process tree. It samples every second, -uploads every 15 seconds, timestamps samples at the source, and flushes at termination. -It reports process, container-when-detectable, and machine facts for CPU, memory, GPU, -VRAM, Ethernet, and RDMA. Capacity is kept separate from usage. - -GreptimeDB is mandatory for History and active reporting. Startup, query, or final-flush -failure is surfaced; there is no SQLite, in-memory, or direct-file query fallback. - -Open `http://127.0.0.1:8095/` for the Chinese desktop dashboard. It supports two run -groups, canonical metric-tree selection, aggregate curves, resource timelines, and -cross-run comparison. - -For a remote benchmark host, keep the default loopback bind and forward it securely: - -```bash -ssh -L 8095:127.0.0.1:8095 user@benchmark-host -``` - -## Artifacts and reproducibility - -Batch and stream launchers write timestamped artifacts below `artifacts/`. Stream -artifacts include summaries, session and event JSONL, target metadata, normalized -metrics, and a standalone HTML report. - -Every performance result should retain: - -- TeleFuser or SGLang commit and model revision; -- accelerator model/count, driver, CUDA, PyTorch, and dtype; -- workload config and control trace; -- warmup policy and successful/failed session counts; -- offload, cache, attention, and fallback settings. - -See the Chinese [benchmark design](/TeleFuser/zh/benchmark_aiperf_design/) for protocol and -ownership details. +The request contains 59.75 seconds of media, but the configured AIPerf active window is 240 seconds. Allow about four +minutes for the command to finish. Success is `Stream profile sessions: 1/1 succeeded`; reports are created under +`artifacts/telefuser_aiperf/stream_lingbot_v2_1min/`. See the canonical README for model file layout, manual Python +environment selection, history setup, and troubleshooting. + +## Ownership and metric semantics + +| Component | Owner | Responsibility | +|---|---|---| +| TeleFuser runtime | TeleFuser | Emit synchronized phase, chunk, runtime, cache, and environment facts | +| Batch target adapter | AIPerf | Convert `/v1/videos` HTTP events into the standard request timeline | +| LiveKit source adapter | TeleFuser | Convert room, track, status, metrics, and control events into session results | +| Aggregation and history | AIPerf | Apply warmup, percentiles, throughput, artifacts, GreptimeDB, and visualization | +| Contracts and workloads | TeleFuser | Fix target capabilities, inputs, settings, and reproducible launch commands | + +Target facts follow these rules: + +- Durations use a monotonic clock; cross-process samples also retain source UTC timestamps. +- CUDA phase boundaries synchronize the measured target device. +- Values are finite and non-negative; unavailable values are omitted or `null`, never fabricated as zero. +- Memory uses bytes in the raw protocol and is converted only for display. +- The target does not exclude warmup, calculate percentiles, or produce cross-run conclusions. + +| Scope | Examples | Aggregation rule | +|---|---|---| +| Event | Frame or response arrival | Preserve the event timeline | +| Request/session | First output, session latency | Calculate independently for each request or session | +| Run | Success rate, throughput, percentiles | Aggregate after AIPerf excludes warmup | + +Client delivery, target pipeline residence, target phase time, and resource utilization remain separate dimensions. +Fields without equivalent semantics remain private or unavailable instead of being forced into a common metric. + +## Validated one-minute LingBot-World v2 replay + +The `stream_lingbot_world_v2_1min.json` workload was validated on 2026-07-28 with four H100 80 GB GPUs, BF16 DiT, +FP32 VAE, SageAttention SM90, `torch.compile` enabled, FSDP disabled, `chunk_size=4`, and 16 FPS. A 60-second request +is truncated to 60 complete latent chunks: 957 output frames representing 59.75 seconds of media. LingBot-World v2 +used its fixed `local_attn_size=18` and `sink_size=6` window, so the 240 latent-frame request retained a fixed +27,144-token KV capacity rather than a duration-sized global KV cache. + +| Measurement | Result | +|---|---:| +| Successful sessions | 1 / 1 | +| Target chunks / generated frames | 60 / 957 | +| Client frames received | 947 | +| Target steady chunks / frames | 59 / 944 | +| Configured session runtime | 242.255 s | +| First-frame latency | 6,252.773 ms | +| Client delivery rate (`stream_fps`) | 9.397 FPS | +| Weighted steady chunk compute rate | 4.708 FPS | +| Chunk pipeline residence, mean / p99 | 3.399 / 3.901 s | + +The target emitted all 957 frames and the LiveKit client received 947. AIPerf excluded the first target chunk from +steady-state compute aggregation, leaving 944 generated frames across 59 chunks. The 242-second session runtime is +the configured 240-second active-window limit plus connection overhead; it is not the generation time of the +59.75-second media payload. + +Chunk pipeline residence spans actor admission through output and includes time shared by overlapping encode, DiT, +and decode work. It is therefore expected to exceed adjacent delivery intervals; the report's 4.708 weighted +steady chunk compute FPS must not be read as client stream throughput. The Git-installed AIPerf run exited with code +0. The replay artifact is +`artifacts/telefuser_aiperf/stream_lingbot_v2_1min/20260728_083948_fc4344ba/stream_report.html`. + +## Reproducibility + +Every result should retain the TeleFuser commit and AIPerf package version, model revision, accelerator model/count, +driver, CUDA, PyTorch, dtype, complete workload config, warmup rule, success/failure counts, and +offload/cache/attention settings. +Use the dated validation above as one-run evidence, not a universal performance guarantee. Ongoing comparisons belong +in GreptimeDB and replayable artifacts. diff --git a/docs/en/index.md b/docs/en/index.md index db389e3..451adfd 100644 --- a/docs/en/index.md +++ b/docs/en/index.md @@ -36,7 +36,7 @@ Compile-aware ops with eager CUDA Triton kernels and PyTorch native fallbacks.
**Streaming Service** -FastAPI batch serving plus WebRTC media tracks and DataChannel control. +FastAPI batch serving and LiveKit-backed rooms for server-push and resilient interactive WebRTC.
**Feature Cache** @@ -56,8 +56,8 @@ Reusable stages, model configs, schedulers, and pipeline orchestration. | Model | Tasks | Description | |-------|-------|-------------| -| LingBot-World v2 | Bidirectional streaming | Camera-controlled interactive world model via WebRTC | -| LingBot-World-Fast | Bidirectional streaming | Legacy/causal-fast interactive world model via WebRTC DataChannel | +| LingBot-World v2 | Bidirectional streaming | Camera-controlled interactive world model via LiveKit | +| LingBot-World-Fast | Bidirectional streaming | Legacy/causal-fast model via LiveKit reliable data messages | ### Video Generation @@ -88,17 +88,20 @@ pip install telefuser # Batch serving telefuser serve /path/to/pipeline.py --port 8000 -# Stream serving (WebRTC support is included in the default install) -telefuser stream-serve examples/lingbot/lingbot_world_fast_image_to_video_h100.py -p 8088 +# LiveKit-backed streaming (Python SDK included in the base install) +telefuser stream-serve examples/lingbot/lingbot_world_fast_image_to_video_h100.py \ + --livekit-url ws://127.0.0.1:7880 \ + --livekit-api-key devkey --livekit-api-secret secret \ + -p 8088 ``` ## Documentation Sections