diff --git a/docs/mkdocs/hooks/generate_api_readme.py b/docs/mkdocs/hooks/generate_api_readme.py index 3214abd56..e09c6e1ea 100644 --- a/docs/mkdocs/hooks/generate_api_readme.py +++ b/docs/mkdocs/hooks/generate_api_readme.py @@ -144,6 +144,7 @@ def scan_package(package_name: str = "vllm_omni") -> dict[str, list[str]]: "vllm_omni.entrypoints.async_diffusion", "vllm_omni.entrypoints.openai", "vllm_omni.model_executor.models.voxtral_tts.configuration_voxtral_tts", + "vllm_omni.experimental", # optional serving deps not installed in docs build ] if any(module_path.startswith(prefix) for prefix in excluded_prefixes): continue diff --git a/examples/online_serving/joyvl_interaction/bridges/asr_bridge.py b/examples/online_serving/joyvl_interaction/bridges/asr_bridge.py new file mode 100644 index 000000000..042e5808f --- /dev/null +++ b/examples/online_serving/joyvl_interaction/bridges/asr_bridge.py @@ -0,0 +1,131 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import argparse +import base64 +import io +import json +import struct +import wave + +import aiohttp +from aiohttp import web + +_HEADER = struct.Struct(">iii") +_SAMPLE_RATE = 16000 +_DEFAULT_PROMPT = "Transcribe the audio." +_ASR_TEXT_TAG = "" + + +def _clean_transcript(raw: str) -> str: + """Qwen3-ASR emits ``language ``; keep the transcript.""" + text = raw or "" + marker = text.rfind(_ASR_TEXT_TAG) + if marker != -1: + text = text[marker + len(_ASR_TEXT_TAG) :] + return text.replace("", "").strip() + + +def _pcm_to_wav_b64(pcm: bytes, sample_rate: int) -> str: + buf = io.BytesIO() + with wave.open(buf, "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes(pcm) + return base64.b64encode(buf.getvalue()).decode() + + +def _result(text: str, *, final: bool) -> dict: + return { + "asr_response": { + "event_type": "IS_FINAL" if final else "IS_PARTIAL", + "recognition_result": {"hypothesis": [{"text": text, "confidence": None}]}, + }, + "mid": "", + "code": 0, + "msg": "ok", + } + + +async def _transcribe(cfg: dict, pcm: bytes) -> str: + if not pcm: + return "" + wav_b64 = _pcm_to_wav_b64(pcm, cfg["sample_rate"]) + body = { + "model": cfg["model"], + "messages": [ + { + "role": "user", + "content": [ + {"type": "input_audio", "input_audio": {"data": wav_b64, "format": "wav"}}, + {"type": "text", "text": cfg["prompt"]}, + ], + } + ], + "temperature": 0.0, + "max_tokens": cfg["max_tokens"], + } + url = cfg["backend_url"].rstrip("/") + "/v1/chat/completions" + async with aiohttp.ClientSession() as session: + async with session.post(url, json=body) as resp: + payload = await resp.json() + choices = payload.get("choices") or [] + if not choices: + return "" + return _clean_transcript(choices[0].get("message", {}).get("content") or "") + + +async def _handle(request: web.Request) -> web.WebSocketResponse: + ws = web.WebSocketResponse(max_msg_size=0) + await ws.prepare(request) + cfg = request.app["cfg"] + pcm = bytearray() + try: + async for msg in ws: + if msg.type == aiohttp.WSMsgType.BINARY and len(msg.data) >= _HEADER.size: + seqid, _, _ = _HEADER.unpack(msg.data[: _HEADER.size]) + pcm.extend(msg.data[_HEADER.size :]) + if seqid < 0: + text = await _transcribe(cfg, bytes(pcm)) + await ws.send_str(json.dumps(_result(text, final=True), ensure_ascii=False)) + pcm.clear() + elif msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): + break + finally: + if not ws.closed: + await ws.close() + return ws + + +def create_app(backend_url: str, model: str, sample_rate: int, prompt: str, max_tokens: int) -> web.Application: + app = web.Application() + app["cfg"] = { + "backend_url": backend_url, + "model": model, + "sample_rate": sample_rate, + "prompt": prompt, + "max_tokens": max_tokens, + } + app.router.add_get("/v1/asr", _handle) + return app + + +def main() -> None: + parser = argparse.ArgumentParser(description="JoyVL webui <-> Qwen3-ASR (chat input_audio) bridge") + parser.add_argument("--host", default="0.0.0.0") + parser.add_argument("--port", type=int, default=8093) + parser.add_argument("--backend-url", default="http://127.0.0.1:8094") + parser.add_argument("--model", default="Qwen/Qwen3-ASR-1.7B") + parser.add_argument("--sample-rate", type=int, default=_SAMPLE_RATE) + parser.add_argument("--prompt", default=_DEFAULT_PROMPT) + parser.add_argument("--max-tokens", type=int, default=256) + args = parser.parse_args() + app = create_app(args.backend_url, args.model, args.sample_rate, args.prompt, args.max_tokens) + web.run_app(app, host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/examples/online_serving/joyvl_interaction/bridges/tts_bridge.py b/examples/online_serving/joyvl_interaction/bridges/tts_bridge.py new file mode 100644 index 000000000..d6567081a --- /dev/null +++ b/examples/online_serving/joyvl_interaction/bridges/tts_bridge.py @@ -0,0 +1,117 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging + +import aiohttp +from aiohttp import web + +try: + import cn2an # optional: Arabic-digit -> Chinese number normalization +except ImportError: + cn2an = None + +logger = logging.getLogger("tts_bridge") + + +def _normalize(text: str) -> str: + """Spell out Arabic digits in Chinese so short counting outputs synthesize + intelligibly: bare "1次" garbles on Qwen3-TTS (and others), while "一次" + is clean. ``cn2an.transform`` rewrites digits in place ("第3次"->"第三次") + and is pure-Python (~microseconds). No-op if cn2an is unavailable.""" + if not cn2an or not text: + return text + try: + return cn2an.transform(text, "an2cn") + except Exception as err: # never let TN break synthesis + logger.debug("cn2an normalize skipped: %s", err) + return text + + +async def _pump_backend_to_front(back: aiohttp.ClientWebSocketResponse, front: web.WebSocketResponse) -> None: + async for msg in back: + if msg.type == aiohttp.WSMsgType.BINARY: + await front.send_bytes(msg.data) + elif msg.type == aiohttp.WSMsgType.TEXT: + event = json.loads(msg.data).get("type") + if event == "session.done": + await front.send_str(json.dumps({"type": "response.done"})) + return + if event == "error": + await front.send_str(json.dumps({"type": "error", "error": "tts backend error"})) + return + elif msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): + return + + +async def _handle(request: web.Request) -> web.WebSocketResponse: + front = web.WebSocketResponse(heartbeat=20, max_msg_size=0) + await front.prepare(request) + cfg = request.app["cfg"] + + session = aiohttp.ClientSession() + back: aiohttp.ClientWebSocketResponse | None = None + pump: asyncio.Task | None = None + parts: list[str] = [] + try: + async for msg in front: + if msg.type != web.WSMsgType.TEXT: + continue + data = json.loads(msg.data) + + if "config" in data and back is None: + voice = (data["config"] or {}).get("voice") or cfg["voice"] + back = await session.ws_connect(cfg["backend_url"], max_msg_size=0) + await back.send_str( + json.dumps({"type": "session.config", "response_format": cfg["response_format"], "voice": voice}) + ) + pump = asyncio.create_task(_pump_backend_to_front(back, front)) + elif data.get("type") == "input_text.append" and back is not None: + parts.append(data.get("text", "")) + elif data.get("type") == "input_text.commit" and back is not None: + # Normalize the full utterance once (digits may span chunks), then send. + await back.send_str(json.dumps({"type": "input.text", "text": _normalize("".join(parts))})) + parts = [] + await back.send_str(json.dumps({"type": "input.done"})) + if pump is not None: + await pump + break + except Exception as err: + logger.warning("tts bridge error: %s", err) + finally: + if pump is not None and not pump.done(): + pump.cancel() + if back is not None and not back.closed: + await back.close() + await session.close() + if not front.closed: + await front.close() + return front + + +def create_app(backend_url: str, voice: str, response_format: str) -> web.Application: + app = web.Application() + app["cfg"] = {"backend_url": backend_url, "voice": voice, "response_format": response_format} + app.router.add_get("/v1/tts", _handle) + return app + + +def main() -> None: + parser = argparse.ArgumentParser(description="JoyVL webui <-> Qwen3-TTS websocket bridge") + parser.add_argument("--host", default="0.0.0.0") + parser.add_argument("--port", type=int, default=8092) + parser.add_argument("--backend-url", default="ws://127.0.0.1:8091/v1/audio/speech/stream") + parser.add_argument("--voice", default="vivian") + parser.add_argument("--response-format", default="pcm") + args = parser.parse_args() + logging.basicConfig(level=logging.INFO) + web.run_app(create_app(args.backend_url, args.voice, args.response_format), host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/examples/online_serving/joyvl_interaction/cli/run_cli_demo.py b/examples/online_serving/joyvl_interaction/cli/run_cli_demo.py new file mode 100644 index 000000000..8d86e033b --- /dev/null +++ b/examples/online_serving/joyvl_interaction/cli/run_cli_demo.py @@ -0,0 +1,32 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import argparse + +from stream_client import stream + +ICON = {"response": "SPEAK ", "silence": "silence ", "delegate": "DELEGATE"} + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("video") + parser.add_argument("--server", default="http://127.0.0.1:8070") + parser.add_argument("--query", default=None) + parser.add_argument("--fps", type=float, default=1.0) + parser.add_argument("--session-id", default="cli") + args = parser.parse_args() + + for tick in stream(args.video, args.server, args.session_id, args.query, args.fps): + if tick.action == "silence": + continue + line = f"[{tick.t:6.1f}s] {ICON.get(tick.action)} {tick.text}" + if tick.delegation: + line += f" (-> {tick.delegation.get('question', '')})" + print(line, flush=True) + + +if __name__ == "__main__": + main() diff --git a/examples/online_serving/joyvl_interaction/cli/stream_client.py b/examples/online_serving/joyvl_interaction/cli/stream_client.py new file mode 100644 index 000000000..82053021c --- /dev/null +++ b/examples/online_serving/joyvl_interaction/cli/stream_client.py @@ -0,0 +1,79 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import base64 +import time +from collections.abc import Iterator +from dataclasses import dataclass + +import cv2 +import requests + + +@dataclass +class Tick: + index: int + t: float + action: str + text: str + skipped: bool + latency_ms: float + delegation: dict | None + + +def iter_frames(video_path: str, fps: float) -> Iterator[tuple[float, bytes]]: + cap = cv2.VideoCapture(video_path) + src_fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 + step = max(1, round(src_fps / fps)) + idx = 0 + while True: + ok, frame = cap.read() + if not ok: + break + if idx % step == 0: + ok, buf = cv2.imencode(".jpg", frame) + if ok: + yield idx / src_fps, buf.tobytes() + idx += 1 + cap.release() + + +def stream( + video_path: str, + server: str, + session_id: str, + query: str | None = None, + fps: float = 1.0, + realtime: bool = False, +) -> Iterator[Tick]: + """Send each sampled frame as one chat completion; yield the parsed action. + + ``query`` is attached to the first tick to arm the session; inject later + queries by posting another frame turn with a new text part. + """ + url = server.rstrip("/") + "/v1/chat/completions" + requests.post(server.rstrip("/") + "/reset", json={"session_id": session_id}, timeout=10) + for i, (t, jpeg) in enumerate(iter_frames(video_path, fps)): + data_url = "data:image/jpeg;base64," + base64.b64encode(jpeg).decode() + content = [{"type": "image_url", "image_url": {"url": data_url}}] + if i == 0 and query: + content.insert(0, {"type": "text", "text": query}) + resp = requests.post( + url, + json={"session_id": session_id, "messages": [{"role": "user", "content": content}]}, + timeout=120, + ).json() + info = resp.get("interaction", {}) + yield Tick( + index=info.get("frame_index", i + 1), + t=t, + action=info.get("action", "?"), + text=resp["choices"][0]["message"]["content"], + skipped=info.get("inference_skipped", False), + latency_ms=info.get("latency_ms", 0.0), + delegation=info.get("delegation"), + ) + if realtime: + time.sleep(1.0 / fps) diff --git a/examples/online_serving/joyvl_interaction/rtsp/README.md b/examples/online_serving/joyvl_interaction/rtsp/README.md new file mode 100644 index 000000000..6f6fcc73d --- /dev/null +++ b/examples/online_serving/joyvl_interaction/rtsp/README.md @@ -0,0 +1,46 @@ +# RTSP local streaming + +Simulate an RTSP camera from a local video file so the WebUI (or any RTSP client) can +test the RTSP input path without a physical IP camera. RTSP is a **webui-side input** — +the browser pulls the stream and feeds frames to the orchestrator over the normal API; +no serving-layer code is involved. + +Needs [`ffmpeg`](https://ffmpeg.org/) and [MediaMTX](https://github.com/bluenviron/mediamtx/releases) +(download the binary + `mediamtx.yml` and keep them next to these scripts, or point +`MEDIAMTX_BIN` / `MEDIAMTX_CONFIG` at them). + +## 1. Start the RTSP server + +```bash +bash ./mediamtx.sh # MediaMTX, listens on :8554 +``` + +## 2. Push a local video as an RTSP stream + +```bash +bash ./rtsp.sh ./videos/example.mp4 rtsp://127.0.0.1:8554/fire1 +``` + +`rtsp.sh` takes `[video-path] [rtsp-url]` (or set `VIDEO_PATH` / `RTSP_URL`). To stream a +whole directory of clips, one stream per file: + +```bash +bash ./rtsp_all.sh ./videos rtsp://127.0.0.1:8554 +``` + +## 3. Use it in the WebUI + +Enter the URL in the WebUI RTSP box: + +``` +rtsp://127.0.0.1:8554/fire1 +``` + +If the WebUI runs on another machine, replace `127.0.0.1` with the MediaMTX host's IP. + +## Notes + +- `rtsp.sh` transcodes audio to AAC; replace `-c:a aac -b:a 128k -ar 44100` with `-an` + if the source clip has no audio track and your `ffmpeg` errors on it. +- If a client cannot connect, check that port `8554` is reachable and not blocked by a + firewall, container network, or security group. diff --git a/examples/online_serving/joyvl_interaction/rtsp/mediamtx.sh b/examples/online_serving/joyvl_interaction/rtsp/mediamtx.sh new file mode 100755 index 000000000..1af9c0b63 --- /dev/null +++ b/examples/online_serving/joyvl_interaction/rtsp/mediamtx.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Start a local MediaMTX RTSP server (listens on :8554 by default). +# Download the binary + mediamtx.yml from https://github.com/bluenviron/mediamtx/releases +# and either keep them next to this script or point the vars below at them. + +MEDIAMTX_BIN="${MEDIAMTX_BIN:-./mediamtx}" +MEDIAMTX_CONFIG="${MEDIAMTX_CONFIG:-./mediamtx.yml}" + +exec "$MEDIAMTX_BIN" "$MEDIAMTX_CONFIG" diff --git a/examples/online_serving/joyvl_interaction/rtsp/rtsp.sh b/examples/online_serving/joyvl_interaction/rtsp/rtsp.sh new file mode 100755 index 000000000..03f8052dd --- /dev/null +++ b/examples/online_serving/joyvl_interaction/rtsp/rtsp.sh @@ -0,0 +1,38 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Loop a local video file and push it to an RTSP URL, simulating an RTSP camera. +# bash ./rtsp.sh [video-path] [rtsp-output-url] +# Defaults stream ./videos/example.mp4 to rtsp://127.0.0.1:8554/fire1 + +VIDEO_PATH="${1:-${VIDEO_PATH:-./videos/example.mp4}}" +RTSP_URL="${2:-${RTSP_URL:-rtsp://127.0.0.1:8554/fire1}}" + +if [[ ! -f "$VIDEO_PATH" ]]; then + echo "Video file not found: $VIDEO_PATH" >&2 + exit 1 +fi + +echo "Input video: $VIDEO_PATH" +echo "RTSP output: $RTSP_URL" + +# Drop "-c:a aac -b:a 128k -ar 44100" for "-an" if the source has no audio track. +exec ffmpeg \ + -hide_banner \ + -loglevel info \ + -re \ + -stream_loop -1 \ + -i "$VIDEO_PATH" \ + -vf "scale='min(1280,iw)':-2" \ + -c:v libx264 \ + -preset veryfast \ + -tune zerolatency \ + -b:v 2500k \ + -maxrate 2500k \ + -bufsize 5000k \ + -c:a aac \ + -b:a 128k \ + -ar 44100 \ + -f rtsp \ + -rtsp_transport udp \ + "$RTSP_URL" diff --git a/examples/online_serving/joyvl_interaction/rtsp/rtsp_all.sh b/examples/online_serving/joyvl_interaction/rtsp/rtsp_all.sh new file mode 100755 index 000000000..b73f1d871 --- /dev/null +++ b/examples/online_serving/joyvl_interaction/rtsp/rtsp_all.sh @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Push every .mp4 in a directory as its own RTSP stream (one ffmpeg per file), +# named rtsp:///. +# bash ./rtsp_all.sh [video-dir] [rtsp-base-url] + +VIDEO_DIR="${1:-./videos}" +RTSP_BASE="${2:-rtsp://127.0.0.1:8554}" + +if [[ ! -d "$VIDEO_DIR" ]]; then + echo "Video directory not found: $VIDEO_DIR" >&2 + exit 1 +fi + +shopt -s nullglob +VIDEO_FILES=("$VIDEO_DIR"/*.mp4) +shopt -u nullglob + +if [[ ${#VIDEO_FILES[@]} -eq 0 ]]; then + echo "No video files found in: $VIDEO_DIR" >&2 + exit 1 +fi + +echo "Found ${#VIDEO_FILES[@]} video(s) in $VIDEO_DIR" + +PIDS=() +cleanup() { + echo "Stopping all streams..." + for pid in "${PIDS[@]}"; do + kill "$pid" 2>/dev/null || true + done + wait 2>/dev/null +} +trap cleanup EXIT INT TERM + +for video in "${VIDEO_FILES[@]}"; do + stream_name="$(basename "${video%.*}")" + rtsp_url="${RTSP_BASE}/${stream_name}" + echo "Starting stream: $rtsp_url <- $(basename "$video")" + ffmpeg \ + -hide_banner \ + -loglevel error \ + -re \ + -stream_loop -1 \ + -i "$video" \ + -vf "format=yuv420p,scale='min(1280,iw)':'min(720,ih)':force_original_aspect_ratio=decrease:force_divisible_by=2" \ + -c:v libx264 \ + -preset veryfast \ + -tune zerolatency \ + -b:v 4000k \ + -maxrate 4500k \ + -bufsize 9000k \ + -c:a aac \ + -b:a 128k \ + -ar 44100 \ + -f rtsp \ + -rtsp_transport tcp \ + "$rtsp_url" & + PIDS+=($!) +done + +echo "All ${#VIDEO_FILES[@]} streams started. Press Ctrl+C to stop." +wait diff --git a/examples/online_serving/joyvl_interaction/scripts/start_all.sh b/examples/online_serving/joyvl_interaction/scripts/start_all.sh new file mode 100755 index 000000000..d2f512b26 --- /dev/null +++ b/examples/online_serving/joyvl_interaction/scripts/start_all.sh @@ -0,0 +1,108 @@ +#!/bin/bash +# One-shot JoyVL demo: model -> interaction orchestrator -> JD webui. +# +# The model is the only fixed component; every external module is pluggable and +# configured here via env (unset = disabled), matching JoyVL's design: +# +# TTS_URL voice out (point at our tts_bridge, or any OpenAI-Realtime-style TTS WS) +# ASR_URL voice in (point at our asr_bridge, or any compatible ASR WS) +# DELEGATION_BACKEND_URL orchestrator background brain for (OpenAI-compatible; DELEGATION_MODEL/DELEGATION_API_KEY pick model+key) +# DELEGATION_KIND=chat (default) text/VL brain answers; =image a text-to-image model generates a picture +# chat+Claude: DELEGATION_BACKEND_URL=https://api.anthropic.com/v1/ DELEGATION_MODEL=claude-... DELEGATION_API_KEY=sk-ant-... +# image+Qwen-Image: DELEGATION_KIND=image DELEGATION_BACKEND_URL=http://127.0.0.1:8091/v1 +# BACKGROUND_CODEX_API_URL webui-side background agent +# ENABLE_MEMORY (default 1) 3-tier memory; SUMMARIZER_BACKEND_URL/MODEL pick the summarizer (default: reuse main) +# +# Usage: bash start_all.sh +# TTS_URL=ws://host/v1/tts ASR_URL=ws://host/v1/asr GPU=0 bash start_all.sh +set -o pipefail + +CODE="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd)" +VENV="${VENV:-$CODE/.venv}" +WEBUI_DIR="${WEBUI_DIR:-$CODE/joyvl-interaction/joyvl-interaction-webui}" +MODEL="${MODEL:-jdopensource/JoyAI-VL-Interaction-Preview}" +SERVED_NAME="${SERVED_NAME:-JoyAI-VL-Interaction-Preview}" +GPU="${GPU:-0}" +MODEL_PORT="${MODEL_PORT:-8061}" +ORCH_PORT="${ORCH_PORT:-8070}" +WEBUI_PORT="${WEBUI_PORT:-8999}" +WEBUI_USERNAME="${WEBUI_USERNAME:-admin}" +WEBUI_PASSWORD="${WEBUI_PASSWORD:-changeme}" # override via env for any exposed deployment +ENABLE_MEMORY="${ENABLE_MEMORY:-1}" +RESPONSE_DEDUP="${RESPONSE_DEDUP:-0.85}" # demo drops near-duplicate narration; set 1.0 for reference (exact-only) +TTS_URL="${TTS_URL:-}" +ASR_URL="${ASR_URL:-}" +BACKGROUND_CODEX_API_URL="${BACKGROUND_CODEX_API_URL:-}" + +LOG=/tmp/joyvl_demo +mkdir -p "$LOG" + +[ -d "$WEBUI_DIR" ] || { echo "! WEBUI_DIR not found: $WEBUI_DIR (set WEBUI_DIR=...)"; exit 1; } + +kill_port() { + local p + p=$(ss -tlnp 2>/dev/null | grep ":$1 " | grep -oE 'pid=[0-9]+' | head -1 | cut -d= -f2) + [ -n "${p:-}" ] && kill "$p" 2>/dev/null + return 0 +} +wait_http() { + for _ in $(seq 1 150); do + curl -sk -m2 "$1" -o /dev/null 2>/dev/null && return 0 + sleep 2 + done + echo " ! timeout waiting for $1" + return 1 +} + +echo "Resetting ports ${MODEL_PORT} ${ORCH_PORT} ${WEBUI_PORT}…" +for p in "$MODEL_PORT" "$ORCH_PORT" "$WEBUI_PORT"; do kill_port "$p"; done +sleep 3 + +[ -f "$VENV/bin/activate" ] && source "$VENV/bin/activate" + +# detach into a new session so servers outlive this launcher +spawn() { local log="$1"; shift; setsid "$@" > "$log" 2>&1 < /dev/null & } + +echo "[1/3] model on GPU ${GPU}…" +spawn "$LOG/model.log" env HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1 CUDA_VISIBLE_DEVICES="$GPU" \ + vllm serve "$MODEL" --served-model-name "$SERVED_NAME" --port "$MODEL_PORT" \ + --gpu-memory-utilization 0.85 --max-model-len 131072 --enable-prefix-caching \ + --limit-mm-per-prompt '{"image":256,"video":1}' +wait_http "http://127.0.0.1:${MODEL_PORT}/health" && echo " up :${MODEL_PORT}" + +echo "[2/3] orchestrator (memory=${ENABLE_MEMORY})…" +ORCH_ARGS=(--port "$ORCH_PORT" --main-backend-url "http://127.0.0.1:${MODEL_PORT}/v1" + --main-model "$SERVED_NAME" + --response-dedup-threshold "$RESPONSE_DEDUP") +# delegation: point at a stronger background brain to enable it; otherwise keep it off +if [ -n "${DELEGATION_BACKEND_URL:-}" ]; then + ORCH_ARGS+=(--delegation-backend-url "$DELEGATION_BACKEND_URL") + ORCH_ARGS+=(--delegation-kind "${DELEGATION_KIND:-chat}") + [ -n "${DELEGATION_MODEL:-}" ] && ORCH_ARGS+=(--delegation-model "$DELEGATION_MODEL") + [ -n "${DELEGATION_API_KEY:-}" ] && ORCH_ARGS+=(--delegation-api-key "$DELEGATION_API_KEY") +else + ORCH_ARGS+=(--no-delegation) +fi +[ "$ENABLE_MEMORY" = "0" ] && ORCH_ARGS+=(--no-memory) +[ -n "${SUMMARIZER_BACKEND_URL:-}" ] && ORCH_ARGS+=(--summarizer-backend-url "$SUMMARIZER_BACKEND_URL") +[ -n "${SUMMARIZER_MODEL:-}" ] && ORCH_ARGS+=(--summarizer-model "$SUMMARIZER_MODEL") +spawn "$LOG/orch.log" env PYTHONPATH="$CODE" python -m vllm_omni.experimental.fullduplex.joyvl.serving.server "${ORCH_ARGS[@]}" +wait_http "http://127.0.0.1:${ORCH_PORT}/health" && echo " up :${ORCH_PORT}" + +echo "[3/3] JD webui…" +spawn "$LOG/webui.log" env WEBUI_USERNAME="$WEBUI_USERNAME" WEBUI_PASSWORD="$WEBUI_PASSWORD" \ + TTS_URL="$TTS_URL" ASR_URL="$ASR_URL" BACKGROUND_CODEX_API_URL="$BACKGROUND_CODEX_API_URL" \ + bash -c "cd '$WEBUI_DIR' && exec bash scripts/start_server.sh --api-base 'http://127.0.0.1:${ORCH_PORT}/v1'" +webui_up=0 +for _ in $(seq 1 60); do grep -q "Server startup complete" "$LOG/webui.log" 2>/dev/null && { webui_up=1; break; }; sleep 1; done +[ "$webui_up" = 1 ] || { echo "! webui did not start — see $LOG/webui.log"; exit 1; } + +IP=$(hostname -I 2>/dev/null | awk '{print $1}') +echo +echo "=== JoyVL demo ready ===" +echo " webui : https://${IP}:${WEBUI_PORT} (user ${WEBUI_USERNAME} / pass ${WEBUI_PASSWORD})" +echo " memory : $([ "$ENABLE_MEMORY" = 0 ] && echo off || echo on)" +echo " TTS : ${TTS_URL:-}" +echo " ASR : ${ASR_URL:-}" +echo " agent : ${BACKGROUND_CODEX_API_URL:-}" +echo " logs : ${LOG}/{model,orch,webui}.log" diff --git a/examples/online_serving/joyvl_interaction/scripts/start_asr.sh b/examples/online_serving/joyvl_interaction/scripts/start_asr.sh new file mode 100755 index 000000000..cbbb9547e --- /dev/null +++ b/examples/online_serving/joyvl_interaction/scripts/start_asr.sh @@ -0,0 +1,24 @@ +#!/bin/bash +# Serve Qwen3-ASR (the same audio-LLM AURA uses as its ASR stage) and run the +# webui<->ASR bridge. Then start the webui with: +# ASR_URL=ws://127.0.0.1:8093/v1/asr bash scripts/start_server.sh +set -euo pipefail + +ASR_MODEL="${ASR_MODEL:-Qwen/Qwen3-ASR-1.7B}" +ASR_PORT="${ASR_PORT:-8094}" +BRIDGE_PORT="${BRIDGE_PORT:-8093}" +GPU="${GPU:-2}" +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" + +# Qwen3-ASR is audio-in / text-out; served as a normal multimodal chat model. +CUDA_VISIBLE_DEVICES="${GPU}" vllm serve "${ASR_MODEL}" \ + --host 0.0.0.0 --port "${ASR_PORT}" \ + --gpu-memory-utilization 0.3 --trust-remote-code & +ASR_PID=$! +trap 'kill ${ASR_PID} 2>/dev/null || true' EXIT + +until curl -s "http://127.0.0.1:${ASR_PORT}/health" >/dev/null 2>&1; do sleep 2; done +python "${SCRIPT_DIR}/../bridges/asr_bridge.py" \ + --port "${BRIDGE_PORT}" \ + --backend-url "http://127.0.0.1:${ASR_PORT}" \ + --model "${ASR_MODEL}" diff --git a/examples/online_serving/joyvl_interaction/scripts/start_interaction.sh b/examples/online_serving/joyvl_interaction/scripts/start_interaction.sh new file mode 100644 index 000000000..b97e5e9c6 --- /dev/null +++ b/examples/online_serving/joyvl_interaction/scripts/start_interaction.sh @@ -0,0 +1,23 @@ +#!/bin/bash +# Launch the interaction orchestrator in front of a running model server. +set -euo pipefail + +PORT="${PORT:-8070}" +MAIN_BACKEND_URL="${MAIN_BACKEND_URL:-http://127.0.0.1:8061/v1}" +MAIN_MODEL="${MAIN_MODEL:-JoyAI-VL-Interaction-Preview}" +PERSONA="${PERSONA:-default}" + +# By default memory reuses the main model as its own summarizer. Set +# SUMMARIZER_BACKEND_URL/SUMMARIZER_MODEL to a dedicated Qwen3-VL-4B server, +# or pass --no-memory for the lightest setup. +EXTRA=() +[[ -n "${SUMMARIZER_BACKEND_URL:-}" ]] && EXTRA+=(--summarizer-backend-url "${SUMMARIZER_BACKEND_URL}") +[[ -n "${SUMMARIZER_MODEL:-}" ]] && EXTRA+=(--summarizer-model "${SUMMARIZER_MODEL}") +[[ "${NO_MEMORY:-0}" == "1" ]] && EXTRA+=(--no-memory) + +python -m vllm_omni.experimental.fullduplex.joyvl.serving.server \ + --port "${PORT}" \ + --main-backend-url "${MAIN_BACKEND_URL}" \ + --main-model "${MAIN_MODEL}" \ + --persona "${PERSONA}" \ + "${EXTRA[@]}" diff --git a/examples/online_serving/joyvl_interaction/scripts/start_model.sh b/examples/online_serving/joyvl_interaction/scripts/start_model.sh new file mode 100644 index 000000000..4cfa95d53 --- /dev/null +++ b/examples/online_serving/joyvl_interaction/scripts/start_model.sh @@ -0,0 +1,22 @@ +#!/bin/bash +# Serve the JoyVL interaction model as a plain VLM (NOT --omni: it is a +# standard Qwen3-VL autoregressive model, not an omni/diffusion model). +set -euo pipefail + +MODEL="${MODEL:-jdopensource/JoyAI-VL-Interaction-Preview}" +SERVED_NAME="${SERVED_NAME:-JoyAI-VL-Interaction-Preview}" +PORT="${PORT:-8061}" +GPU="${GPU:-0}" +# Covers the default short-term window (chunk_frames=100) + key_frames=0 (all frames +# summarized). Lower MAX_MODEL_LEN / IMAGE_LIMIT (and --chunk-frames) together for ~24GB. +MAX_MODEL_LEN="${MAX_MODEL_LEN:-131072}" +IMAGE_LIMIT="${IMAGE_LIMIT:-256}" + +CUDA_VISIBLE_DEVICES="${GPU}" vllm serve "${MODEL}" \ + --served-model-name "${SERVED_NAME}" \ + --port "${PORT}" \ + --max-model-len "${MAX_MODEL_LEN}" \ + --gpu-memory-utilization 0.85 \ + --enable-prefix-caching \ + --enable-chunked-prefill \ + --limit-mm-per-prompt "{\"image\":${IMAGE_LIMIT},\"video\":1}" diff --git a/examples/online_serving/joyvl_interaction/scripts/start_tts.sh b/examples/online_serving/joyvl_interaction/scripts/start_tts.sh new file mode 100755 index 000000000..2c6c2ab35 --- /dev/null +++ b/examples/online_serving/joyvl_interaction/scripts/start_tts.sh @@ -0,0 +1,25 @@ +#!/bin/bash +# Serve Qwen3-TTS (vLLM-Omni) and run the webui<->TTS bridge. +# Then start the webui with: TTS_URL=ws://127.0.0.1:8092/v1/tts bash scripts/start_server.sh +set -euo pipefail + +TTS_MODEL="${TTS_MODEL:-Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice}" +TTS_PORT="${TTS_PORT:-8091}" +BRIDGE_PORT="${BRIDGE_PORT:-8092}" +GPU="${GPU:-1}" +VOICE="${VOICE:-vivian}" +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" + +CUDA_VISIBLE_DEVICES="${GPU}" vllm-omni serve "${TTS_MODEL}" \ + --deploy-config vllm_omni/deploy/qwen3_tts.yaml \ + --host 0.0.0.0 --port "${TTS_PORT}" \ + --gpu-memory-utilization 0.9 --trust-remote-code --omni & +TTS_PID=$! +trap 'kill ${TTS_PID} 2>/dev/null || true' EXIT + +# wait for the TTS server, then run the bridge in the foreground +until curl -s "http://127.0.0.1:${TTS_PORT}/health" >/dev/null 2>&1; do sleep 2; done +python "${SCRIPT_DIR}/../bridges/tts_bridge.py" \ + --port "${BRIDGE_PORT}" \ + --backend-url "ws://127.0.0.1:${TTS_PORT}/v1/audio/speech/stream" \ + --voice "${VOICE}" diff --git a/mkdocs.yml b/mkdocs.yml index 74f2145fe..8d5ffaa1b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -102,6 +102,7 @@ plugins: - "vllm_omni.entrypoints.async_diffusion" # avoid importing vllm in mkdocs building - "vllm_omni.entrypoints.openai" # avoid importing vllm in mkdocs building - "vllm_omni.entrypoints.openai.protocol" # avoid importing vllm in mkdocs building + - "vllm_omni.experimental" # avoid importing optional serving deps (openai, fastapi, uvicorn, httpx) in mkdocs building - mkdocstrings: handlers: python: diff --git a/pyproject.toml b/pyproject.toml index c02a079f6..c8a8799d8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,7 +71,9 @@ dev = [ ] demo = [ - "gradio>=6.7.0", + "opencv-python>=4.12.0.88", # cli/stream_client frame reading + "requests>=2.28.0", # cli/stream_client HTTP + "cn2an>=0.5.22", # Chinese number normalization for TTS (e.g. "1次" -> "一次") ] # Local-test-only extras @@ -181,6 +183,7 @@ ignore = [ # These files keep upstream conventions (star imports, long lines, math-symbol # argument names) so they stay easy to diff against upstream releases. "vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni_llm.py" = ["E402", "E501", "F403", "F405", "N803"] +"vllm_omni/experimental/fullduplex/joyvl/decision/prompts.py" = ["E501"] # System/summarizer prompts are long prose lines [tool.ruff.lint.flake8-tidy-imports.banned-api] "librosa".msg = "The librosa module is banned, use vllm.multimodal helpers instead" diff --git a/recipes/JD/JoyAI-VL-Interaction.md b/recipes/JD/JoyAI-VL-Interaction.md new file mode 100644 index 000000000..60f6e1a86 --- /dev/null +++ b/recipes/JD/JoyAI-VL-Interaction.md @@ -0,0 +1,192 @@ +# JoyAI-VL-Interaction + +> Real-time streaming video-language interaction (proactive speak / silence / delegate) + +## Summary + +- Vendor: JD (Joy Future Academy) +- Model: [`jdopensource/JoyAI-VL-Interaction-Preview`](https://huggingface.co/jdopensource/JoyAI-VL-Interaction-Preview) (8B, vanilla Qwen3-VL weights) +- Task: Per-tick proactive interaction over a live video stream — the model decides + on its own each second to speak, stay silent, or delegate a hard question +- Mode: Online serving — an OpenAI-compatible interaction orchestrator in front of + a plain `vllm serve` backend +- Maintainer: Community + +## When to use this recipe + +Use this to stand up the streaming-interaction serving layer (`vllm_omni/experimental/fullduplex/`). +The model is served unchanged by `vllm serve`; this layer adds session state, 3-tier +summary memory, the per-tick decision, and pluggable ASR / TTS / delegation. For the +framework internals and how to add another full-duplex model, see +[`vllm_omni/experimental/fullduplex/README.md`](../../vllm_omni/experimental/fullduplex/README.md). + +## Environment + +- OS: Linux +- Python: 3.10+ +- Hardware: 1x GPU. The default `T_s=100` frame window + `--max-model-len 131072` wants + ≈48GB+; for ~24GB, lower `--chunk-frames`, `--max-model-len`, and the image limit together +- vLLM / vLLM-Omni: versions from your current checkout + +## Start server + +From repository root: + +```bash +# 1. Serve the model (plain `vllm serve`, NOT --omni; it is vanilla Qwen3-VL). +# The image limit must cover the short-term frame window (chunk_frames, default 100 = T_s); +# prefix caching keeps the accumulating window cheap. Lower both for smaller GPUs. +vllm serve jdopensource/JoyAI-VL-Interaction-Preview \ + --served-model-name JoyAI-VL-Interaction-Preview --port 8061 \ + --max-model-len 131072 --enable-prefix-caching \ + --limit-mm-per-prompt '{"image":256,"video":1}' + +# 2. Interaction orchestrator (OpenAI-compatible, :8070) +python -m vllm_omni.experimental.fullduplex.joyvl.serving.server --port 8070 \ + --main-backend-url http://127.0.0.1:8061/v1 --main-model JoyAI-VL-Interaction-Preview +``` + +Optional one-shot launch (model + orchestrator + JD webui + ASR/TTS/background, +all env-configurable). The JD webui frontend is external — set `WEBUI_DIR` to point at it: + +```bash +bash examples/online_serving/joyvl_interaction/scripts/start_all.sh +``` + +## Using the model + +The orchestrator is OpenAI-compatible: send **one user turn per video frame** (~1 fps) to +`/v1/chat/completions` with an `x-session-id` header, and attach an optional **standing +instruction** as a text part. Each reply carries an `interaction` block — `action` is +`silence` / `response` / `delegate` and `text` is what to say: + +```bash +curl -s http://127.0.0.1:8070/v1/chat/completions \ + -H 'x-session-id: s1' -H 'content-type: application/json' -d '{ + "messages": [{"role": "user", "content": [ + {"type": "text", "text": "Alert me if a fire breaks out"}, + {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,..."}} + ]}]}' | jq .interaction +``` + +Send the standing instruction once (it persists for the session); subsequent turns send +just the frame. Ready-made headless client: `cli/run_cli_demo.py`. Reset a session with +`POST /reset {"session_id": "..."}`. + +**What to ask it** — give a standing task and let it act on its own each second: + +- **Proactive alerting** — "tell me when someone enters", "alert me if a fire breaks out" +- **Streaming Q&A** — ask about what's on screen; it answers once it has the evidence +- **Time & memory** — "how many people have walked past?", "what did you see earlier?" + (the 3-tier summary memory lets it recall beyond the live frame window) +- **Read & translate** — "translate the on-screen text as it appears" +- **Delegate hard questions** — handed to a background brain (see below) +- **Live commentary** — `--persona talkative` for continuous, danmaku-style narration + +**Personas** (`--persona`): `default` (speak on meaningful events or to answer), `silent` +(answer only when asked, never delegate), `talkative` (proactively narrate). + +**Tuning:** `--chunk-frames` (short-term window `T_s`), `--response-dedup-threshold` +(lower drops more near-duplicate narration), and `--no-memory` to disable the 3-tier +summaries. The model stays silent until the first instruction arrives +(`force_silence_before_query`, on by default). + +## Delegation (background brain) + +When the model judges a question too hard, it emits ` ` and the +orchestrator hands it to a **background brain** — any OpenAI-compatible endpoint you +self-host. Enable it by pointing the orchestrator at one: + +```bash +python -m vllm_omni.experimental.fullduplex.joyvl.serving.server --port 8070 \ + --main-backend-url http://127.0.0.1:8061/v1 --main-model JoyAI-VL-Interaction-Preview \ + --delegation-backend-url /v1 \ + --delegation-model --delegation-kind chat +``` + +`--delegation-kind` picks the bridge: + +- `chat` — a stronger text/VL model answers (`/chat/completions`) +- `image` — a text-to-image model generates a picture (`/images/generations`, e.g. Qwen-Image) +- `edit` — an image-edit model restyles the current frame (e.g. Qwen-Image-Edit) +- `router` — dispatch each request to chat / image / edit by inspecting it (set + `--delegation-image-url` / `--delegation-edit-url` for the latter two) +- `stub` — canned answers for tests/demos only (no backend needed) + +`chat`/`image`/`edit`/`router` each need a backend URL — **without one, delegation stays +off** (the model's delegate note is still spoken, but nothing is folded back). The brain is +**bring-your-own**: a larger vLLM you serve, or any OpenAI-compatible API (e.g. +`--delegation-backend-url https://api.anthropic.com/v1/ --delegation-model claude-... +--delegation-api-key …`). The reference deployment instead drives the `codex` CLI as the +brain via a separate background-agent service; that agent runs with its own credentials +and bypasses its sandbox, so it is **not bundled here** — self-host a plain +OpenAI-compatible endpoint instead. + +## Host the WebUI demo + +For the full browser experience — live webcam / RTSP input, voice (ASR/TTS), and the +per-tick decision stream — run JD's official WebUI in front of the orchestrator. Clone the +model repo and start its WebUI pointed at the orchestrator (`:8070`): + +```bash +git clone https://github.com/jd-opensource/JoyAI-VL-Interaction.git +cd JoyAI-VL-Interaction/services/webui +uv venv && uv pip install -e . +bash scripts/start_server.sh --api-base http://127.0.0.1:8070/v1 +``` + +Open the printed HTTPS URL, allow the camera (or enter an RTSP URL), and give a standing +instruction. `examples/online_serving/joyvl_interaction/scripts/start_all.sh` can launch +the model + orchestrator + WebUI together — point `WEBUI_DIR` at your clone. + +## Verification + +```bash +# headless: stream a clip and print the per-tick decision timeline +# (the CLI reads video frames via OpenCV: `uv pip install opencv-python`) +python examples/online_serving/joyvl_interaction/cli/run_cli_demo.py \ + path/to/video.mp4 --query "Alert me if a fire breaks out" + +pytest tests/fullduplex # framework + JoyVL unit tests +``` + +## Testing with an RTSP stream (optional) + +RTSP is a **webui-side input** — the browser pulls the stream and feeds frames to the +orchestrator over the normal API; no serving-layer code is involved. To simulate an RTSP +camera from a local video file (no physical IP camera needed), use the helper scripts in +[`examples/online_serving/joyvl_interaction/rtsp/`](../../examples/online_serving/joyvl_interaction/rtsp/), +which wrap [MediaMTX](https://github.com/bluenviron/mediamtx/releases) + `ffmpeg`: + +```bash +cd examples/online_serving/joyvl_interaction/rtsp + +# 1. Local RTSP server (MediaMTX, listens on :8554) +bash ./mediamtx.sh + +# 2. Push a local video file as an RTSP stream (another terminal) +bash ./rtsp.sh ./videos/example.mp4 rtsp://127.0.0.1:8554/fire1 + +# 3. In the WebUI RTSP box, enter: rtsp://127.0.0.1:8554/fire1 +# (replace 127.0.0.1 with the MediaMTX host IP if the webui runs on another machine) +``` + +See the directory's `README.md` for streaming a whole video folder (`rtsp_all.sh`) and +the audio-track caveat. + +## Notes + +- `--omni` is **not** used: the model is standard Qwen3-VL, so stock `vllm serve` + runs the forward pass; this recipe only adds the interaction/serving layer. +- On a host without `nvcc` / `ninja`, `vllm serve` of the 8B can crash engine-core in the + FlashInfer sampler JIT (`FileNotFoundError: 'ninja'`) during `profile_run`. Set + `VLLM_USE_FLASHINFER_SAMPLER=0` (or install `ninja`) to work around it. +- `force_silence_before_query` is on by default — the model stays silent until an + instruction arrives; give a standing task (e.g. "translate the on-screen text") + to arm proactive output. +- Speech is external and pluggable: point `ASR_URL` / `TTS_URL` at the bridges in + `examples/online_serving/joyvl_interaction/bridges/` or any compatible service. +- The decision prompts, sampling, and 3-tier summary memory (`T_s=100`, mid→long every + 5 chunks, `key_frames=0` = summarize all chunk frames) are aligned to the JoyVL + reference adapter so per-tick behavior matches the released model; the framework only + supplies the serving structure. diff --git a/recipes/README.md b/recipes/README.md index e3ee2bc56..30f1577be 100644 --- a/recipes/README.md +++ b/recipes/README.md @@ -64,6 +64,7 @@ recipes/ | [`StabilityAI/Stable-Diffusion-3.5.md`](./StabilityAI/Stable-Diffusion-3.5.md) | Text-to-image serving (SD 3.5-medium and SD 3.5-large) | 1x RTX A6000 48GB | | [`zai-org/GLM-TTS.md`](./zai-org/GLM-TTS.md) | Online serving for Chinese/English zero-shot voice-cloned TTS | 1x A40 48GB | | [`GLM/GLM-Image.md`](./GLM/GLM-Image.md) | Online serving for image generation | 1x A800 80GB / 2x A800 80GB | +| [`JD/JoyAI-VL-Interaction.md`](./JD/JoyAI-VL-Interaction.md) | Real-time streaming video-language interaction (proactive speak/silence/delegate) | 1x GPU 24GB+ | Within a single recipe file, include different hardware support sections such as `GPU`, `ROCm`, and `NPU`, and add concrete tested configurations like diff --git a/tests/fullduplex/test_joyvl_brain.py b/tests/fullduplex/test_joyvl_brain.py new file mode 100644 index 000000000..0367eebb3 --- /dev/null +++ b/tests/fullduplex/test_joyvl_brain.py @@ -0,0 +1,96 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +import pytest + +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import parse_action +from vllm_omni.experimental.fullduplex.joyvl.memory.brain import InteractionBrain + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +class _FakeSummarizer: + def __init__(self): + self.chunks = 0 + + async def summarize_chunk(self, chunk_index, frame_range, frames): + self.chunks += 1 + return f"summary-of-chunk-{chunk_index}" + + async def compress_to_long_term(self, existing, mids): + return (existing + " | " if existing else "") + f"compressed({len(mids)})" + + +class _FakeDelegation: + def __init__(self): + self._ready = {} + + async def submit(self, question, note, frames): + self._ready["t1"] = question + return "t1" + + async def poll(self, task_id): + from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import DelegationResult + + return DelegationResult(task_id, "ready", digest=f"answer to {self._ready[task_id]}") + + +def test_query_freshness_and_archive_on_change(): + b = InteractionBrain() + assert b.update_query("count bottles") is True + assert b.update_query("count bottles") is False + b.record_response("1 bottle") + assert b.update_query("now describe") is True + assert b.memory.qa_history[-1].query == "count bottles" + assert b.memory.qa_history[-1].responses == [("0.0s", "1 bottle")] + + +def test_query_in_frame_on_arrival_then_prefix_on_later_chunks(): + b = InteractionBrain(frame_seconds=1.0) + b.tick(3) + b.update_query("alert me if a fire breaks out") + assert b.query_in_current_chunk is True + assert "alert me if a fire breaks out" not in b.build_prefix() + b.record_response("a fire is breaking out") + b.close_chunk() + assert b.query_in_current_chunk is False + prefix = b.build_prefix() + assert "alert me if a fire breaks out" in prefix + assert "a fire is breaking out" in prefix + + +@pytest.mark.asyncio +async def test_flush_archives_qa_and_summarizes(): + summ = _FakeSummarizer() + b = InteractionBrain(summarizer=summ, chunk_frames=3, long_term_every_n_chunks=2, frame_seconds=1.0) + b.update_query("watch the room") + b.tick(3) + assert b.should_flush() is True + b.record_response("a person enters") + await b.flush([("0.0s", "u0"), ("2.0s", "u2")]) + assert summ.chunks == 1 + assert b.chunk_index == 2 + assert b.should_flush() is False + assert b.memory.qa_history[-1].responses == [("2.0s", "a person enters")] + assert len(b.memory.mid_term_summaries) == 1 + + +@pytest.mark.asyncio +async def test_long_term_compression_after_n_chunks(): + summ = _FakeSummarizer() + b = InteractionBrain(summarizer=summ, chunk_frames=1, long_term_every_n_chunks=2, frame_seconds=1.0) + for _ in range(2): + await b.flush([("0.0s", "u")]) + assert b.memory.long_term_memory.startswith("compressed(") + assert b.memory.mid_term_summaries == [] + + +@pytest.mark.asyncio +async def test_delegation_submit_and_fold(): + b = InteractionBrain(delegation=_FakeDelegation()) + action = parse_action(" hold on solve this") + info = await b.submit_delegation(action, []) + assert info["status"] == "submitted" + folded = await b.fold_delegations() + assert folded["status"] == "ready" + assert b.memory.qa_history[-1].responses[0][1] == "answer to solve this" diff --git a/tests/fullduplex/test_joyvl_policy.py b/tests/fullduplex/test_joyvl_policy.py new file mode 100644 index 000000000..4082c4c9c --- /dev/null +++ b/tests/fullduplex/test_joyvl_policy.py @@ -0,0 +1,282 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +import asyncio + +import pytest + +from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import ( + DelegationResult, + ImageEditDelegationBridge, + ImageGenDelegationBridge, + OpenAIDelegationBridge, +) +from vllm_omni.experimental.fullduplex.joyvl.decision.policy import JoyVLPolicy, sample_frames + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def test_sample_frames_keeps_recent(): + assert sample_frames(["a", "b"], 4) == ["a", "b"] + out = sample_frames([str(i) for i in range(10)], 4) + assert out[-1] == "9" and len(out) == 4 + + +def test_build_messages_stable_head_and_query(): + p = JoyVLPolicy(num_frames=4) + p.tick(2) + p.set_query("count the bottles") + msgs, user = p.build_messages([{"type": "image_url", "image_url": {"url": "x"}}]) + assert msgs[0]["role"] == "system" and "" in msgs[0]["content"] + + assert any("count the bottles" in part.get("text", "") for part in user["content"]) + + +def test_commit_records_only_spoken(): + p = JoyVLPolicy() + p.set_query("q") + assert p.commit("").action.value == "silence" + assert p.brain.response_records == [] + assert p.commit(" hello").action.value == "response" + assert p.brain.response_records == [("0.0s", "hello")] + + +def test_commit_dedup_default_is_exact_match_only(): + p = JoyVLPolicy() # threshold 1.0 == reference behavior + assert p.commit(" a red ball appears").action.value == "response" + assert p.commit(" a red ball appears").action.value == "silence" # exact repeat dropped + assert p.commit(" a red ball is appearing").action.value == "response" # near-dup kept + + +def test_commit_dedup_threshold_drops_near_duplicates(): + p = JoyVLPolicy(response_dedup_threshold=0.8) + assert p.commit(" a red ball appears").action.value == "response" + assert p.commit(" a red ball appears!").action.value == "silence" # near-dup dropped + assert p.commit(" the sky is now blue").action.value == "response" # distinct kept + + +def test_commit_dedup_rearms_after_silence(): + p = JoyVLPolicy(response_dedup_threshold=0.85) + assert p.commit(" someone raises a hand").action.value == "response" + # no gap: an immediate repeat is still suppressed + assert p.commit(" someone raises a hand").action.value == "silence" + # a genuine silent tick (hand lowered) re-arms; the recurrence speaks again + assert p.commit("").action.value == "silence" + assert p.commit(" someone raises a hand").action.value == "response" + + +class _FakeBrain(OpenAIDelegationBridge): + """Override only the network call so submit/poll plumbing is exercised for real.""" + + def __init__(self, answer, gate=None): + super().__init__("http://unused/v1", "background-model") + self._answer = answer + self._gate = gate + + async def _complete(self, messages): + if self._gate is not None: + await self._gate.wait() + return self._answer + + +@pytest.mark.asyncio +async def test_openai_delegation_bridge_is_nonblocking_and_resolves(): + gate = asyncio.Event() + bridge = _FakeBrain("the bottle is on the left shelf", gate=gate) + frames = [("0.0s", "data:image/jpeg;base64,AAA")] + task_id = await bridge.submit("where is the bottle?", "let me check", frames) + + # poll must not block on the slow background brain + assert (await bridge.poll(task_id)).status == "pending" + + gate.set() + await asyncio.sleep(0) + for _ in range(10): + result = await bridge.poll(task_id) + if result.status != "pending": + break + await asyncio.sleep(0) + assert result.is_ready + assert "bottle is on the left shelf" in result.digest + assert (await bridge.poll("missing")).status == "error" + + +class _FakeImageBrain(ImageGenDelegationBridge): + async def _call_images(self, payload): + # echo the prompt back so the test can assert it was passed through + return {"data": [{"b64_json": "QUJD"}]} # "ABC" + + +@pytest.mark.asyncio +async def test_image_delegation_bridge_returns_media_not_text(): + bridge = _FakeImageBrain("http://unused/v1") + task_id = await bridge.submit("draw a red dragon", "", []) + for _ in range(10): + result = await bridge.poll(task_id) + if result.status != "pending": + break + await asyncio.sleep(0) + assert result.is_ready + # the image rides in media as a data URL; digest is only a short text placeholder + assert result.media == "data:image/png;base64,QUJD" + assert "QUJD" not in result.digest and "red dragon" in result.digest + + +class _FakeEditBrain(ImageEditDelegationBridge): + def __init__(self, base_url, model): + super().__init__(base_url, model) + self.seen_frame = None + + async def _call_chat(self, frame, instruction): + self.seen_frame = frame + return {"choices": [{"message": {"content": [{"image_url": {"url": "data:image/png;base64,Q0FSVA=="}}]}}]} + + +@pytest.mark.asyncio +async def test_image_edit_delegation_conditions_on_frame_and_returns_media(): + bridge = _FakeEditBrain("http://unused/v1", "Qwen-Image-Edit") + frames = [("0.0s", "data:image/jpeg;base64,OLD"), ("1.0s", "data:image/jpeg;base64,NOW")] + task_id = await bridge.submit("把画面变成卡通风格", "", frames) + for _ in range(10): + result = await bridge.poll(task_id) + if result.status != "pending": + break + await asyncio.sleep(0) + assert result.is_ready + assert bridge.seen_frame == "data:image/jpeg;base64,NOW" # latest frame, not the old one + assert result.media == "data:image/png;base64,Q0FSVA==" + assert result.media not in result.digest # image rides in media, not context + + +@pytest.mark.asyncio +async def test_image_edit_delegation_errors_without_frame(): + bridge = _FakeEditBrain("http://unused/v1", "Qwen-Image-Edit") + task_id = await bridge.submit("cartoonify", "", []) + for _ in range(10): + result = await bridge.poll(task_id) + if result.status != "pending": + break + await asyncio.sleep(0) + assert result.status == "error" + + +class _NamedBridge: + def __init__(self, name): + self.name = name + + async def submit(self, question, note, frames): + return f"{self.name}-1" + + async def poll(self, task_id): + return DelegationResult(task_id, "ready", digest=self.name) + + +@pytest.mark.asyncio +async def test_routing_delegation_dispatches_by_request(): + from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import RoutingDelegationBridge + + chat, image, edit = _NamedBridge("chat"), _NamedBridge("image"), _NamedBridge("edit") + r = RoutingDelegationBridge(chat=chat, image=image, edit=edit) + + async def route(q): + tid = await r.submit(q, "", []) + return (await r.poll(tid)).digest + + assert await route("把你看到的画面变成卡通风格") == "edit" # restyle current view + assert await route("画一只红色的龙") == "image" # generate a new image + assert await route("这道数学题的答案是什么") == "chat" # hard question -> chat brain + + +class _Delegation: + async def submit(self, question, note, frames): + return "t1" + + async def poll(self, task_id): + return DelegationResult(task_id, "ready", digest="background answer") + + +def test_build_delegation_off_unless_backend_configured(): + from vllm_omni.experimental.fullduplex.joyvl.serving.config import InteractionConfig + from vllm_omni.experimental.fullduplex.joyvl.serving.server import SessionManager + + # No backend URL -> delegation disabled (None), NOT a silent stub that folds fake answers. + assert SessionManager._build_delegation(InteractionConfig()) is None + assert SessionManager._build_delegation(InteractionConfig(delegation_kind="router")) is None + # Stub only on explicit opt-in. + assert SessionManager._build_delegation(InteractionConfig(delegation_kind="stub")) is not None + # A real chat backend -> a live bridge. + cfg = InteractionConfig(delegation_backend_url="http://x/v1", delegation_model="m") + assert SessionManager._build_delegation(cfg) is not None + + +@pytest.mark.asyncio +async def test_delegation_cancel_drops_task(): + from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import StubDelegationBridge + + b = StubDelegationBridge(ready_after_ticks=1) + task_id = await b.submit("q", "", []) + b.cancel(task_id) + # after cancel the task is gone -> poll reports it as unknown rather than leaking + assert (await b.poll(task_id)).status == "error" + + +@pytest.mark.asyncio +async def test_delegation_aclose_clears_pending(): + from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import StubDelegationBridge + + b = StubDelegationBridge(ready_after_ticks=5) + task_id = await b.submit("q", "", []) + await b.aclose() + assert (await b.poll(task_id)).status == "error" + + +@pytest.mark.asyncio +async def test_session_reset_cancels_and_awaits_consolidation(): + from vllm_omni.experimental.fullduplex.joyvl.serving.config import InteractionConfig + from vllm_omni.experimental.fullduplex.joyvl.serving.session import InteractionSession + + class _Backend: + async def generate(self, *a, **k): + return "", None + + async def aclose(self): + pass + + class _Summarizer: + async def summarize_chunk(self, *a, **k): + await asyncio.sleep(100) # hang so the consolidation task is in-flight + return "x" + + async def compress_to_long_term(self, *a, **k): + return "" + + async def aclose(self): + pass + + sess = InteractionSession("s", InteractionConfig(), _Backend(), summarizer=_Summarizer()) + sess._spawn_consolidation(1, [("0.0s", "u")]) + assert len(sess._consolidating) == 1 + await sess.reset() # must cancel AND await the in-flight task, not just clear the set + assert len(sess._consolidating) == 0 + + +@pytest.mark.asyncio +async def test_manager_aclose_closes_without_error(): + from vllm_omni.experimental.fullduplex.joyvl.serving.config import InteractionConfig + from vllm_omni.experimental.fullduplex.joyvl.serving.server import SessionManager + + mgr = SessionManager(InteractionConfig(enable_memory=False, enable_delegation=False)) + await mgr.aclose() + await mgr.aclose() # idempotent + + +@pytest.mark.asyncio +async def test_delegation_submit_and_fold(): + p = JoyVLPolicy(delegation=_Delegation()) + action = p.commit(" hold on hard question") + assert action.action.value == "delegate" + info = await p.submit_if_delegate(action) + assert info["status"] == "submitted" + folded = await p.fold_delegations() + assert folded["status"] == "ready" + assert p.brain.memory.qa_history[-1].responses[0][1] == "background answer" diff --git a/tests/fullduplex/test_runtime.py b/tests/fullduplex/test_runtime.py new file mode 100644 index 000000000..fe25ce42f --- /dev/null +++ b/tests/fullduplex/test_runtime.py @@ -0,0 +1,172 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +import asyncio +from collections.abc import AsyncIterator + +import pytest + +from vllm_omni.experimental.fullduplex.core import protocol as ev +from vllm_omni.experimental.fullduplex.core.adapter import DuplexCapability, OutputChunk +from vllm_omni.experimental.fullduplex.core.runtime import DuplexRuntime +from vllm_omni.experimental.fullduplex.core.session import DuplexSession, DuplexSessionConfig +from vllm_omni.experimental.fullduplex.joyvl.adapter import JoyVLDuplexAdapter + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +async def _feed(events): + for e in events: + yield e + + +def _collector(): + out: list[dict] = [] + + async def emit(event: dict) -> None: + out.append(event) + + return out, emit + + +class _FakeAdapter: + def __init__(self, chunks, barge_after=None): + self._chunks = chunks + self._barge_after = barge_after + + def capabilities(self): + return DuplexCapability(frozenset({"text"}), frozenset({"text"}), proactive=False) + + async def on_input(self, session, modality, data): + pass + + def should_respond(self, session): + return True + + async def respond(self, session) -> AsyncIterator[OutputChunk]: + for i, c in enumerate(self._chunks): + if self._barge_after is not None and i == self._barge_after: + session.barge_in() + yield OutputChunk("text", c) + + async def on_barge_in(self, session): + pass + + async def on_playback_ack(self, session, cursor): + pass + + +@pytest.mark.asyncio +async def test_runtime_basic_response(): + session = DuplexSession("s", DuplexSessionConfig(output_modalities=("text",))) + rt = DuplexRuntime(session, _FakeAdapter(["a", "b"])) + out, emit = _collector() + await rt.run(_feed([{"type": ev.INPUT_COMMIT}, {"type": ev.CLOSE}]), emit) + types = [e["type"] for e in out] + assert types == [ev.RESPONSE_CREATED, ev.RESPONSE_DELTA, ev.RESPONSE_DELTA, ev.RESPONSE_DONE] + assert [e["data"] for e in out if e["type"] == ev.RESPONSE_DELTA] == ["a", "b"] + + +@pytest.mark.asyncio +async def test_runtime_barge_in_drops_stale_output(): + session = DuplexSession("s") + rt = DuplexRuntime(session, _FakeAdapter(["a", "b", "c"], barge_after=1)) + out, emit = _collector() + await rt.run(_feed([{"type": ev.INPUT_COMMIT}, {"type": ev.CLOSE}]), emit) + data = [e["data"] for e in out if e["type"] == ev.RESPONSE_DELTA] + assert data == ["a"] + assert ev.RESPONSE_DONE not in [e["type"] for e in out] + + +class _SlowAdapter: + def capabilities(self): + return DuplexCapability(frozenset({"text"}), frozenset({"text"}), proactive=False) + + async def on_input(self, session, modality, data): + pass + + def should_respond(self, session): + return True + + async def respond(self, session) -> AsyncIterator[OutputChunk]: + for c in ["a", "b", "c", "d", "e"]: + await asyncio.sleep(0.02) + yield OutputChunk("text", c) + + async def on_barge_in(self, session): + pass + + async def on_playback_ack(self, session, cursor): + pass + + +@pytest.mark.asyncio +async def test_runtime_cancel_event_interrupts_active_response(): + session = DuplexSession("s") + rt = DuplexRuntime(session, _SlowAdapter()) + out, emit = _collector() + + async def feed(): + yield {"type": ev.INPUT_COMMIT} + await asyncio.sleep(0.03) + yield {"type": ev.RESPONSE_CANCEL} + yield {"type": ev.CLOSE} + + await rt.run(feed(), emit) + deltas = [e["data"] for e in out if e["type"] == ev.RESPONSE_DELTA] + types = [e["type"] for e in out] + assert len(deltas) < 5 + assert ev.RESPONSE_CANCELLED in types + assert ev.RESPONSE_DONE not in types + + +@pytest.mark.asyncio +async def test_runtime_new_response_supersedes_inflight_without_blocking(): + # A second response trigger arriving while the first is still streaming must + # supersede it (barge-in) rather than block the input loop until it finishes. + session = DuplexSession("s") + rt = DuplexRuntime(session, _SlowAdapter()) + out, emit = _collector() + + async def feed(): + yield {"type": ev.INPUT_COMMIT} + await asyncio.sleep(0.03) # let the first response emit ~one chunk + yield {"type": ev.INPUT_COMMIT} # supersedes the in-flight first response + await asyncio.sleep(0.2) # let the second response finish + yield {"type": ev.CLOSE} + + await rt.run(feed(), emit) + created = [e for e in out if e["type"] == ev.RESPONSE_CREATED] + done = [e for e in out if e["type"] == ev.RESPONSE_DONE] + # two responses created; only the latest one completes (the first was cancelled) + assert len(created) == 2 + assert len(done) == 1 + assert done[0]["response_index"] == created[-1]["response_index"] + + +@pytest.mark.asyncio +async def test_joyvl_adapter_speaks_then_silences(): + replies = iter([" a fire is breaking out", ""]) + + async def fake_generate(messages): + return next(replies) + + adapter = JoyVLDuplexAdapter(fake_generate, num_frames=4) + cfg = DuplexSessionConfig(input_modalities=("video", "text"), output_modalities=("text",), proactive=True) + rt = DuplexRuntime(DuplexSession("s", cfg), adapter) + out, emit = _collector() + await rt.run( + _feed( + [ + {"type": ev.INPUT_APPEND, "modality": "text", "data": "alert me if a fire breaks out"}, + {"type": ev.INPUT_APPEND, "modality": "video", "data": "data:image/jpeg;base64,AAA"}, + {"type": ev.INPUT_APPEND, "modality": "video", "data": "data:image/jpeg;base64,BBB"}, + {"type": ev.CLOSE}, + ] + ), + emit, + ) + deltas = [e["data"] for e in out if e["type"] == ev.RESPONSE_DELTA] + assert "a fire is breaking out" in deltas + + assert deltas.count("a fire is breaking out") == 1 diff --git a/vllm_omni/experimental/__init__.py b/vllm_omni/experimental/__init__.py new file mode 100644 index 000000000..f8b58d229 --- /dev/null +++ b/vllm_omni/experimental/__init__.py @@ -0,0 +1,3 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Experimental vLLM-Omni features (APIs may change without notice).""" diff --git a/vllm_omni/experimental/fullduplex/README.md b/vllm_omni/experimental/fullduplex/README.md new file mode 100644 index 000000000..7f9e22fa6 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/README.md @@ -0,0 +1,80 @@ +# Full-duplex interaction framework + +A model-agnostic framework for real-time, full-duplex (streaming-in / streaming-out) +model interaction, plus the JoyVL implementation built on it. + +To **run the JoyVL model**, see the recipe: +[`recipes/JD/JoyAI-VL-Interaction.md`](../../../recipes/JD/JoyAI-VL-Interaction.md). +This README covers the framework itself and how to add a new model. + +## Layout + +``` +vllm_omni/experimental/fullduplex/ + core/ generic full-duplex framework (model-agnostic): DuplexRuntime (event + loop + epoch barge-in), DuplexSession, DuplexAdapter (ABC), protocol + joyvl/ JoyVL implementation (model-specific): + adapter.py JoyVLDuplexAdapter (implements core.DuplexAdapter) + decision/ policy + output_parser + prompts (speak/silence/delegate) + memory/ InteractionBrain — 3-tier summary memory (async) + serving/ OpenAI-compatible HTTP orchestrator + bridges/ model backend + delegation +``` + +`core/` is the only part shared across models; data planes differ by model and are +intentionally not shared. + +## Scope + +The runnable serving path is `joyvl/serving/` driving `decision/` + `memory/` directly. +`joyvl/adapter.py` + `core/` are a **demonstration** of how a model plugs into the +generic full-duplex framework (exercised by `tests/fullduplex/`), not the serving path +the HTTP orchestrator currently uses — a fused-audio model (e.g. MiniCPM-o) is the case +`core/` is built for. + +## Adding a full-duplex model + +The seam is `core.DuplexAdapter`. `core/` owns the session lifecycle, epoch-based +barge-in, playback cursor, and the event protocol — you implement only model policy. + +1. Create a sibling package `vllm_omni/experimental/fullduplex//` next to `joyvl/`; keep + model-specific code there and do not touch `core/`. +2. Implement one `DuplexAdapter` (three required methods; the rest have defaults): + + ```python + from collections.abc import AsyncIterator + from vllm_omni.experimental.fullduplex.core.adapter import DuplexAdapter, DuplexCapability, OutputChunk + from vllm_omni.experimental.fullduplex.core.session import DuplexSession + + class MyModelAdapter(DuplexAdapter): + def capabilities(self) -> DuplexCapability: + return DuplexCapability( + input_modalities=frozenset({"audio", "text"}), + output_modalities=frozenset({"audio", "text"}), + proactive=True, # speak without being asked? + ) + + async def on_input(self, session: DuplexSession, modality: str, data) -> None: + ... # buffer/route an incoming chunk + + async def respond(self, session: DuplexSession) -> AsyncIterator[OutputChunk]: + async for piece in self._model_stream(session): + yield OutputChunk(modality="audio", data=piece) # runtime drops stale + # chunks after a barge-in + + # optional: should_respond / on_barge_in / on_playback_ack + ``` + +3. Run it through the shared runtime — no new control-plane code: + + ```python + from vllm_omni.experimental.fullduplex.core.runtime import DuplexRuntime + from vllm_omni.experimental.fullduplex.core.session import DuplexSession, DuplexSessionConfig + + rt = DuplexRuntime(DuplexSession("sid", DuplexSessionConfig()), MyModelAdapter()) + await rt.run(input_events, emit) + ``` + +`joyvl/adapter.py` is the worked demonstration (currently exercised by tests, not the +HTTP serving path). Promote a helper from a model package up into `core/` only once a +second model actually needs it. diff --git a/vllm_omni/experimental/fullduplex/__init__.py b/vllm_omni/experimental/fullduplex/__init__.py new file mode 100644 index 000000000..0615b7681 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/__init__.py @@ -0,0 +1,16 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from vllm_omni.experimental.fullduplex.core.adapter import DuplexAdapter, DuplexCapability, OutputChunk +from vllm_omni.experimental.fullduplex.core.runtime import DuplexRuntime +from vllm_omni.experimental.fullduplex.core.session import DuplexSession, DuplexSessionConfig, DuplexState + +__all__ = [ + "DuplexAdapter", + "DuplexCapability", + "DuplexRuntime", + "DuplexSession", + "DuplexSessionConfig", + "DuplexState", + "OutputChunk", +] diff --git a/vllm_omni/experimental/fullduplex/core/__init__.py b/vllm_omni/experimental/fullduplex/core/__init__.py new file mode 100644 index 000000000..208f01a7c --- /dev/null +++ b/vllm_omni/experimental/fullduplex/core/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project diff --git a/vllm_omni/experimental/fullduplex/core/adapter.py b/vllm_omni/experimental/fullduplex/core/adapter.py new file mode 100644 index 000000000..3fe400d49 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/core/adapter.py @@ -0,0 +1,43 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any + +from vllm_omni.experimental.fullduplex.core.session import DuplexSession + + +@dataclass +class DuplexCapability: + input_modalities: frozenset[str] + output_modalities: frozenset[str] + + proactive: bool = False + + +@dataclass +class OutputChunk: + modality: str + data: Any + + +class DuplexAdapter(ABC): + @abstractmethod + def capabilities(self) -> DuplexCapability: ... + + @abstractmethod + async def on_input(self, session: DuplexSession, modality: str, data: Any) -> None: ... + + @abstractmethod + def respond(self, session: DuplexSession) -> AsyncIterator[OutputChunk]: ... + + def should_respond(self, session: DuplexSession) -> bool: + return True + + async def on_barge_in(self, session: DuplexSession) -> None: ... + + async def on_playback_ack(self, session: DuplexSession, cursor: int) -> None: ... diff --git a/vllm_omni/experimental/fullduplex/core/protocol.py b/vllm_omni/experimental/fullduplex/core/protocol.py new file mode 100644 index 000000000..0056e1aba --- /dev/null +++ b/vllm_omni/experimental/fullduplex/core/protocol.py @@ -0,0 +1,40 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from typing import Any + +INPUT_APPEND = "input.append" +INPUT_COMMIT = "input.commit" +RESPONSE_CREATE = "response.create" +RESPONSE_CANCEL = "response.cancel" +PLAYBACK_ACK = "playback.ack" +CLOSE = "close" + + +RESPONSE_CREATED = "response.created" +RESPONSE_DELTA = "response.delta" +RESPONSE_DONE = "response.done" +RESPONSE_CANCELLED = "response.cancelled" +ERROR = "error" + + +def created(response_index: int) -> dict[str, Any]: + return {"type": RESPONSE_CREATED, "response_index": response_index} + + +def delta(response_index: int, modality: str, data: Any) -> dict[str, Any]: + return {"type": RESPONSE_DELTA, "response_index": response_index, "modality": modality, "data": data} + + +def done(response_index: int) -> dict[str, Any]: + return {"type": RESPONSE_DONE, "response_index": response_index} + + +def cancelled(response_index: int) -> dict[str, Any]: + return {"type": RESPONSE_CANCELLED, "response_index": response_index} + + +def error(message: str) -> dict[str, Any]: + return {"type": ERROR, "message": message} diff --git a/vllm_omni/experimental/fullduplex/core/runtime.py b/vllm_omni/experimental/fullduplex/core/runtime.py new file mode 100644 index 000000000..86ca011ef --- /dev/null +++ b/vllm_omni/experimental/fullduplex/core/runtime.py @@ -0,0 +1,86 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import asyncio +import contextlib +from collections.abc import AsyncIterator, Awaitable, Callable + +from vllm_omni.experimental.fullduplex.core import protocol as ev +from vllm_omni.experimental.fullduplex.core.adapter import DuplexAdapter +from vllm_omni.experimental.fullduplex.core.session import DuplexSession, DuplexState + +Emit = Callable[[dict], Awaitable[None]] + + +class DuplexRuntime: + def __init__(self, session: DuplexSession, adapter: DuplexAdapter) -> None: + self.session = session + self.adapter = adapter + self._capabilities = adapter.capabilities() + + async def run(self, inputs: AsyncIterator[dict], emit: Emit) -> None: + task: asyncio.Task | None = None + async for event in inputs: + etype = event.get("type") + if etype == ev.INPUT_APPEND: + modality = event.get("modality", "") + if modality not in self._capabilities.input_modalities: + await emit(ev.error(f"unsupported input modality: {modality}")) + continue + await self.adapter.on_input(self.session, modality, event.get("data")) + self.session.state = DuplexState.LISTENING + if self.session.config.proactive and self.adapter.should_respond(self.session): + task = await self._start_response(task, emit) + elif etype in (ev.INPUT_COMMIT, ev.RESPONSE_CREATE): + if self.adapter.should_respond(self.session): + task = await self._start_response(task, emit) + elif etype == ev.RESPONSE_CANCEL: + await self._barge_in() + await self._cancel(task) + task = None + await emit(ev.cancelled(self.session.response_index)) + elif etype == ev.PLAYBACK_ACK: + await self.adapter.on_playback_ack(self.session, int(event.get("cursor", 0))) + elif etype == ev.CLOSE: + break + if task is not None and not task.done(): + await task + self.session.state = DuplexState.CLOSED + + async def _start_response(self, prev: asyncio.Task | None, emit: Emit) -> asyncio.Task: + # A new response supersedes any in-flight one. Mark the old response stale and + # cancel it (barge-in) instead of awaiting its full stream — otherwise the input + # loop would block until the old response finishes and could not service a + # cancel / barge-in event in time, breaking the full-duplex contract. + if prev is not None and not prev.done(): + await self._barge_in() + await self._cancel(prev) + return asyncio.create_task(self._respond(emit)) + + @staticmethod + async def _cancel(task: asyncio.Task | None) -> None: + if task is not None and not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + async def _respond(self, emit: Emit) -> None: + response_index, epoch = self.session.begin_response() + await emit(ev.created(response_index)) + try: + async for chunk in self.adapter.respond(self.session): + if self.session.is_stale(epoch): + return + await emit(ev.delta(response_index, chunk.modality, chunk.data)) + except Exception as err: + await emit(ev.error(f"response failed: {err}")) + finally: + if not self.session.is_stale(epoch): + await emit(ev.done(response_index)) + self.session.end_response() + + async def _barge_in(self) -> None: + self.session.barge_in() + await self.adapter.on_barge_in(self.session) diff --git a/vllm_omni/experimental/fullduplex/core/session.py b/vllm_omni/experimental/fullduplex/core/session.py new file mode 100644 index 000000000..844f5a4b6 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/core/session.py @@ -0,0 +1,52 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field + + +class DuplexState(enum.Enum): + IDLE = "idle" + LISTENING = "listening" + RESPONDING = "responding" + CLOSED = "closed" + + +@dataclass +class DuplexSessionConfig: + input_modalities: tuple[str, ...] = ("audio",) + output_modalities: tuple[str, ...] = ("audio",) + sample_rate: int = 24000 + + proactive: bool = False + + +@dataclass +class DuplexSession: + session_id: str + config: DuplexSessionConfig = field(default_factory=DuplexSessionConfig) + state: DuplexState = DuplexState.IDLE + + epoch: int = 0 + response_index: int = 0 + + playback_cursor: int = 0 + + def begin_response(self) -> tuple[int, int]: + self.response_index += 1 + self.state = DuplexState.RESPONDING + return self.response_index, self.epoch + + def end_response(self) -> None: + if self.state is DuplexState.RESPONDING: + self.state = DuplexState.LISTENING + + def barge_in(self) -> int: + self.epoch += 1 + self.state = DuplexState.LISTENING + return self.epoch + + def is_stale(self, epoch: int) -> bool: + return epoch != self.epoch diff --git a/vllm_omni/experimental/fullduplex/joyvl/__init__.py b/vllm_omni/experimental/fullduplex/joyvl/__init__.py new file mode 100644 index 000000000..d3429ef82 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/__init__.py @@ -0,0 +1,19 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import ( + Action, + ParsedAction, + parse_action, +) +from vllm_omni.experimental.fullduplex.joyvl.serving.config import InteractionConfig +from vllm_omni.experimental.fullduplex.joyvl.serving.session import InteractionSession, StepResult + +__all__ = [ + "Action", + "InteractionConfig", + "InteractionSession", + "ParsedAction", + "StepResult", + "parse_action", +] diff --git a/vllm_omni/experimental/fullduplex/joyvl/adapter.py b/vllm_omni/experimental/fullduplex/joyvl/adapter.py new file mode 100644 index 000000000..16e220c29 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/adapter.py @@ -0,0 +1,72 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from collections.abc import AsyncIterator, Awaitable, Callable +from typing import Any + +from vllm_omni.experimental.fullduplex.core.adapter import DuplexAdapter, DuplexCapability, OutputChunk +from vllm_omni.experimental.fullduplex.core.session import DuplexSession +from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import DelegationBridge, StubDelegationBridge +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import Action +from vllm_omni.experimental.fullduplex.joyvl.decision.policy import JoyVLPolicy, sample_frames + +GenerateFn = Callable[[list[dict[str, Any]]], Awaitable[str]] + + +class JoyVLDuplexAdapter(DuplexAdapter): + def __init__( + self, + generate: GenerateFn, + *, + persona: str = "default", + num_frames: int = 4, + chunk_frames: int = 200, + frame_seconds: float = 1.0, + delegation: DelegationBridge | None = None, + ) -> None: + self._generate = generate + self._policy = JoyVLPolicy( + persona=persona, + num_frames=num_frames, + chunk_frames=chunk_frames, + frame_seconds=frame_seconds, + delegation=delegation or StubDelegationBridge(), + ) + self._frames: list[str] = [] + self._pending_query: str | None = None + + def capabilities(self) -> DuplexCapability: + return DuplexCapability(frozenset({"video", "text"}), frozenset({"text"}), proactive=True) + + async def on_input(self, session: DuplexSession, modality: str, data: Any) -> None: + if modality == "video": + self._policy.tick() + self._frames.append(data) + elif modality == "text": + self._pending_query = data + + def should_respond(self, session: DuplexSession) -> bool: + return bool(self._frames) + + async def respond(self, session: DuplexSession) -> AsyncIterator[OutputChunk]: + policy = self._policy + await policy.fold_delegations() + policy.set_query(self._pending_query) + self._pending_query = None + + parts = [{"type": "image_url", "image_url": {"url": f}} for f in sample_frames(self._frames, policy.num_frames)] + messages, _ = policy.build_messages(parts) + action = policy.commit(await self._generate(messages)) + + frames = [(str(i), f) for i, f in enumerate(self._frames)] + await policy.submit_if_delegate(action, frames) + if action.action is not Action.SILENCE and action.text: + yield OutputChunk("text", action.text) + if policy.needs_flush(): + await policy.flush(frames) + self._frames.clear() + + async def on_barge_in(self, session: DuplexSession) -> None: + self._pending_query = None diff --git a/vllm_omni/experimental/fullduplex/joyvl/bridges/__init__.py b/vllm_omni/experimental/fullduplex/joyvl/bridges/__init__.py new file mode 100644 index 000000000..208f01a7c --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/bridges/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project diff --git a/vllm_omni/experimental/fullduplex/joyvl/bridges/backend.py b/vllm_omni/experimental/fullduplex/joyvl/bridges/backend.py new file mode 100644 index 000000000..1aa2d57d2 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/bridges/backend.py @@ -0,0 +1,60 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from typing import Any, Protocol + +from openai import AsyncOpenAI + + +class ModelBackend(Protocol): + async def generate( + self, + messages: list[dict[str, Any]], + *, + max_tokens: int, + temperature: float, + top_p: float, + extra_body: dict[str, Any] | None = None, + ) -> tuple[str, dict[str, Any] | None]: ... + + async def aclose(self) -> None: ... + + +class OpenAIBackend: + def __init__( + self, + base_url: str, + model: str, + api_key: str = "EMPTY", + timeout: float = 300.0, + ) -> None: + self.model = model + self._client = AsyncOpenAI(base_url=base_url, api_key=api_key, timeout=timeout) + + async def generate( + self, + messages: list[dict[str, Any]], + *, + max_tokens: int, + temperature: float, + top_p: float, + extra_body: dict[str, Any] | None = None, + ) -> tuple[str, dict[str, Any] | None]: + response = await self._client.chat.completions.create( + model=self.model, + messages=messages, + max_tokens=max_tokens, + temperature=temperature, + top_p=top_p, + extra_body=extra_body or {}, + ) + if not response.choices: + return "", None + text = response.choices[0].message.content or "" + usage = response.usage.model_dump() if getattr(response, "usage", None) else None + return text, usage + + async def aclose(self) -> None: + await self._client.close() diff --git a/vllm_omni/experimental/fullduplex/joyvl/bridges/delegation.py b/vllm_omni/experimental/fullduplex/joyvl/bridges/delegation.py new file mode 100644 index 000000000..c61ca474b --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/bridges/delegation.py @@ -0,0 +1,389 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from typing import Any, Protocol + +import httpx +from openai import AsyncOpenAI + +from vllm_omni.experimental.fullduplex.joyvl.decision.prompts import DELEGATION_SOLVER_PROMPT + + +def _cancel_task(tasks: dict[str, asyncio.Task], task_id: str) -> None: + task = tasks.pop(task_id, None) + if task is not None and not task.done(): + task.cancel() + + +async def _aclose_tasks(tasks: dict[str, asyncio.Task]) -> None: + pending = list(tasks.values()) + for task in pending: + if not task.done(): + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + tasks.clear() + + +@dataclass +class DelegationResult: + task_id: str + status: str + digest: str = "" + media: str = "" + + @property + def is_ready(self) -> bool: + return self.status == "ready" + + +class DelegationBridge(Protocol): + async def submit(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: ... + + async def poll(self, task_id: str) -> DelegationResult: ... + + def cancel(self, task_id: str) -> None: + """Drop a task and cancel its background work if still running. + + Called when a session is reset/evicted so finished-but-unpolled tasks do not + accumulate on a long-lived shared bridge. + """ + ... + + async def aclose(self) -> None: + """Cancel+await all pending tasks and close any held client (server shutdown).""" + ... + + +class StubDelegationBridge: + def __init__(self, ready_after_ticks: int = 2) -> None: + self._ready_after = max(1, ready_after_ticks) + self._tasks: dict[str, dict[str, Any]] = {} + self._counter = 0 + + async def submit(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: + self._counter += 1 + task_id = f"deleg-{self._counter}" + self._tasks[task_id] = {"question": question, "polls": 0} + return task_id + + async def poll(self, task_id: str) -> DelegationResult: + task = self._tasks.get(task_id) + if task is None: + return DelegationResult(task_id, "error", "unknown task") + task["polls"] += 1 + if task["polls"] < self._ready_after: + return DelegationResult(task_id, "pending") + question = task["question"] + return DelegationResult( + task_id, + "ready", + digest=f"(background result for: {question}) — stub digest; wire a real brain here.", + ) + + def cancel(self, task_id: str) -> None: + self._tasks.pop(task_id, None) + + async def aclose(self) -> None: + self._tasks.clear() + + +class OpenAIDelegationBridge: + """Hand a delegated question to a stronger background model over an OpenAI-compatible API. + + Non-blocking by design: ``submit`` launches the request as a background task and returns + immediately; ``poll`` only checks whether that task has finished, so the per-tick interaction + loop never stalls waiting on the slower background brain. + """ + + def __init__( + self, + base_url: str, + model: str, + api_key: str = "EMPTY", + *, + max_frames: int = 8, + max_tokens: int = 512, + temperature: float = 0.2, + timeout: float = 120.0, + system_prompt: str = DELEGATION_SOLVER_PROMPT, + ) -> None: + self.model = model + self._client = AsyncOpenAI(base_url=base_url, api_key=api_key, timeout=timeout) + self._max_frames = max_frames + self._max_tokens = max_tokens + self._temperature = temperature + self._system_prompt = system_prompt + self._tasks: dict[str, asyncio.Task] = {} + self._counter = 0 + + async def submit(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: + self._counter += 1 + task_id = f"deleg-{self._counter}" + self._tasks[task_id] = asyncio.create_task(self._solve(question, note, frames)) + return task_id + + async def poll(self, task_id: str) -> DelegationResult: + task = self._tasks.get(task_id) + if task is None: + return DelegationResult(task_id, "error", "unknown task") + if not task.done(): + return DelegationResult(task_id, "pending") + self._tasks.pop(task_id, None) + try: + digest = task.result() + except Exception as exc: # surface any background failure as an error result, never crash the loop + return DelegationResult(task_id, "error", str(exc)) + return DelegationResult(task_id, "ready", digest=digest) + + async def _solve(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: + content: list[dict[str, Any]] = [] + for time_range, data_url in frames[-self._max_frames :]: + content.append({"type": "text", "text": f"<{time_range}>"}) + content.append({"type": "image_url", "image_url": {"url": data_url}}) + content.append({"type": "text", "text": question or note or "Answer the user's question."}) + messages = [ + {"role": "system", "content": self._system_prompt}, + {"role": "user", "content": content}, + ] + return await self._complete(messages) + + async def _complete(self, messages: list[dict[str, Any]]) -> str: + response = await self._client.chat.completions.create( + model=self.model, + messages=messages, + max_tokens=self._max_tokens, + temperature=self._temperature, + ) + if not response.choices: + return "" + return (response.choices[0].message.content or "").strip() + + def cancel(self, task_id: str) -> None: + _cancel_task(self._tasks, task_id) + + async def aclose(self) -> None: + await _aclose_tasks(self._tasks) + await self._client.close() + + +class ImageGenDelegationBridge: + """Delegate an image-generation request to a text-to-image model over the + OpenAI-compatible ``/v1/images/generations`` endpoint (e.g. vLLM-Omni Qwen-Image). + + Non-blocking: ``submit`` launches generation as a background task and returns at once; + ``poll`` only checks completion. The generated image is returned as a ``data:`` URL in + ``DelegationResult.media`` (surfaced to the client), while ``digest`` stays a short text + placeholder so the base64 image never enters the model's context. + """ + + def __init__( + self, + base_url: str, + *, + size: str = "1024x1024", + timeout: float = 300.0, + ) -> None: + self._base_url = base_url.rstrip("/") + self._size = size + self._timeout = timeout + self._tasks: dict[str, asyncio.Task] = {} + self._counter = 0 + + async def submit(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: + self._counter += 1 + task_id = f"deleg-{self._counter}" + self._tasks[task_id] = asyncio.create_task(self._generate(question or note)) + return task_id + + async def poll(self, task_id: str) -> DelegationResult: + task = self._tasks.get(task_id) + if task is None: + return DelegationResult(task_id, "error", "unknown task") + if not task.done(): + return DelegationResult(task_id, "pending") + self._tasks.pop(task_id, None) + try: + prompt, data_url = task.result() + except Exception as exc: # surface any background failure as an error result, never crash the loop + return DelegationResult(task_id, "error", str(exc)) + return DelegationResult(task_id, "ready", digest=f"(generated an image for: {prompt})", media=data_url) + + async def _generate(self, prompt: str) -> tuple[str, str]: + data = await self._call_images({"prompt": prompt, "size": self._size}) + items = data.get("data") or [] + b64 = items[0].get("b64_json") if items and isinstance(items[0], dict) else None + if not b64: + raise RuntimeError("image endpoint returned no image data") + return prompt, f"data:image/png;base64,{b64}" + + async def _call_images(self, payload: dict[str, Any]) -> dict[str, Any]: + async with httpx.AsyncClient(timeout=self._timeout) as client: + resp = await client.post(f"{self._base_url}/images/generations", json=payload) + resp.raise_for_status() + return resp.json() + + def cancel(self, task_id: str) -> None: + _cancel_task(self._tasks, task_id) + + async def aclose(self) -> None: + await _aclose_tasks(self._tasks) + + +def _extract_image_url(data: dict[str, Any]) -> str | None: + choices = data.get("choices") or [] + if not choices: + return None + content = (choices[0].get("message") or {}).get("content") + if isinstance(content, list): + for part in content: + if isinstance(part, dict) and part.get("image_url"): + return (part["image_url"] or {}).get("url") + if isinstance(content, str) and content.startswith("data:image/"): + return content + return None + + +class ImageEditDelegationBridge: + """Delegate "restyle / cartoonify what you currently see" to an image-editing model + (e.g. vLLM-Omni Qwen-Image-Edit) over the OpenAI-compatible ``/v1/chat/completions`` + endpoint, which takes an input image + instruction and returns the edited image. + + Conditions on the latest frame passed to ``submit`` — so it preserves the real + composition instead of re-imagining it. Non-blocking submit/poll; the edited image + rides in ``DelegationResult.media`` while ``digest`` stays a short text placeholder. + """ + + _DEFAULT_INSTRUCTION = ( + "Convert this image into a clean cartoon / anime illustration style. " + "Preserve the composition, subjects, poses, and overall layout." + ) + + def __init__(self, base_url: str, model: str, *, timeout: float = 300.0, instruction: str | None = None) -> None: + self._base_url = base_url.rstrip("/") + self._model = model + self._timeout = timeout + self._instruction = instruction or self._DEFAULT_INSTRUCTION + self._tasks: dict[str, asyncio.Task] = {} + self._counter = 0 + + async def submit(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: + self._counter += 1 + task_id = f"deleg-{self._counter}" + frame = frames[-1][1] if frames else None + self._tasks[task_id] = asyncio.create_task(self._edit(question or note, frame)) + return task_id + + async def poll(self, task_id: str) -> DelegationResult: + task = self._tasks.get(task_id) + if task is None: + return DelegationResult(task_id, "error", "unknown task") + if not task.done(): + return DelegationResult(task_id, "pending") + self._tasks.pop(task_id, None) + try: + label, data_url = task.result() + except Exception as exc: # surface any background failure as an error result, never crash the loop + return DelegationResult(task_id, "error", str(exc)) + return DelegationResult(task_id, "ready", digest=f"(restyled the current view: {label})", media=data_url) + + async def _edit(self, request_text: str, frame: str | None) -> tuple[str, str]: + if not frame: + raise RuntimeError("no frame available to restyle") + instruction = request_text or self._instruction + data = await self._call_chat(frame, instruction) + url = _extract_image_url(data) + if not url: + raise RuntimeError("image-edit endpoint returned no image") + return (request_text or "cartoon"), url + + async def _call_chat(self, frame: str, instruction: str) -> dict[str, Any]: + body = { + "model": self._model, + "messages": [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": frame}}, + {"type": "text", "text": instruction}, + ], + } + ], + } + async with httpx.AsyncClient(timeout=self._timeout) as client: + resp = await client.post(f"{self._base_url}/chat/completions", json=body) + resp.raise_for_status() + return resp.json() + + def cancel(self, task_id: str) -> None: + _cancel_task(self._tasks, task_id) + + async def aclose(self) -> None: + await _aclose_tasks(self._tasks) + + +_EDIT_STYLE_KW = ("卡通", "漫画", "动漫", "cartoon", "anime", "风格", "restyle", "变成", "画成") +_EDIT_SCENE_KW = ("看到", "画面", "当前", "现在", "this", "what you see", "场景", "镜头") +_IMAGE_KW = ("画", "生成", "draw", "generate", "picture", "image", "图", "render") + + +class RoutingDelegationBridge: + """Dispatch each delegation to the right backend by inspecting the request text. + + - "restyle / cartoonify what you currently see" -> image-edit (conditions on the frame) + - "draw / generate an image of X" -> text-to-image + - everything else (hard questions) -> the chat / VL background brain + + Wraps sub-bridges that each implement the DelegationBridge protocol; any may be None. + Routing is heuristic and demo-grade, not a trained classifier. + """ + + def __init__(self, *, chat: Any = None, image: Any = None, edit: Any = None) -> None: + self._chat = chat + self._image = image + self._edit = edit + self._route: dict[str, tuple[Any, str]] = {} + self._counter = 0 + + def _pick(self, question: str, note: str) -> Any: + text = f"{question} {note}".lower() + if self._edit and any(k in text for k in _EDIT_STYLE_KW) and any(k in text for k in _EDIT_SCENE_KW): + return self._edit + if self._image and any(k in text for k in _IMAGE_KW): + return self._image + return self._chat or self._image or self._edit + + async def submit(self, question: str, note: str, frames: list[tuple[str, str]]) -> str: + bridge = self._pick(question, note) + if bridge is None: + raise RuntimeError("no delegation backend configured") + self._counter += 1 + key = f"route-{self._counter}" + self._route[key] = (bridge, await bridge.submit(question, note, frames)) + return key + + async def poll(self, task_id: str) -> DelegationResult: + entry = self._route.get(task_id) + if entry is None: + return DelegationResult(task_id, "error", "unknown task") + bridge, inner_id = entry + res = await bridge.poll(inner_id) + if res.status in ("ready", "error"): + self._route.pop(task_id, None) + return DelegationResult(task_id, res.status, res.digest, res.media) + + def cancel(self, task_id: str) -> None: + entry = self._route.pop(task_id, None) + if entry is not None: + bridge, inner_id = entry + bridge.cancel(inner_id) + + async def aclose(self) -> None: + self._route.clear() + for bridge in (self._chat, self._image, self._edit): + if bridge is not None: + await bridge.aclose() diff --git a/vllm_omni/experimental/fullduplex/joyvl/decision/__init__.py b/vllm_omni/experimental/fullduplex/joyvl/decision/__init__.py new file mode 100644 index 000000000..208f01a7c --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/decision/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project diff --git a/vllm_omni/experimental/fullduplex/joyvl/decision/output_parser.py b/vllm_omni/experimental/fullduplex/joyvl/decision/output_parser.py new file mode 100644 index 000000000..cea17fd1c --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/decision/output_parser.py @@ -0,0 +1,75 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import enum +from dataclasses import dataclass + +SILENCE_TOKEN = "" +RESPONSE_TOKEN = "" +DELEGATION_TAG = "" +_DELEGATION_MARKERS = ("", "") + + +class Action(enum.Enum): + SILENCE = "silence" + RESPONSE = "response" + DELEGATE = "delegate" + + +@dataclass +class ParsedAction: + action: Action + text: str = "" + delegated_question: str | None = None + raw: str = "" + + @property + def spoke(self) -> bool: + return self.action is not Action.SILENCE + + +def parse_action(raw: str) -> ParsedAction: + text = (raw or "").replace("\r\n", "\n").replace("\r", "\n").strip() + if not text: + return ParsedAction(Action.SILENCE, raw=raw or "") + + marker = _first_marker(text) + if marker != -1 and text[marker:].startswith(SILENCE_TOKEN): + return ParsedAction(Action.SILENCE, raw=raw) + + if marker == -1: + payload = _collapse(text.splitlines()[0]) + else: + after = text[marker + len(RESPONSE_TOKEN) :].strip() + payload = _collapse(after.splitlines()[0]) if after else "" + + for tag in _DELEGATION_MARKERS: + idx = payload.find(tag) + if idx != -1: + return ParsedAction( + Action.DELEGATE, + text=_collapse(payload[:idx]), + delegated_question=_collapse(payload[idx + len(tag) :]) or None, + raw=raw, + ) + + return ParsedAction(Action.RESPONSE, text=payload, raw=raw) + + +def to_token_form(action: ParsedAction) -> str: + if action.action is Action.SILENCE: + return SILENCE_TOKEN + if action.action is Action.DELEGATE: + return f"{RESPONSE_TOKEN} {action.text} {DELEGATION_TAG} {action.delegated_question or ''}".strip() + return f"{RESPONSE_TOKEN} {action.text}".strip() + + +def _first_marker(text: str) -> int: + present = [i for i in (text.find(SILENCE_TOKEN), text.find(RESPONSE_TOKEN)) if i != -1] + return min(present) if present else -1 + + +def _collapse(text: str) -> str: + return " ".join(text.split()) diff --git a/vllm_omni/experimental/fullduplex/joyvl/decision/policy.py b/vllm_omni/experimental/fullduplex/joyvl/decision/policy.py new file mode 100644 index 000000000..7ebe87d55 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/decision/policy.py @@ -0,0 +1,128 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from difflib import SequenceMatcher +from typing import Any + +from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import DelegationBridge +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import Action, ParsedAction, parse_action +from vllm_omni.experimental.fullduplex.joyvl.decision.prompts import SYSTEM_PROMPTS, USER_QUERY_HEADER +from vllm_omni.experimental.fullduplex.joyvl.memory.brain import InteractionBrain +from vllm_omni.experimental.fullduplex.joyvl.memory.memory import Summarizer + + +def sample_frames(frames: list[str], num_frames: int) -> list[str]: + n = len(frames) + if num_frames <= 0 or n <= num_frames: + return list(frames) + stride = max(1, n // num_frames) + idx = [i * stride for i in range(num_frames - 1)] + [n - 1] + return [frames[i] for i in idx] + + +class JoyVLPolicy: + def __init__( + self, + *, + persona: str = "default", + system_prompt: str | None = None, + num_frames: int = 4, + summarizer: Summarizer | None = None, + delegation: DelegationBridge | None = None, + chunk_frames: int = 16, + long_term_every_n_chunks: int = 5, + keep_qa_history: bool = True, + frame_seconds: float = 1.0, + enable_delegation: bool = True, + response_dedup_threshold: float = 1.0, + ) -> None: + self.response_dedup_threshold = response_dedup_threshold + self.brain = InteractionBrain( + summarizer=summarizer, + delegation=delegation, + chunk_frames=chunk_frames, + long_term_every_n_chunks=long_term_every_n_chunks, + keep_qa_history=keep_qa_history, + frame_seconds=frame_seconds, + enable_delegation=enable_delegation, + ) + self.system_prompt = system_prompt or SYSTEM_PROMPTS.get(persona, SYSTEM_PROMPTS["default"]) + self.num_frames = num_frames + + def tick(self, n: int = 1) -> None: + self.brain.tick(n) + + def observe(self, time_range: str, data_url: str) -> None: + self.brain.observe(time_range, data_url) + + @property + def working_frames(self) -> list[tuple[str, str]]: + return self.brain.working_frames + + def take_working_frames(self) -> list[tuple[str, str]]: + return self.brain.take_working_frames() + + def set_query(self, query: str | None) -> bool: + return self.brain.update_query(query) + + def should_respond(self) -> bool: + return True + + def user_content(self, frame_parts: list[dict[str, Any]]) -> list[dict[str, Any]]: + content: list[dict[str, Any]] = [] + prefix = self.brain.build_prefix() + if prefix: + content.append({"type": "text", "text": prefix}) + content.extend(frame_parts) + if self.brain.current_query and self.brain.query_in_current_chunk: + content.append({"type": "text", "text": f"{USER_QUERY_HEADER}\n{self.brain.current_query}"}) + return content + + def build_messages(self, frame_parts: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, Any]]: + user = {"role": "user", "content": self.user_content(frame_parts)} + return [{"role": "system", "content": self.system_prompt}, user], user + + def _is_repeat(self, text: str) -> bool: + last = self.brain.last_response_text + if not last: + return False + if text == last: + return True + if self.response_dedup_threshold >= 1.0: + return False + similarity = SequenceMatcher(None, text.strip().lower(), last.strip().lower()).ratio() + return similarity >= self.response_dedup_threshold + + def commit(self, response_text: str) -> ParsedAction: + action = parse_action(response_text) + if action.action is Action.SILENCE: + self.brain.last_response_text = "" + return action + if action.action is Action.RESPONSE and action.text and self._is_repeat(action.text): + return ParsedAction(Action.SILENCE, raw="") + self.brain.last_response_text = action.text + if self.brain.current_query: + self.brain.record_response(action.text) + return action + + async def fold_delegations(self) -> dict[str, Any] | None: + return await self.brain.fold_delegations() + + async def submit_if_delegate( + self, action: ParsedAction, frames: list[tuple[str, str]] | None = None + ) -> dict | None: + return await self.brain.submit_delegation(action, frames or []) + + def needs_flush(self) -> bool: + return self.brain.should_flush() + + def close_chunk(self) -> int: + return self.brain.close_chunk() + + async def consolidate(self, chunk_index: int, frames: list[tuple[str, str]] | None = None) -> None: + await self.brain.consolidate(chunk_index, frames or []) + + async def flush(self, frames: list[tuple[str, str]] | None = None) -> None: + await self.brain.flush(frames or []) diff --git a/vllm_omni/experimental/fullduplex/joyvl/decision/prompts.py b/vllm_omni/experimental/fullduplex/joyvl/decision/prompts.py new file mode 100644 index 000000000..b7fa95f0a --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/decision/prompts.py @@ -0,0 +1,182 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +DEFAULT_SYSTEM_PROMPT = """You are a real-time video streaming assistant observing a continuous camera feed frame by frame. The last frame represents the current moment. +## Action Format +At every inference step you MUST choose exactly one of the following three actions: +**Stay silent** — output ONLY: + +Choose this when nothing noteworthy has changed in the scene, no user query is pending, or there is nothing useful to say. +**Speak** — output the token followed by a concise reply: + Your reply here. +Choose this when you observe something worth reporting or a significant state change, or when you can answer a user question based on available evidence. + +**Delegate** — when a question is too hard or error-prone to answer reliably yourself, speak a brief note that you're delegating, then hand the question to the background solver: + Brief note that you're delegating. """.strip() + +SILENT_SYSTEM_PROMPT = """You are a real-time video streaming assistant observing a continuous camera feed frame by frame. The last frame represents the current moment. +## Action Format +At every inference step you MUST choose exactly one of the following three actions: +**Stay silent** — output ONLY: + +Choose this when nothing noteworthy has changed in the scene, no user query is pending, or there is nothing useful to say. +**Speak** — output the token followed by a concise reply: + Your reply here. +Choose this when you observe something worth reporting or a significant state change, or when you can answer a user question based on available evidence. + +Do NOT delegate or defer questions to other models. Answer user queries directly based on everything you have seen.""".strip() + +TALKATIVE_SYSTEM_PROMPT = """You are a real-time video streaming assistant observing a continuous camera feed frame by frame. The last frame represents the current moment. +## Action Format +At every inference step you MUST choose exactly one of the following three actions: +**Stay silent** — output ONLY: + +Choose this when nothing noteworthy has changed in the scene, no user query is pending, or there is nothing useful to say. +**Speak** — output the token followed by a concise reply: + Your reply here. +Choose this when you observe something worth reporting or a significant state change, or when you can answer a user question based on available evidence. + +**Delegate** — when a question is too hard or error-prone to answer reliably yourself, speak a brief note that you're delegating, then hand the question to the background solver: + Brief note that you're delegating. + +## Style +Proactively speak whenever you observe a meaningful event, state change, or anomaly — do not wait for the user to ask. However, avoid repeating information you have already reported. When the scene shows no obvious change, choose silence.""".strip() + +SYSTEM_PROMPTS = { + "default": DEFAULT_SYSTEM_PROMPT, + "silent": SILENT_SYSTEM_PROMPT, + "talkative": TALKATIVE_SYSTEM_PROMPT, +} + + +DELEGATION_SOLVER_PROMPT = ( + "You are the background solver for a real-time video agent. The foreground agent delegated a " + "question it could not answer quickly and reliably. Use the attached recent frames to answer it. " + "Reply with a single concise, factual digest the foreground agent can relay to the user — no " + "preamble, no restating the question. If the frames are insufficient, say so plainly." +) + + +USER_QUERY_HEADER = "[User Query (IMPORTANT — follow this instruction)]" + +VIDEO_HISTORY_HEADER = ( + "[Video History]\n" + "The following are summaries of earlier video segments you can no longer see. " + "Use them as background context, but always prioritize the current visual frames " + "and the User Query below when making decisions.\n" + "IMPORTANT: These summaries are written by an external system in a descriptive style. " + "Do NOT imitate their writing style in your responses.\n" +) + +QA_HISTORY_HEADER = "[Q&A History]\nThe following are previous queries and the system's responses.\n\n" + +QA_QUERY_LABEL = "Query" +QA_RESPONSE_LABEL = "Response" + + +EMPTY_CHUNK_SUMMARY = "在帧 {frame_range} 中未观察到明显的视觉变化。" + +MID_TERM_SUMMARY_PROMPT = """\ +你正在为一个长时间运行的视频智能体编写中期记忆。用户消息包含第 {chunk_index} 个片段的带时间戳关键帧,覆盖时间范围 {frame_range};每张图片前标注其采样时间段。这些帧在本次之后不再可用,因此你的段落需要保留下游模型在没有视觉信息时进行回忆、推理和回答后续问题所需的关键证据。 + +【输出格式】 +- 撰写一个事实性段落{length_instruction}。信息密集时充分利用空间;接近静止或几乎无新信息时(如长时间静止画面、近乎重复的帧),用 1-2 句简短记录即可,不要凑长度。 +- 身份、文字、数值不确定时不要猜测,用"疑似 X"、"约 X"或"不可读"等方式标注;不可读时仍保留可观察到的类别、外观、动作或位置特征。 +- 段落主体使用中文叙述;画面里能看到的英文(术语、人名、作品名、短语、年代标注等)每次出现都写成"中文译名(English 原文)",不许只留一种,不许半中半英,画面没出现的英文不要补。 + +【需保留的信息(按优先级从高到低)】 + +最高优先级 — 任务连续性必需,必须保留: +- 片段结束时的交接状态:屏幕上剩余什么、任务推进到哪一步、哪些物体仍然可用、哪些动作未完成或仍在进行。 +- 不可逆事件与状态变化:物体的出现、消失、移动、开合、组合、分离、转移;重要操作的来源、目的地与结果状态。 +- 中间结果、任务进度、部分完成或被中断的步骤。 +- 因果链:当一个动作引发另一个动作时,保持前后关系,不要列成孤立片段。 + +高优先级 — 信息价值高且难以重建: +- 可读的文字、标签、标题、数字、测量值、计数、日期、标识符、规格。多值同时出现时保留"项目→值"的映射,不要列出脱离上下文的数字。 +- 在本片段内首次出现或明显异常的事件、实体或行为(仅按本片段判断,不需考虑跨片段历史)。 +- 重要物体的最终位置与关键空间关系。 + +中优先级 — 背景与陪衬: +- 出现的人物、身体部位、工具、容器、食材、设备等实体清单。 +- 场景布局与持续存在的背景:首次出现时记录一次,发生有意义变化时更新,未变化则不重复。 +- 重要事件与转换的稀疏时间锚点。 + +【撰写规则】 + +时间引用: +- 整段最多使用 3-5 个显式时间锚点,且仅在能改善事件区分或转换识别时使用。优先采用约 {preferred_time_span} 的合并范围,一个连续动作阶段最多使用一个显式时间范围。例如写"从 349s 到 364s,吉他手持续弹奏、抬起左手并鞠躬",而不是 349s-350s、351s-352s、353s-355s 分别写。 +- 图片前的时间标注是证据锚点,不要逐个复制到输出中;相邻帧中重复或连续的动作合并为一个范围,并在其中描述可见变化。 + +详略与抽象层级: +- 详写最高与高优先级信息;略写稳定背景与重复动作。 +- 保留具体步骤,不要用"准备继续"、"做相关操作"等空话替代可见动作。 +- 镜头切换、机位移动或视角变化本身无需逐次提及;仅当它们揭示新元素、改变可读信息或影响动作连续性时才记录。 +- 当场景包含结构化信息(表格、菜单、表单、HUD、字幕等)时,保留字段-值对应关系,而不仅概述主题。 +- 这是记忆而非字幕:均衡保留状态、实体、动作、文字与数值,而不是只偏重叙事、新颖性或运动。 + +事实性约束: +- 仅使用所提供关键帧中直接可见的细节;不要补全画面外的内容或推测后续发展。 +- 按关键帧的时间顺序描述,但不要逐一叙述每个采样间隔。 +- 避免泛泛的总结性短语、元评论(如"本片段展示了……")和重复句式。 + +仅输出该段落。""" + +LONG_TERM_COMPRESS_PROMPT = """\ +你正在为一个长时间运行的视频智能体将多段中期记忆压缩为长期记忆。时间段 {merged_range} 的原始帧将不再可用,因此输出需要保留下游模型在此之后进行回忆、推理和回答后续问题所需的关键证据。 + +需要合并的中期摘要: +{summaries_text} + +【任务定位】 +这是压缩任务,不是简单合并。需要按优先级取舍以降低信息密度,而不是把所有内容堆在一起。输入由 N 段中期摘要组成,每段都有自己的期末交接状态;合并后只有**最后一段的期末状态**对下游有意义,前面各段的期末状态若已被后续事件覆盖(如"拿起又放下的杯子"、"打开又关闭的菜单"),应作为过程信息处理或省略。 + +【输出格式】 +- 撰写一个统一的事实性段落(不是按片段分开的摘要),在目标长度内{length_instruction}按优先级取舍。 +- 中期摘要中的不确定性标注("疑似 X"、"约 X"、"不可读")必须传递到长期记忆,不要在合并时被"洗"成确定表述。 + +【保留优先级】 + +最高优先级 — 必须保留: +- 合并期**结束时**的最终交接状态:屏幕上剩余什么、任务推进到哪一步、哪些物体仍然可用、哪些动作未完成或仍在进行。 +- 跨片段的不可逆事件与状态变化:物体的最终出现/消失、关键移动、组合、分离、转移;重要操作的来源、目的地与结果状态。 +- 可读的文字、标签、数字、测量值、计数、日期、标识符、规格,以及"项目→值"的映射关系。 +- 因果链:当一个动作引发后续动作时,保持依赖关系,不要扁平化为孤立事实。 + +高优先级 — 应当保留: +- 跨多个片段持续存在的实体(人物、关键工具、关键物体)及其演变。 +- 任务的整体进度结构与阶段性里程碑。 +- 在合并范围内首次出现或明显异常的事件。 +- 重要物体在合并期末的位置与关键空间关系。 + +可压缩 — 主动降低密度或丢弃: +- 已被后续事件覆写的中间状态——仅在该过程本身有信息价值时才保留。 +- 跨片段未变化的稳定背景,只在首次描述时记录一次。 +- 同一动作的多次相似重复,按下方"重复处理"规则归并。 +- 过程性的细微步骤,在更高层动作已被记录时可以省略。 + +【合并规则】 + +跨片段一致性: +- 同一实体在不同片段中名称不一致时(如"红色杯子" → "马克杯" → "容器"),统一为信息量最高、最具体的那个,并在首次出现时简要标注其他称呼。 +- 多个片段对同一事件描述冲突时,优先采信时间靠后或信息更具体的版本;如果无法判定,用"X 或 Y"并列保留,不要单方面选边。 +- 同一事件在相邻片段边界处被重复描述时,合并为一次记录。 + +重复处理: +- 当重复次数本身携带信息时(如"敲门 5 次"、"刷新页面 3 次"),保留次数。 +- 当重复是过程性持续时(如"持续搅拌"、"反复检查"),用一个时间范围加动作描述合并,例如"从 349s 到 364s 持续弹奏并多次抬手致意"。 +- 不要将多个不同的动作或状态变化压缩为模糊总结。 + +【时间引用】 +- 整段使用 3-7 个显式时间锚点(按 {merged_range} 跨度调整),仅在能改善事件区分或阶段切换识别时使用。 +- 不要保留源摘要的 标题,也不要复制源摘要中"在 s..."的连续时间戳节奏。 +- 连续或重复的相邻动作合并为一个范围,并在范围内描述可见变化。 + +【事实性约束】 +- 仅使用中期摘要中已有的内容;不要补全画面外信息或推测后续发展。 +- 保留具体步骤与可读数值,不要用"准备继续"、"做相关操作"等空话替代。 +- 避免泛泛的总结性短语、元评论(如"这段记忆包含……")和重复句式。 + +仅输出统一的叙述文本,不要输出其他内容。""" diff --git a/vllm_omni/experimental/fullduplex/joyvl/memory/__init__.py b/vllm_omni/experimental/fullduplex/joyvl/memory/__init__.py new file mode 100644 index 000000000..208f01a7c --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/memory/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project diff --git a/vllm_omni/experimental/fullduplex/joyvl/memory/brain.py b/vllm_omni/experimental/fullduplex/joyvl/memory/brain.py new file mode 100644 index 000000000..98625adc1 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/memory/brain.py @@ -0,0 +1,178 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import asyncio +from typing import Any + +from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import DelegationBridge +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import Action, ParsedAction +from vllm_omni.experimental.fullduplex.joyvl.memory.memory import ( + MidTermSummary, + QAEntry, + SessionMemory, + Summarizer, + build_memory_prefix, +) + + +class InteractionBrain: + def __init__( + self, + *, + summarizer: Summarizer | None = None, + delegation: DelegationBridge | None = None, + chunk_frames: int = 16, + long_term_every_n_chunks: int = 5, + keep_qa_history: bool = True, + frame_seconds: float = 1.0, + enable_delegation: bool = True, + ) -> None: + self._summarizer = summarizer + self._delegation = delegation + self._chunk_frames = chunk_frames + self._long_term_every_n = long_term_every_n_chunks + self._keep_qa_history = keep_qa_history + self._frame_seconds = frame_seconds + self._enable_delegation = enable_delegation + self.reset() + + def reset(self) -> None: + # Cancel any in-flight/finished-but-unpolled delegations so the shared bridge + # does not accumulate abandoned tasks when a session is reset or evicted. + if self._delegation is not None: + for task in getattr(self, "_pending_delegations", []): + self._delegation.cancel(task["task_id"]) + self.memory = SessionMemory() + self.current_query: str | None = None + self.query_time: str | None = None + self.query_in_current_chunk = False + self.frame_index = 0 + self.chunk_index = 1 + self._chunk_frame_count = 0 + self.working_frames: list[tuple[str, str]] = [] + self.last_response_text = "" + self.response_records: list[tuple[str, str]] = [] + self._pending_delegations: list[dict[str, str]] = [] + self._consolidate_lock = asyncio.Lock() + + def now(self) -> str: + return f"{self.frame_index * self._frame_seconds:.1f}s" + + def last_frame_time(self) -> str: + return f"{max(0, self.frame_index - 1) * self._frame_seconds:.1f}s" + + def tick(self, n: int = 1) -> None: + self.frame_index += n + self._chunk_frame_count += n + + def observe(self, time_range: str, data_url: str) -> None: + self.tick() + self.working_frames.append((time_range, data_url)) + + def take_working_frames(self) -> list[tuple[str, str]]: + frames = self.working_frames + self.working_frames = [] + return frames + + def should_flush(self) -> bool: + return self._chunk_frames > 0 and self._chunk_frame_count >= self._chunk_frames + + def update_query(self, query: str | None) -> bool: + q = (query or "").strip() + if not q or q == self.current_query: + return False + if self.current_query is not None: + self.archive_query() + self.current_query = q + self.query_time = self.now() + self.query_in_current_chunk = True + return True + + def archive_query(self) -> None: + if self.current_query and self.response_records: + self.memory.qa_history.append( + QAEntry( + self.current_query, + self.query_time or "", + list(self.response_records), + self.chunk_index, + ) + ) + self.response_records = [] + + def record_response(self, text: str) -> None: + if text and self.current_query: + self.response_records.append((self.last_frame_time(), text)) + + def build_prefix(self) -> str: + return build_memory_prefix( + self.memory, + current_query=self.current_query, + query_in_current_chunk=self.query_in_current_chunk, + keep_qa_history=self._keep_qa_history, + current_chunk_index=self.chunk_index, + ) + + def close_chunk(self) -> int: + self.archive_query() + closed = self.chunk_index + self.chunk_index += 1 + self._chunk_frame_count = 0 + self.query_in_current_chunk = False + return closed + + async def consolidate(self, chunk_index: int, frames: list[tuple[str, str]]) -> None: + if self._summarizer is None or not frames: + return + frame_range = f"{frames[0][0]}-{frames[-1][0]}" + async with self._consolidate_lock: + summary = await self._summarizer.summarize_chunk(chunk_index, frame_range, frames) + self.memory.mid_term_summaries.append(MidTermSummary(chunk_index, frame_range, summary)) + if len(self.memory.mid_term_summaries) >= self._long_term_every_n: + self.memory.long_term_memory = await self._summarizer.compress_to_long_term( + self.memory.long_term_memory, self.memory.mid_term_summaries + ) + self.memory.mid_term_summaries.clear() + + async def flush(self, frames: list[tuple[str, str]]) -> None: + await self.consolidate(self.close_chunk(), frames) + + async def submit_delegation(self, action: ParsedAction, frames: list[tuple[str, str]]) -> dict[str, Any] | None: + if action.action is not Action.DELEGATE or self._delegation is None or not self._enable_delegation: + return None + task_id = await self._delegation.submit(action.delegated_question or "", action.text, frames) + self._pending_delegations.append({"task_id": task_id, "question": action.delegated_question or ""}) + return {"task_id": task_id, "status": "submitted", "question": action.delegated_question} + + async def fold_delegations(self) -> dict[str, Any] | None: + if not self._pending_delegations or self._delegation is None: + return None + folded: dict[str, Any] | None = None + still_pending: list[dict[str, str]] = [] + for task in self._pending_delegations: + result = await self._delegation.poll(task["task_id"]) + if result.is_ready: + self.memory.qa_history.append( + QAEntry( + f"[delegated] {task['question']}", + self.query_time or "", + [("", result.digest)], + self.chunk_index, + ) + ) + folded = { + "task_id": result.task_id, + "status": "ready", + "question": task["question"], + "digest": result.digest, + } + if result.media: + folded["media"] = result.media + elif result.status == "error": + folded = {"task_id": result.task_id, "status": "error", "question": task["question"]} + else: + still_pending.append(task) + self._pending_delegations = still_pending + return folded diff --git a/vllm_omni/experimental/fullduplex/joyvl/memory/memory.py b/vllm_omni/experimental/fullduplex/joyvl/memory/memory.py new file mode 100644 index 000000000..92612cc89 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/memory/memory.py @@ -0,0 +1,180 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from vllm_omni.experimental.fullduplex.joyvl.bridges.backend import ModelBackend +from vllm_omni.experimental.fullduplex.joyvl.decision import prompts + + +@dataclass +class WorkingChunk: + messages: list[dict[str, Any]] = field(default_factory=list) + + +@dataclass +class MidTermSummary: + chunk_index: int + frame_range: str + summary_text: str + + +@dataclass +class QAEntry: + query: str + query_time: str + responses: list[tuple[str, str]] = field(default_factory=list) + archived_in_chunk: int = 0 + + +@dataclass +class SessionMemory: + long_term_memory: str = "" + mid_term_summaries: list[MidTermSummary] = field(default_factory=list) + qa_history: list[QAEntry] = field(default_factory=list) + + +def build_memory_prefix( + memory: SessionMemory, + *, + current_query: str | None, + query_in_current_chunk: bool, + keep_qa_history: bool, + current_chunk_index: int, +) -> str: + has_query = bool(current_query) + sections: list[str] = [] + + history_parts: list[str] = [] + if has_query and memory.long_term_memory: + history_parts.append(memory.long_term_memory) + for entry in memory.mid_term_summaries: + history_parts.append(f"<{entry.frame_range}>\n{entry.summary_text}") + if history_parts: + sections.append(prompts.VIDEO_HISTORY_HEADER + "\n\n".join(history_parts)) + + if keep_qa_history and has_query: + qa_entries = [e for e in memory.qa_history if e.archived_in_chunk < current_chunk_index] + if qa_entries: + lines = [] + for idx, entry in enumerate(qa_entries, 1): + parts = [f"#{idx} [{prompts.QA_QUERY_LABEL}@{entry.query_time or 'N/A'}] {entry.query}"] + for resp_time, payload in entry.responses: + parts.append(f"[{prompts.QA_RESPONSE_LABEL}@{resp_time}] {payload}") + lines.append("\n".join(parts)) + sections.append(prompts.QA_HISTORY_HEADER + "\n".join(lines)) + + if has_query and not query_in_current_chunk: + sections.append(prompts.USER_QUERY_HEADER + "\n" + current_query.strip()) + + return "\n\n".join(sections) + + +def _sample_indices(n: int, budget: int) -> list[int]: + if n == 0: + return [] + if budget <= 0 or n <= budget: + return list(range(n)) + if budget == 1: + return [n // 2] + selected: list[int] = [] + last_pos = -1 + for slot in range(budget): + target = round(slot * (n - 1) / (budget - 1)) + target = max(target, last_pos + 1) + target = min(target, n - (budget - slot)) + selected.append(target) + last_pos = target + return selected + + +class Summarizer: + def __init__( + self, + backend: ModelBackend, + *, + key_frames_per_chunk: int = 0, + mid_term_max_tokens: int = 4000, + long_term_max_tokens: int = 4000, + preferred_time_span: float = 10.0, + ) -> None: + self._backend = backend + self._key_frames = key_frames_per_chunk + self._mid_max_tokens = mid_term_max_tokens + self._long_max_tokens = long_term_max_tokens + self._preferred_time_span = preferred_time_span + + async def summarize_chunk( + self, + chunk_index: int, + frame_range: str, + frames: list[tuple[str, str]], + ) -> str: + if not frames: + return prompts.EMPTY_CHUNK_SUMMARY.format(frame_range=frame_range) + + picked = [frames[i] for i in _sample_indices(len(frames), self._key_frames)] + prompt = prompts.MID_TERM_SUMMARY_PROMPT.format( + chunk_index=chunk_index, + frame_range=frame_range, + length_instruction="", + preferred_time_span=f"{self._preferred_time_span:g} seconds", + ) + content: list[dict[str, Any]] = [{"type": "text", "text": prompt}] + for time_range, data_url in picked: + content.append({"type": "text", "text": f"<{time_range}>"}) + content.append({"type": "image_url", "image_url": {"url": data_url}}) + + text, _ = await self._backend.generate( + [{"role": "user", "content": content}], + max_tokens=self._mid_max_tokens, + temperature=0.8, + top_p=0.9, + extra_body={"top_k": 40, "repetition_penalty": 1.1, "presence_penalty": 1.0}, + ) + return text.strip() + + async def compress_to_long_term( + self, + existing_long_term: str, + mid_terms: list[MidTermSummary], + ) -> str: + if not mid_terms: + return existing_long_term + + merged_range = f"{_range_start(mid_terms[0].frame_range)}-{_range_end(mid_terms[-1].frame_range)}" + summaries_text = "\n\n".join(f"<{m.frame_range}>\n{m.summary_text}" for m in mid_terms) + prompt = prompts.LONG_TERM_COMPRESS_PROMPT.format( + merged_range=merged_range, + summaries_text=summaries_text, + length_instruction="", + ) + compressed, _ = await self._backend.generate( + [{"role": "user", "content": prompt}], + max_tokens=self._long_max_tokens, + temperature=0.3, + top_p=0.7, + extra_body={"top_k": 30, "repetition_penalty": 1.1, "presence_penalty": 0.5}, + ) + block = f"<{merged_range}>\n{compressed.strip()}" + return f"{existing_long_term.rstrip()}\n\n{block}" if existing_long_term else block + + async def aclose(self) -> None: + await self._backend.aclose() + + +def _range_start(frame_range: str) -> str: + for sep in (" ~ ", "-"): + if sep in frame_range: + return frame_range.split(sep, 1)[0].strip() + return frame_range.strip() + + +def _range_end(frame_range: str) -> str: + for sep in (" ~ ", "-"): + if sep in frame_range: + return frame_range.split(sep, 1)[-1].strip() + return frame_range.strip() diff --git a/vllm_omni/experimental/fullduplex/joyvl/serving/__init__.py b/vllm_omni/experimental/fullduplex/joyvl/serving/__init__.py new file mode 100644 index 000000000..208f01a7c --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/serving/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project diff --git a/vllm_omni/experimental/fullduplex/joyvl/serving/config.py b/vllm_omni/experimental/fullduplex/joyvl/serving/config.py new file mode 100644 index 000000000..3aa1265e5 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/serving/config.py @@ -0,0 +1,80 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +from dataclasses import dataclass, field + +from vllm_omni.experimental.fullduplex.joyvl.decision.prompts import SYSTEM_PROMPTS + + +@dataclass +class SamplingConfig: + max_tokens: int = 256 + temperature: float = 0.8 + top_p: float = 0.9 + top_k: int = 40 + repetition_penalty: float = 1.1 + presence_penalty: float = 1.5 + + +@dataclass +class InteractionConfig: + main_backend_url: str = "http://127.0.0.1:8061/v1" + main_model: str = "JoyAI-VL-Interaction-Preview" + api_key: str = "EMPTY" + + persona: str = "default" + frame_seconds: float = 1.0 + sampling: SamplingConfig = field(default_factory=SamplingConfig) + + force_silence_before_query: bool = True + + response_dedup_threshold: float = 1.0 + + enable_memory: bool = True + + summarizer_backend_url: str | None = None + summarizer_model: str | None = None + + chunk_frames: int = 100 + + mid_term_key_frames: int = 0 + + long_term_every_n_chunks: int = 5 + mid_term_max_tokens: int = 4000 + long_term_max_tokens: int = 4000 + keep_qa_history: bool = True + + enable_delegation: bool = True + delegation_backend_url: str | None = None + delegation_model: str | None = None + delegation_api_key: str | None = None + delegation_max_tokens: int = 512 + delegation_kind: str = "chat" + delegation_image_url: str | None = None + delegation_edit_url: str | None = None + delegation_edit_model: str | None = None + + session_timeout_seconds: float = 3600.0 + request_timeout_seconds: float = 300.0 + + @property + def system_prompt(self) -> str: + return SYSTEM_PROMPTS.get(self.persona, SYSTEM_PROMPTS["default"]) + + @property + def resolved_summarizer_url(self) -> str: + return self.summarizer_backend_url or self.main_backend_url + + @property + def resolved_summarizer_model(self) -> str: + return self.summarizer_model or self.main_model + + @property + def resolved_delegation_model(self) -> str: + return self.delegation_model or self.main_model + + @property + def resolved_delegation_api_key(self) -> str: + return self.delegation_api_key or self.api_key diff --git a/vllm_omni/experimental/fullduplex/joyvl/serving/server.py b/vllm_omni/experimental/fullduplex/joyvl/serving/server.py new file mode 100644 index 000000000..12aac292b --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/serving/server.py @@ -0,0 +1,358 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import time +import uuid +from collections.abc import AsyncIterator +from typing import Any + +import uvicorn +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from vllm_omni.experimental.fullduplex.joyvl.bridges.backend import OpenAIBackend +from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import ( + DelegationBridge, + ImageEditDelegationBridge, + ImageGenDelegationBridge, + OpenAIDelegationBridge, + RoutingDelegationBridge, + StubDelegationBridge, +) +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import to_token_form +from vllm_omni.experimental.fullduplex.joyvl.memory.memory import Summarizer +from vllm_omni.experimental.fullduplex.joyvl.serving.config import InteractionConfig +from vllm_omni.experimental.fullduplex.joyvl.serving.session import InteractionSession, StepResult + + +class SessionManager: + def __init__(self, config: InteractionConfig) -> None: + self.config = config + self._backend = OpenAIBackend( + config.main_backend_url, config.main_model, config.api_key, config.request_timeout_seconds + ) + self._summarizer: Summarizer | None = None + if config.enable_memory: + summarizer_backend = OpenAIBackend( + config.resolved_summarizer_url, + config.resolved_summarizer_model, + config.api_key, + config.request_timeout_seconds, + ) + self._summarizer = Summarizer( + summarizer_backend, + key_frames_per_chunk=config.mid_term_key_frames, + mid_term_max_tokens=config.mid_term_max_tokens, + long_term_max_tokens=config.long_term_max_tokens, + ) + self._delegation = self._build_delegation(config) if config.enable_delegation else None + self._sessions: dict[str, InteractionSession] = {} + self._locks: dict[str, asyncio.Lock] = {} + + @staticmethod + def _build_delegation(config: InteractionConfig) -> DelegationBridge | None: + # Returns None (delegation off) unless a real backend is configured. The stub is + # only returned on explicit opt-in (--delegation-kind stub) for tests/demos — it + # would otherwise fold fake answers into session memory. + kind = config.delegation_kind + url = config.delegation_backend_url + + def chat_bridge(backend_url: str) -> OpenAIDelegationBridge: + return OpenAIDelegationBridge( + backend_url, + config.resolved_delegation_model, + config.resolved_delegation_api_key, + max_tokens=config.delegation_max_tokens, + timeout=config.request_timeout_seconds, + ) + + if kind == "stub": + return StubDelegationBridge() + if kind == "router": + if not (url or config.delegation_image_url or config.delegation_edit_url): + return None + return RoutingDelegationBridge( + chat=chat_bridge(url) if url else None, + image=ImageGenDelegationBridge(config.delegation_image_url, timeout=config.request_timeout_seconds) + if config.delegation_image_url + else None, + edit=ImageEditDelegationBridge( + config.delegation_edit_url, + config.delegation_edit_model or config.main_model, + timeout=config.request_timeout_seconds, + ) + if config.delegation_edit_url + else None, + ) + if url and kind == "image": + return ImageGenDelegationBridge(url, timeout=config.request_timeout_seconds) + if url and kind == "edit": + return ImageEditDelegationBridge( + url, config.resolved_delegation_model, timeout=config.request_timeout_seconds + ) + if url: + return chat_bridge(url) + return None + + def _get(self, session_id: str) -> InteractionSession: + session = self._sessions.get(session_id) + if session is None: + session = InteractionSession( + session_id, + self.config, + self._backend, + summarizer=self._summarizer, + delegation=self._delegation, + ) + self._sessions[session_id] = session + self._locks[session_id] = asyncio.Lock() + return session + + async def step(self, session_id: str, frames: list[str], query: str | None) -> StepResult: + await self._evict_expired() + # capture session + lock together (no await between) so a concurrent reset cannot + # swap them out from under us; the lock then serializes step against reset. + session = self._get(session_id) + lock = self._locks[session_id] + async with lock: + return await session.step(frames, query) + + async def reset(self, session_id: str) -> None: + # Acquire the per-session lock so reset waits for any in-flight step() to finish + # instead of mutating/dropping a session mid-request. + lock = self._locks.get(session_id) + if lock is None: + return + async with lock: + session = self._sessions.pop(session_id, None) + self._locks.pop(session_id, None) + if session is not None: + await session.reset() + + async def set_persona(self, session_id: str, persona: str) -> bool: + session = self._get(session_id) + lock = self._locks[session_id] + async with lock: + return session.set_persona(persona) + + async def _evict_expired(self) -> None: + ttl = self.config.session_timeout_seconds + if ttl <= 0: + return + now = time.monotonic() + expired = [ + sid + for sid, sess in self._sessions.items() + if now - sess.last_access > ttl and not self._locks[sid].locked() + ] + for sid in expired: + await self.reset(sid) + + async def aclose(self) -> None: + for sid in list(self._sessions): + lock = self._locks.get(sid) + if lock is not None: + async with lock: + session = self._sessions.pop(sid, None) + self._locks.pop(sid, None) + if session is not None: + await session.aclose() + await self._backend.aclose() + if self._summarizer is not None: + await self._summarizer.aclose() + if self._delegation is not None: + await self._delegation.aclose() + + +def _extract_frames_and_query(payload: dict[str, Any]) -> tuple[list[str], str | None]: + messages = payload.get("messages") or [] + frames: list[str] = [] + texts: list[str] = [] + for message in messages: + if message.get("role") != "user": + continue + content = message.get("content") + if isinstance(content, str): + texts.append(content) + continue + for part in content or []: + ptype = part.get("type") + if ptype == "image_url": + url = (part.get("image_url") or {}).get("url") + if url: + frames.append(url) + elif ptype == "text" and part.get("text"): + texts.append(part["text"]) + query = "\n".join(t.strip() for t in texts if t.strip()) or None + return frames, query + + +def _session_id(request: Request, payload: dict[str, Any]) -> str: + return ( + request.headers.get("x-streaming-session") + or request.headers.get("x-session-id") + or payload.get("session_id") + or payload.get("user") + or "default" + ) + + +def _completion_response(model: str, result: StepResult) -> dict[str, Any]: + action = result.action + memory = {"long_term_memory": result.long_term_memory, "mid_term_summaries": result.mid_term_summaries} + return { + "id": f"intchat-{uuid.uuid4().hex}", + "object": "chat.completion", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": to_token_form(action)}, + "finish_reason": "stop", + } + ], + "usage": None, + "interaction": { + "action": action.action.value, + "spoke": action.spoke, + "text": action.text, + "delegated_question": action.delegated_question, + "delegation": result.delegation, + "chunk_index": result.chunk_index, + "frame_index": result.frame_index, + "inference_skipped": result.inference_skipped, + "latency_ms": result.latency_ms, + "memory": memory, + }, + } + + +def create_app(config: InteractionConfig) -> FastAPI: + manager = SessionManager(config) + + @contextlib.asynccontextmanager + async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + try: + yield + finally: + # close httpx/AsyncOpenAI clients and cancel+await pending tasks on shutdown/reload + await manager.aclose() + + app = FastAPI(title="vLLM-Omni Interaction Server", lifespan=lifespan) + + @app.get("/health") + async def health() -> dict[str, str]: + return {"status": "ok"} + + @app.get("/v1/models") + async def models() -> dict[str, Any]: + return {"object": "list", "data": [{"id": config.main_model, "object": "model", "owned_by": "vllm-omni"}]} + + @app.post("/v1/chat/completions") + async def chat_completions(request: Request) -> JSONResponse: + payload = await request.json() + frames, query = _extract_frames_and_query(payload) + if not frames: + return JSONResponse({"error": "interaction server requires at least one image_url frame"}, status_code=400) + result = await manager.step(_session_id(request, payload), frames, query) + return JSONResponse(_completion_response(config.main_model, result)) + + @app.post("/reset") + @app.post("/v1/streaming/reset") + async def reset(request: Request) -> dict[str, str]: + payload = await request.json() if await request.body() else {} + await manager.reset(_session_id(request, payload)) + return {"status": "reset"} + + @app.post("/v1/streaming/persona") + async def persona(request: Request) -> dict[str, Any]: + payload = await request.json() if await request.body() else {} + ok = await manager.set_persona(_session_id(request, payload), payload.get("persona", "default")) + return {"status": "ok" if ok else "unknown_persona"} + + return app + + +def _build_config(args: argparse.Namespace) -> InteractionConfig: + config = InteractionConfig( + main_backend_url=args.main_backend_url, + main_model=args.main_model, + persona=args.persona, + enable_memory=not args.no_memory, + summarizer_backend_url=args.summarizer_backend_url, + summarizer_model=args.summarizer_model, + enable_delegation=not args.no_delegation, + delegation_backend_url=args.delegation_backend_url, + delegation_model=args.delegation_model, + delegation_api_key=args.delegation_api_key, + delegation_kind=args.delegation_kind, + delegation_image_url=args.delegation_image_url, + delegation_edit_url=args.delegation_edit_url, + delegation_edit_model=args.delegation_edit_model, + force_silence_before_query=not args.no_force_silence, + ) + if args.chunk_frames is not None: + config.chunk_frames = args.chunk_frames + config.response_dedup_threshold = args.response_dedup_threshold + config.sampling.max_tokens = args.max_tokens + config.sampling.temperature = args.temperature + return config + + +def main() -> None: + parser = argparse.ArgumentParser(description="vLLM-Omni streaming interaction server") + parser.add_argument("--host", default="0.0.0.0") + parser.add_argument("--port", type=int, default=8070) + parser.add_argument("--main-backend-url", default="http://127.0.0.1:8061/v1") + parser.add_argument("--main-model", default="JoyAI-VL-Interaction-Preview") + parser.add_argument("--persona", default="default", choices=["default", "silent", "talkative"]) + parser.add_argument("--summarizer-backend-url", default=None) + parser.add_argument("--summarizer-model", default=None) + parser.add_argument("--chunk-frames", type=int, default=None) + parser.add_argument( + "--response-dedup-threshold", + type=float, + default=1.0, + help="1.0 drops only exact repeats (reference); < 1.0 also drops near-duplicate narration", + ) + parser.add_argument("--max-tokens", type=int, default=128) + parser.add_argument("--temperature", type=float, default=0.8) + parser.add_argument("--no-memory", action="store_true", help="disable mid/long-term summarizer memory") + parser.add_argument("--no-delegation", action="store_true", help="disable the delegation bridge") + parser.add_argument( + "--delegation-backend-url", + default=None, + help="OpenAI-compatible endpoint for the background brain; unset falls back to the stub", + ) + parser.add_argument("--delegation-model", default=None, help="model name for the background brain") + parser.add_argument( + "--delegation-api-key", + default=None, + help="API key for the background brain endpoint (e.g. an Anthropic key for a claude-* model)", + ) + parser.add_argument( + "--delegation-kind", + default="chat", + choices=["chat", "image", "edit", "router", "stub"], + help="chat = text/VL brain; image = text-to-image; edit = restyle the frame; " + "router = dispatch by request; stub = canned demo/test answers. chat/image/edit/router " + "need a backend URL — without one, delegation stays off.", + ) + parser.add_argument("--delegation-image-url", default=None, help="router mode: text-to-image endpoint") + parser.add_argument("--delegation-edit-url", default=None, help="router mode: image-edit endpoint") + parser.add_argument("--delegation-edit-model", default=None, help="router mode: image-edit model name") + parser.add_argument("--no-force-silence", action="store_true", help="run the model before any user query") + args = parser.parse_args() + + app = create_app(_build_config(args)) + uvicorn.run(app, host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/vllm_omni/experimental/fullduplex/joyvl/serving/session.py b/vllm_omni/experimental/fullduplex/joyvl/serving/session.py new file mode 100644 index 000000000..31e3b5092 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/joyvl/serving/session.py @@ -0,0 +1,179 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import asyncio +import logging +import time +from dataclasses import dataclass, field +from typing import Any + +from vllm_omni.experimental.fullduplex.joyvl.bridges.backend import ModelBackend +from vllm_omni.experimental.fullduplex.joyvl.bridges.delegation import DelegationBridge +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import Action, ParsedAction +from vllm_omni.experimental.fullduplex.joyvl.decision.policy import JoyVLPolicy +from vllm_omni.experimental.fullduplex.joyvl.decision.prompts import SYSTEM_PROMPTS, USER_QUERY_HEADER +from vllm_omni.experimental.fullduplex.joyvl.memory.memory import Summarizer, WorkingChunk +from vllm_omni.experimental.fullduplex.joyvl.serving.config import InteractionConfig + +logger = logging.getLogger("vllm_omni.experimental.fullduplex.joyvl") + + +@dataclass +class StepResult: + action: ParsedAction + chunk_index: int + frame_index: int + inference_skipped: bool + latency_ms: float + long_term_memory: str + mid_term_summaries: list[dict[str, Any]] = field(default_factory=list) + delegation: dict[str, Any] | None = None + + +class InteractionSession: + def __init__( + self, + session_id: str, + config: InteractionConfig, + backend: ModelBackend, + summarizer: Summarizer | None = None, + delegation: DelegationBridge | None = None, + ) -> None: + self.session_id = session_id + self.config = config + self._backend = backend + self._policy = JoyVLPolicy( + persona=config.persona, + summarizer=summarizer, + delegation=delegation, + chunk_frames=config.chunk_frames, + long_term_every_n_chunks=config.long_term_every_n_chunks, + keep_qa_history=config.keep_qa_history, + frame_seconds=config.frame_seconds, + enable_delegation=config.enable_delegation, + response_dedup_threshold=config.response_dedup_threshold, + ) + self.chunk = WorkingChunk() + self._system_prompt = config.system_prompt + self.last_access = time.monotonic() + self._consolidating: set[asyncio.Task] = set() + + def set_persona(self, persona: str) -> bool: + prompt = SYSTEM_PROMPTS.get(persona) + if prompt is None: + return False + self._system_prompt = prompt + return True + + async def step(self, frames: list[str], query: str | None = None, t: float | None = None) -> StepResult: + self.last_access = time.monotonic() + started = time.perf_counter() + policy = self._policy + brain = policy.brain + + base = t if t is not None else brain.frame_index * self.config.frame_seconds + time_ranges = [f"{base + i * self.config.frame_seconds:.1f}s" for i in range(len(frames))] + + delegation_info = await policy.fold_delegations() + + if policy.needs_flush(): + self._spawn_consolidation(policy.close_chunk(), policy.take_working_frames()) + self.chunk = WorkingChunk() + + query_is_fresh = policy.set_query(query) + + for tr, url in zip(time_ranges, frames): + policy.observe(tr, url) + self.chunk.messages.append(self._frame_message(time_ranges, frames, include_query=query_is_fresh)) + + if self.config.force_silence_before_query and not brain.current_query: + action = ParsedAction(Action.SILENCE, raw="") + skipped = True + else: + action = policy.commit(await self._infer()) + skipped = False + self.chunk.messages.append({"role": "assistant", "content": action.raw or ""}) + + submitted = await policy.submit_if_delegate(action, list(policy.working_frames)) + if submitted: + delegation_info = submitted + + return StepResult( + action=action, + chunk_index=brain.chunk_index, + frame_index=brain.frame_index, + inference_skipped=skipped, + latency_ms=round((time.perf_counter() - started) * 1000, 1), + long_term_memory=brain.memory.long_term_memory, + mid_term_summaries=[ + {"chunk_index": m.chunk_index, "frame_range": m.frame_range, "summary_text": m.summary_text} + for m in brain.memory.mid_term_summaries + ], + delegation=delegation_info, + ) + + async def reset(self) -> None: + await self._cancel_consolidation() + self._policy.brain.reset() + self.chunk = WorkingChunk() + + async def aclose(self) -> None: + await self._cancel_consolidation() + + async def _cancel_consolidation(self) -> None: + tasks = list(self._consolidating) + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + self._consolidating.clear() + + def _spawn_consolidation(self, chunk_index: int, frames: list[tuple[str, str]]) -> None: + task = asyncio.create_task(self._policy.consolidate(chunk_index, frames)) + self._consolidating.add(task) + task.add_done_callback(self._on_consolidation_done) + + def _on_consolidation_done(self, task: asyncio.Task) -> None: + self._consolidating.discard(task) + if not task.cancelled() and task.exception() is not None: + logger.warning("memory consolidation failed for session %s: %s", self.session_id, task.exception()) + + async def _infer(self) -> str: + s = self.config.sampling + extra_body = { + "skip_special_tokens": False, + "greedy": False, + "top_k": s.top_k, + "repetition_penalty": s.repetition_penalty, + "presence_penalty": s.presence_penalty, + } + raw, _ = await self._backend.generate( + self._build_api_messages(), + max_tokens=s.max_tokens, + temperature=s.temperature, + top_p=s.top_p, + extra_body=extra_body, + ) + return raw + + def _build_api_messages(self) -> list[dict[str, Any]]: + messages: list[dict[str, Any]] = [{"role": "system", "content": self._system_prompt}] + chunk_messages = [dict(m) for m in self.chunk.messages] + prefix = self._policy.brain.build_prefix() + if prefix and chunk_messages: + head = chunk_messages[0] + head["content"] = [{"type": "text", "text": prefix}] + list(head["content"]) + messages.extend(chunk_messages) + return messages + + def _frame_message(self, time_ranges: list[str], frames: list[str], include_query: bool) -> dict[str, Any]: + content: list[dict[str, Any]] = [] + query = self._policy.brain.current_query + if include_query and query: + content.append({"type": "text", "text": f"{USER_QUERY_HEADER}\n{query.strip()}"}) + for tr, url in zip(time_ranges, frames): + content.append({"type": "text", "text": f"<{tr}>"}) + content.append({"type": "image_url", "image_url": {"url": url}}) + return {"role": "user", "content": content}