[Fix] JoyVL serving: align with reference engine (bounded long-term memory, timestamps, max_pixels) + recipe updates (#4623)

Signed-off-by: lishunyang12 <125541396+lishunyang12@users.noreply.github.com>

recipes/JD/JoyAI-VL-Interaction.md

@@ -5,7 +5,7 @@
 ## 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)
+- Model: [`jdopensource/JoyAI-VL-Interaction-Preview`](https://huggingface.co/jdopensource/JoyAI-VL-Interaction-Preview) (8B, Qwen3-VL architecture with weights retrained for streaming interaction)
 - 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
@@ -25,7 +25,8 @@ framework internals and how to add another full-duplex model, see
 - 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
+  ≈48GB+; for ~24GB, lower `--chunk-frames`, `--max-model-len`, and the image limit together,
+  and/or load the weights in fp8 (see "Smaller GPU: fp8 weights" below — ~9.9 GiB vs 16.8 GiB)
 - vLLM / vLLM-Omni: versions from your current checkout
 
 ## Start server
@@ -33,7 +34,7 @@ framework internals and how to add another full-duplex model, see
 From repository root:
 
 ```bash
-# 1. Serve the model (plain `vllm serve`, NOT --omni; it is vanilla Qwen3-VL).
+# 1. Serve the model (plain `vllm serve`, NOT --omni; it uses the Qwen3-VL architecture).
 #    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 \
@@ -46,6 +47,26 @@ 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
 ```
 
+### Smaller GPU: fp8 weights
+
+Add `--quantization fp8` to the step-1 `vllm serve` to load the weights in fp8 online
+from the bf16 checkpoint (no separate quantized checkpoint needed; uses vLLM's generic
+fp8 path, since the model is a standard Qwen3-VL VLM):
+
+```bash
+vllm serve jdopensource/JoyAI-VL-Interaction-Preview \
+  --served-model-name JoyAI-VL-Interaction-Preview --port 8061 \
+  --quantization fp8 --max-model-len 131072 --enable-prefix-caching \
+  --limit-mm-per-prompt '{"image":256,"video":1}'
+```
+
+Measured here: model weights **9.9 GiB** vs **16.8 GiB** bf16 (**−41%**), which lets the 8B
+fit a much smaller card. This is a **memory** option, not a speed one: at one request per
+tick (the streaming regime) fp8 was **not** faster (≈19% slower in our test) because
+single-stream decode is memory-latency-bound and pays fp8's dynamic-scale overhead; its
+throughput gain needs batching. Decisions stay coherent, but fp8 can shift the
+speak/silence boundary, so for timing-critical alerting validate output against bf16 first.
+
 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:
 
@@ -176,14 +197,22 @@ 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.
+- `--omni` is **not** used: the model keeps the Qwen3-VL architecture (only the
+  weights are retrained), 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.
+- **Frame resolution is the main latency knob.** The system/memory prefix is prefix-cached,
+  so per-tick *new* compute is dominated by the new frame's vision tokens. Measured on the
+  8B: ~256×192 frames hit Qwen3-VL's min-pixel floor (~72 vision tokens, ~17 ms/tick) vs
+  ~302 tokens / ~38 ms at 640×480 — about 2× cheaper. Downsample frames to ~256×192 for the
+  tightest latency / highest concurrency; one GPU then sustains ~150–180 concurrent 1 fps
+  streams with p95 < 200 ms. Per-tick latency is already far inside the 1 fps budget, so
+  serving is rarely the bottleneck — resolution is the lever if you need more headroom.
 - 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

tests/fullduplex/test_joyvl_brain.py

@@ -17,8 +17,8 @@ class _FakeSummarizer:
         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)})"
+    async def compress_to_long_term(self, mids):
+        return f"compressed({len(mids)})"
 
 
 class _FakeDelegation:
@@ -42,7 +42,7 @@ def test_query_freshness_and_archive_on_change():
     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")]
+    assert b.memory.qa_history[-1].responses == [("0.0 seconds", "1 bottle")]
 
 
 def test_query_in_frame_on_arrival_then_prefix_on_later_chunks():
@@ -71,7 +71,7 @@ async def test_flush_archives_qa_and_summarizes():
     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 b.memory.qa_history[-1].responses == [("2.0 seconds", "a person enters")]
     assert len(b.memory.mid_term_summaries) == 1
 
 
@@ -85,6 +85,20 @@ async def test_long_term_compression_after_n_chunks():
     assert b.memory.mid_term_summaries == []
 
 
+@pytest.mark.asyncio
+async def test_long_term_window_caps_blocks():
+    # Long-term memory must not grow unbounded over hours: oldest blocks are dropped
+    # once the sliding window is full (reference window = 15; here 2 for the test).
+    summ = _FakeSummarizer()
+    b = InteractionBrain(
+        summarizer=summ, chunk_frames=1, long_term_every_n_chunks=1, long_term_window=2, frame_seconds=1.0
+    )
+    for _ in range(4):
+        await b.flush([("0.0s", "u")])
+    assert len(b.memory.long_term_blocks) == 2  # capped, not 4
+    assert b.memory.long_term_memory.count("compressed(") == 2
+
+
 @pytest.mark.asyncio
 async def test_delegation_submit_and_fold():
     b = InteractionBrain(delegation=_FakeDelegation())

tests/fullduplex/test_joyvl_policy.py

@@ -38,7 +38,7 @@ def test_commit_records_only_spoken():
     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")]
+    assert p.brain.response_records == [("0.0 seconds", "hello")]
 
 
 def test_commit_dedup_default_is_exact_match_only():

vllm_omni/experimental/fullduplex/joyvl/decision/policy.py

@@ -33,6 +33,7 @@ class JoyVLPolicy:
         delegation: DelegationBridge | None = None,
         chunk_frames: int = 16,
         long_term_every_n_chunks: int = 5,
+        long_term_window: int = 15,
         keep_qa_history: bool = True,
         frame_seconds: float = 1.0,
         enable_delegation: bool = True,
@@ -44,6 +45,7 @@ class JoyVLPolicy:
             delegation=delegation,
             chunk_frames=chunk_frames,
             long_term_every_n_chunks=long_term_every_n_chunks,
+            long_term_window=long_term_window,
             keep_qa_history=keep_qa_history,
             frame_seconds=frame_seconds,
             enable_delegation=enable_delegation,

vllm_omni/experimental/fullduplex/joyvl/memory/brain.py

@@ -25,6 +25,7 @@ class InteractionBrain:
         delegation: DelegationBridge | None = None,
         chunk_frames: int = 16,
         long_term_every_n_chunks: int = 5,
+        long_term_window: int = 15,
         keep_qa_history: bool = True,
         frame_seconds: float = 1.0,
         enable_delegation: bool = True,
@@ -33,6 +34,7 @@ class InteractionBrain:
         self._delegation = delegation
         self._chunk_frames = chunk_frames
         self._long_term_every_n = long_term_every_n_chunks
+        self._long_term_window = long_term_window
         self._keep_qa_history = keep_qa_history
         self._frame_seconds = frame_seconds
         self._enable_delegation = enable_delegation
@@ -58,10 +60,10 @@ class InteractionBrain:
         self._consolidate_lock = asyncio.Lock()
 
     def now(self) -> str:
-        return f"{self.frame_index * self._frame_seconds:.1f}s"
+        return f"{self.frame_index * self._frame_seconds:.1f} seconds"
 
     def last_frame_time(self) -> str:
-        return f"{max(0, self.frame_index - 1) * self._frame_seconds:.1f}s"
+        return f"{max(0, self.frame_index - 1) * self._frame_seconds:.1f} seconds"
 
     def tick(self, n: int = 1) -> None:
         self.frame_index += n
@@ -131,9 +133,14 @@ class InteractionBrain:
             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
-                )
+                block = await self._summarizer.compress_to_long_term(self.memory.mid_term_summaries)
+                if block:
+                    self.memory.long_term_blocks.append(block)
+                    window = self._long_term_window
+                    if window > 0 and len(self.memory.long_term_blocks) > window:
+                        # drop oldest blocks so long-term memory stays bounded over hours
+                        del self.memory.long_term_blocks[: len(self.memory.long_term_blocks) - window]
+                    self.memory.long_term_memory = "\n\n".join(self.memory.long_term_blocks)
                 self.memory.mid_term_summaries.clear()
 
     async def flush(self, frames: list[tuple[str, str]]) -> None:

vllm_omni/experimental/fullduplex/joyvl/memory/memory.py

@@ -33,6 +33,7 @@ class QAEntry:
 @dataclass
 class SessionMemory:
     long_term_memory: str = ""
+    long_term_blocks: list[str] = field(default_factory=list)
     mid_term_summaries: list[MidTermSummary] = field(default_factory=list)
     qa_history: list[QAEntry] = field(default_factory=list)
 
@@ -91,6 +92,35 @@ def _sample_indices(n: int, budget: int) -> list[int]:
     return selected
 
 
+def _resize_data_url(data_url: str, max_pixels: int) -> str:
+    """Downscale a base64 image data URL so width*height <= max_pixels (preserving
+    aspect ratio), mirroring the reference summarizer. No-op if already within budget,
+    not a data URL, or PIL is unavailable."""
+    if max_pixels <= 0 or not data_url.startswith("data:"):
+        return data_url
+    import base64
+    import io
+    import re
+
+    match = re.match(r"data:image/\w+;base64,(.+)", data_url)
+    if not match:
+        return data_url
+    try:
+        from PIL import Image
+
+        img = Image.open(io.BytesIO(base64.b64decode(match.group(1))))
+        w, h = img.size
+        if w * h <= max_pixels:
+            return data_url
+        scale = (max_pixels / (w * h)) ** 0.5
+        img = img.resize((max(1, int(w * scale)), max(1, int(h * scale))), Image.LANCZOS)
+        buf = io.BytesIO()
+        img.convert("RGB").save(buf, format="JPEG")
+        return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
+    except Exception:
+        return data_url
+
+
 class Summarizer:
     def __init__(
         self,
@@ -100,12 +130,14 @@ class Summarizer:
         mid_term_max_tokens: int = 4000,
         long_term_max_tokens: int = 4000,
         preferred_time_span: float = 10.0,
+        max_pixels: int = 262144,
     ) -> 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
+        self._max_pixels = max_pixels
 
     async def summarize_chunk(
         self,
@@ -126,24 +158,25 @@ class Summarizer:
         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}})
+            content.append({"type": "image_url", "image_url": {"url": _resize_data_url(data_url, self._max_pixels)}})
 
         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},
+            extra_body={"top_k": 40, "repetition_penalty": 1.1, "presence_penalty": 1.0, "greedy": False},
         )
         return text.strip()
 
-    async def compress_to_long_term(
-        self,
-        existing_long_term: str,
-        mid_terms: list[MidTermSummary],
-    ) -> str:
+    async def compress_to_long_term(self, mid_terms: list[MidTermSummary]) -> str:
+        """Compress a batch of mid-term summaries into a single long-term block.
+
+        Returns one block; the caller (InteractionBrain) accumulates blocks and applies
+        the sliding window, matching the reference long-term memory behavior.
+        """
         if not mid_terms:
-            return existing_long_term
+            return ""
 
         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)
@@ -157,10 +190,9 @@ class Summarizer:
             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},
+            extra_body={"top_k": 30, "repetition_penalty": 1.1, "presence_penalty": 0.5, "greedy": False},
         )
-        block = f"<{merged_range}>\n{compressed.strip()}"
-        return f"{existing_long_term.rstrip()}\n\n{block}" if existing_long_term else block
+        return f"<{merged_range}>\n{compressed.strip()}"
 
     async def aclose(self) -> None:
         await self._backend.aclose()

vllm_omni/experimental/fullduplex/joyvl/serving/config.py

@@ -26,6 +26,7 @@ class InteractionConfig:
 
     persona: str = "default"
     frame_seconds: float = 1.0
+    max_pixels: int = 262144
     sampling: SamplingConfig = field(default_factory=SamplingConfig)
 
     force_silence_before_query: bool = True
@@ -42,6 +43,7 @@ class InteractionConfig:
     mid_term_key_frames: int = 0
 
     long_term_every_n_chunks: int = 5
+    long_term_memory_window: int = 15
     mid_term_max_tokens: int = 4000
     long_term_max_tokens: int = 4000
     keep_qa_history: bool = True

vllm_omni/experimental/fullduplex/joyvl/serving/server.py

@@ -49,6 +49,7 @@ class SessionManager:
                 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,
+                max_pixels=config.max_pixels,
             )
         self._delegation = self._build_delegation(config) if config.enable_delegation else None
         self._sessions: dict[str, InteractionSession] = {}

vllm_omni/experimental/fullduplex/joyvl/serving/session.py

@@ -50,6 +50,7 @@ class InteractionSession:
             delegation=delegation,
             chunk_frames=config.chunk_frames,
             long_term_every_n_chunks=config.long_term_every_n_chunks,
+            long_term_window=config.long_term_memory_window,
             keep_qa_history=config.keep_qa_history,
             frame_seconds=config.frame_seconds,
             enable_delegation=config.enable_delegation,
@@ -74,7 +75,7 @@ class InteractionSession:
         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))]
+        time_ranges = [f"{base + i * self.config.frame_seconds:.1f} seconds" for i in range(len(frames))]
 
         delegation_info = await policy.fold_delegations()
 
@@ -175,5 +176,5 @@ class InteractionSession:
             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}})
+            content.append({"type": "image_url", "image_url": {"url": url}, "max_pixels": self.config.max_pixels})
         return {"role": "user", "content": content}