[Feature] JoyAI-VL-Interaction streaming interaction serving layer (#4575)

Signed-off-by: lishunyang12 <125541396+lishunyang12@users.noreply.github.com>
Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
Signed-off-by: lishunyang <lishunyang12@163.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
Co-authored-by: Yueqian Lin <linyueqian@outlook.com>
This commit is contained in:
authored and GitHub committed 2026-06-21 07:42:51 +00:00
1 parent 950e66a44c
commit e241005a71
45 files changed
+3699 -1

No files matched your search

+1
View File
@@ -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
@@ -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 = "<asr_text>"
def _clean_transcript(raw: str) -> str:
"""Qwen3-ASR emits ``language <Lang><asr_text><transcript>``; 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("</asr_text>", "").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()
@@ -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()
@@ -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()
@@ -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)
@@ -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.
+11
View File
@@ -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"
+38
View File
@@ -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"
+64
View File
@@ -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://<base>/<filename-without-extension>.
# 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
@@ -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 </delegation> (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:-<unset>}"
echo " ASR : ${ASR_URL:-<unset>}"
echo " agent : ${BACKGROUND_CODEX_API_URL:-<unset>}"
echo " logs : ${LOG}/{model,orch,webui}.log"
@@ -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}"
@@ -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[@]}"
@@ -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}"
@@ -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}"
+1
View File
@@ -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:
+4 -1
View File
@@ -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"
+192
View File
@@ -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 `</delegation> <question>` 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 <brain-endpoint>/v1 \
--delegation-model <brain-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.
+1
View File
@@ -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
+96
View File
@@ -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("</response> hold on <delegation> 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"
+282
View File
@@ -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 "</silence>" 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("</silence>").action.value == "silence"
assert p.brain.response_records == []
assert p.commit("</response> 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("</response> a red ball appears").action.value == "response"
assert p.commit("</response> a red ball appears").action.value == "silence" # exact repeat dropped
assert p.commit("</response> 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("</response> a red ball appears").action.value == "response"
assert p.commit("</response> a red ball appears!").action.value == "silence" # near-dup dropped
assert p.commit("</response> 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("</response> someone raises a hand").action.value == "response"
# no gap: an immediate repeat is still suppressed
assert p.commit("</response> someone raises a hand").action.value == "silence"
# a genuine silent tick (hand lowered) re-arms; the recurrence speaks again
assert p.commit("</silence>").action.value == "silence"
assert p.commit("</response> 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 "</silence>", 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("</response> hold on <delegation> 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"
+172
View File
@@ -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(["</response> a fire is breaking out", "</silence>"])
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
+3
View File
@@ -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)."""
@@ -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/<model>/` 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.
@@ -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",
]
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -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: ...
@@ -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}
@@ -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)
@@ -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
@@ -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",
]
@@ -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
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -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()
@@ -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()
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -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 = "</silence>"
RESPONSE_TOKEN = "</response>"
DELEGATION_TAG = "</delegation>"
_DELEGATION_MARKERS = ("</delegation>", "<delegation>")
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())
@@ -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="</silence>")
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 [])
@@ -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:
</silence>
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:
</response> 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:
</response> Brief note that you're delegating. </delegation> <the question>""".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:
</silence>
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:
</response> 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:
</silence>
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:
</response> 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:
</response> Brief note that you're delegating. <delegation> <the question>
## 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} 跨度调整),仅在能改善事件区分或阶段切换识别时使用。
- 不要保留源摘要的 <Xs-Ys> 标题,也不要复制源摘要中"在 <X>s..."的连续时间戳节奏。
- 连续或重复的相邻动作合并为一个范围,并在范围内描述可见变化。
【事实性约束】
- 仅使用中期摘要中已有的内容;不要补全画面外信息或推测后续发展。
- 保留具体步骤与可读数值,不要用"准备继续"、"做相关操作"等空话替代。
- 避免泛泛的总结性短语、元评论(如"这段记忆包含……")和重复句式。
仅输出统一的叙述文本,不要输出其他内容。"""
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -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
@@ -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()
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -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
@@ -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()
@@ -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="</silence>")
skipped = True
else:
action = policy.commit(await self._infer())
skipped = False
self.chunk.messages.append({"role": "assistant", "content": action.raw or "</silence>"})
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}