From 3dc056e20302dbb6af334ab8ebc2e7417758029f Mon Sep 17 00:00:00 2001 From: Devdeep Ray Date: Tue, 11 Aug 2026 21:27:04 +0000 Subject: [PATCH 1/3] refactor(simple-vlm): use native live vision tool Signed-off-by: Devdeep Ray --- DEPENDENCIES.md | 22 +- agent-samples/simple-vlm-example/README.md | 10 +- .../simple-vlm-example/worker/pyproject.toml | 2 +- .../worker/simple_vlm_example_worker/app.py | 39 ++-- agent-sdk/xr-ai-nat/README.md | 10 + agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py | 10 +- agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py | 56 +++++ agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py | 220 ++++++++++++++++++ agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py | 72 ++++++ docs/source/components/agent-sdk.md | 7 + tests/test_native_tools.py | 40 +++- tests/test_simple_vlm_example_worker.py | 119 +++++----- 12 files changed, 501 insertions(+), 106 deletions(-) create mode 100644 agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py create mode 100644 agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py create mode 100644 agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md index adc263500..0a3b9b6ae 100644 --- a/DEPENDENCIES.md +++ b/DEPENDENCIES.md @@ -554,16 +554,18 @@ the latest video frame via streaming VLM and replies with both | Sub-project | Package | Internal deps | External deps | |---|---|---|---| | Orchestrator | `simple-vlm-example` | `xr-ai-launcher` | — | -| Worker | `simple-vlm-example-worker` | `xr-ai-hub-client [editable]`, `xr-ai-logging [editable]`, `xr-ai-models [editable]`, `xr-ai-nat[vision,voice] [editable]`, `xr-ai-voice [editable]`, `xr-ai-voicegate [editable]` | loguru >=0.7, pyyaml >=6.0 (`xr-ai-voice` pulls in VAD, pipecat-ai, numpy, and scipy; `xr-ai-nat[vision]` pulls in httpx, numpy, and Pillow) | - -The packaged worker registers `StreamingVisionConfig` in-process and maps it to -`VoiceSession` with `xr_ai_nat.adapters.as_voice_handler`. `VoiceSession` owns -readiness, hub transport, signals, the private Pipecat pipeline, and cleanup; -`TextMessageInput` routes `"ping"` and ad-hoc text through the same -participant-aware path as speech. Voice-gate behavior (magic phrases, follow-up -grace, listening chime, stop acknowledgement), VAD/STT, and sentence-batched TTS -remain provided by the shared voice runtime. The sample has no direct -`xr-ai-pipecat` or MCP dependency. +| Worker | `simple-vlm-example-worker` | `xr-ai-hub-client [editable]`, `xr-ai-logging [editable]`, `xr-ai-models [editable]`, `xr-ai-nat[relay,live-vision] [editable]`, `xr-ai-voice [editable]`, `xr-ai-voicegate [editable]` | loguru >=0.7, pyyaml >=6.0 (`xr-ai-voice` pulls in VAD, pipecat-ai, numpy, and scipy; `xr-ai-nat[live-vision]` pulls in numpy and Pillow) | + +The packaged worker invokes `LiveVisionTool` directly and maps its streamed +participant-scoped response to `VoiceSession`. The tool acquires a current frame +through `xr-ai-hub-client` and runs both its tool lifecycle and nested VLM stream +through NeMo Relay. `VoiceSession` owns readiness, hub transport, signals, the +private Pipecat pipeline, and cleanup; `TextMessageInput` routes `"ping"` and +ad-hoc text through the same participant-aware path as speech. Voice-gate +behavior (magic phrases, follow-up grace, listening chime, stop acknowledgement), +VAD/STT, and sentence-batched TTS remain provided by the shared voice runtime. +The sample has no direct `xr-ai-pipecat` or MCP dependency and selects no legacy +NeMo Agent Toolkit extra. Worker calls stt-server (8103), vlm-server (8100), and piper-tts-server (8105) over HTTP via `xr-ai-models` SDK — no model weights loaded diff --git a/agent-samples/simple-vlm-example/README.md b/agent-samples/simple-vlm-example/README.md index 98f7ad11f..57e38ef90 100644 --- a/agent-samples/simple-vlm-example/README.md +++ b/agent-samples/simple-vlm-example/README.md @@ -18,11 +18,11 @@ The worker is a package under `worker/simple_vlm_example_worker/`: - `prompts/system.txt` owns the VLM system prompt. `VoiceSession` owns STT/TTS/VLM readiness, the hub voice transport, voice-gate -processing, streaming TTS, signals, and cleanup. The application registers -`StreamingVisionConfig` in-process and adapts that native function with -`xr_ai_nat.adapters.as_voice_handler`. Typed text uses the same participant-aware -turn path as speech. Participant leave events release cached live-frame state, -and a newer turn cancels and interrupts a superseded response. +processing, streaming TTS, signals, and cleanup. The application invokes the +native `LiveVisionTool` directly; it acquires the participant's current frame +and streams an injected VLM through Relay. Typed text uses the same +participant-aware turn path as speech. Participant leave events release cached +live-frame state, and a newer turn cancels and interrupts a superseded response. No MCP client or MCP tool invocation is part of this sample. diff --git a/agent-samples/simple-vlm-example/worker/pyproject.toml b/agent-samples/simple-vlm-example/worker/pyproject.toml index ed825e72c..95d78d2c6 100644 --- a/agent-samples/simple-vlm-example/worker/pyproject.toml +++ b/agent-samples/simple-vlm-example/worker/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ "xr-ai-hub-client", "xr-ai-logging", "xr-ai-models", - "xr-ai-nat[vision,voice]", + "xr-ai-nat[relay,live-vision]", "xr-ai-voice", "xr-ai-voicegate", "loguru>=0.7", diff --git a/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py b/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py index ce134aa29..b577304a1 100644 --- a/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py +++ b/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py @@ -9,31 +9,29 @@ from pathlib import Path from loguru import logger -from nat.builder.workflow_builder import WorkflowBuilder -from nat.plugin_api import Function from xr_ai_logging import setup_logging from xr_ai_models import load_models_config, make_stt, make_tts, make_vlm -from xr_ai_nat.adapters import as_voice_handler -from xr_ai_nat.functions.vision import ( - StreamingVisionConfig, - VisionRequest, -) +from xr_ai_nat.live_vision import LiveVisionTool, VisionRequest from xr_ai_voice import TextMessageInput, VadConfig, VoiceHandler, VoiceSession from xr_ai_voicegate import load_voice_gate_config from .config import WorkerConfig -def _make_vision_handler(vision: Function) -> VoiceHandler: - return as_voice_handler( - vision, - request=lambda turn: VisionRequest( - participant_id=turn.participant_id, - query=turn.text, - ), - response=lambda chunk: chunk.text, - streaming=True, - ) +def _make_vision_handler(vision: LiveVisionTool) -> VoiceHandler: + async def handle(turn): + async def response(): + async for chunk in vision.stream( + VisionRequest( + participant_id=turn.participant_id, + query=turn.text, + ) + ): + yield chunk.text + + return response() + + return handle def _text_transform(default_prompt: str) -> Callable[[str], str]: @@ -70,15 +68,14 @@ async def run_app( idle_timeout_secs=config.idle_timeout_secs, ) - async with session, WorkflowBuilder() as builder: - vision_config = StreamingVisionConfig( + async with session: + vision = LiveVisionTool( endpoint=session.transport.endpoint, vlm=vlm, system_prompt=config.system_prompt, frame_max_age_s=config.frame_max_age_s, frame_timeout_s=config.frame_timeout_s, ) - vision = await builder.add_function("perception", vision_config) TextMessageInput( session=session, transform=_text_transform(config.default_prompt), @@ -88,7 +85,7 @@ async def run_app( logger.info("simple-vlm-example starting") await session.run( _make_vision_handler(vision), - on_participant_left=vision_config.release, + on_participant_left=vision.release, interrupt_on_supersede=True, ) logger.info("simple-vlm-example stopped") diff --git a/agent-sdk/xr-ai-nat/README.md b/agent-sdk/xr-ai-nat/README.md index cf8822796..097862107 100644 --- a/agent-sdk/xr-ai-nat/README.md +++ b/agent-sdk/xr-ai-nat/README.md @@ -60,6 +60,16 @@ custom, Fabric-backed, or framework-backed runner through the same registered invocation path. Relay observes model calls inside a tool-backed runner; the application never calls an LLM client as a separate control path. +## Live vision tool + +Install `xr-ai-nat[relay,live-vision]` for `LiveVisionTool`. It is a normal +streaming tool: its request carries a participant ID and question, it acquires a +fresh frame through `xr-ai-hub-client`, and it calls an injected `VLMService` +through a nested Relay LLM stream. `release(participant_id)` clears cached frame +state when the participant leaves. Voice applications can map its chunks to a +`VoiceHandler`; the event dispatcher follow-up invokes this same tool from voice +and autonomous triggers. + ## Legacy NAT compatibility ## Shared value models and the service boundary diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py b/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py index dfcdb78c0..3e06eaf08 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py @@ -4,6 +4,14 @@ """Toolkit-independent native XR tools with legacy NAT compatibility.""" from .agent_runner import AgentRunner, as_agent_tool +from .streaming import StreamingTool from .tools import Tool, ToolInvocationResult, ToolSet -__all__ = ["AgentRunner", "Tool", "ToolInvocationResult", "ToolSet", "as_agent_tool"] +__all__ = [ + "AgentRunner", + "StreamingTool", + "Tool", + "ToolInvocationResult", + "ToolSet", + "as_agent_tool", +] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py b/agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py new file mode 100644 index 000000000..05ac34c41 --- /dev/null +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Convert hub frames into JPEG data URLs accepted by VLM services.""" + +from __future__ import annotations + +import base64 +import io + +import numpy as np +from PIL import Image +from xr_ai_hub import FrameData, PixelFormat + + +def _yuv_to_rgb(y: np.ndarray, u: np.ndarray, v: np.ndarray) -> Image.Image: + y = y.astype(np.float32) - 16.0 + u = u.astype(np.float32) - 128.0 + v = v.astype(np.float32) - 128.0 + red = np.clip(1.164 * y + 1.596 * v, 0, 255) + green = np.clip(1.164 * y - 0.392 * u - 0.813 * v, 0, 255) + blue = np.clip(1.164 * y + 2.017 * u, 0, 255) + return Image.fromarray(np.stack([red, green, blue], axis=-1).astype(np.uint8), "RGB") + + +def frame_to_pil(frame: FrameData) -> Image.Image: + """Convert a hub frame to RGB pixels suitable for JPEG encoding.""" + + width, height = frame.width, frame.height + data = np.frombuffer(frame.data, dtype=np.uint8) + if frame.fmt == PixelFormat.RGB24: + return Image.fromarray(data.reshape(height, width, 3), "RGB") + if frame.fmt == PixelFormat.RGBA: + return Image.fromarray(data.reshape(height, width, 4), "RGBA").convert("RGB") + if frame.fmt == PixelFormat.BGRA: + bgra = data.reshape(height, width, 4) + return Image.fromarray(bgra[:, :, [2, 1, 0]], "RGB") + y_end = width * height + y = data[:y_end].reshape(height, width) + if frame.fmt == PixelFormat.I420: + uv_size = (width // 2) * (height // 2) + u = data[y_end : y_end + uv_size].reshape(height // 2, width // 2).repeat(2, 0).repeat(2, 1) + v = data[y_end + uv_size :].reshape(height // 2, width // 2).repeat(2, 0).repeat(2, 1) + return _yuv_to_rgb(y, u, v) + if frame.fmt == PixelFormat.NV12: + uv = data[y_end:].reshape(height // 2, width) + return _yuv_to_rgb(y, uv[:, 0::2].repeat(2, 0).repeat(2, 1), uv[:, 1::2].repeat(2, 0).repeat(2, 1)) + raise ValueError(f"Unsupported pixel format: {frame.fmt!r}") + + +def encode_image(image: Image.Image) -> str: + """Encode one image as an OpenAI-compatible JPEG data URL.""" + + buffer = io.BytesIO() + image.save(buffer, format="JPEG", quality=90) + return f"data:image/jpeg;base64,{base64.b64encode(buffer.getvalue()).decode()}" diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py new file mode 100644 index 000000000..c1a11edcc --- /dev/null +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py @@ -0,0 +1,220 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""A normal streaming tool for questions about a participant's current frame.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator, Mapping +from typing import Any + +import nemo_relay +from nemo_relay.codecs import OpenAIChatCodec +from pydantic import BaseModel, ConfigDict, Field +from xr_ai_hub import FrameUnavailable, LiveFrameSource, ProcessorEndpoint +from xr_ai_models import VLMService + +from ._pixels import encode_image, frame_to_pil +from .streaming import StreamingTool + +_LOGGER = logging.getLogger(__name__) +_VLM_CALL_NAME = "xr-ai-vlm" + + +class VisionRequest(BaseModel): + """Ask one question about a participant's current live camera frame.""" + + model_config = ConfigDict(extra="forbid") + + participant_id: str = Field(description="Participant whose camera frame should be inspected.") + query: str = Field(min_length=1, description="Question to answer from the camera frame.") + + +class VisionChunk(BaseModel): + """One streamed text fragment from a current-frame answer.""" + + text: str = Field(description="A partial fragment of the streamed answer text.") + + +class LiveVisionTool(StreamingTool[VisionRequest, VisionChunk]): + """A participant-scoped current-frame VLM tool with an injected model service.""" + + def __init__( + self, + *, + endpoint: ProcessorEndpoint, + vlm: VLMService, + system_prompt: str = "", + frame_max_age_s: float = 2.0, + frame_timeout_s: float = 5.0, + ) -> None: + if frame_max_age_s <= 0.0: + raise ValueError("frame_max_age_s must be positive") + if frame_timeout_s <= 0.0: + raise ValueError("frame_timeout_s must be positive") + self.endpoint = endpoint + self.vlm = vlm + self.system_prompt = system_prompt + self.frames = LiveFrameSource( + endpoint, + max_age_s=frame_max_age_s, + timeout_s=frame_timeout_s, + ) + super().__init__( + "look_at_current_frame", + "Answer a question about a participant's current live camera view.", + VisionRequest, + VisionChunk, + self._stream_current, + ) + + def release(self, participant_id: str) -> None: + """Forget cached frame state after a participant disconnects.""" + + self.frames.release(participant_id) + + async def _stream_current(self, request: VisionRequest) -> AsyncIterator[VisionChunk]: + try: + image_url = await self._current_image(request.participant_id) + except FrameUnavailable as exc: + yield VisionChunk(text=str(exc)) + return + except Exception: + _LOGGER.exception("Live frame conversion failed") + yield VisionChunk(text="VLM server unavailable — please retry.") + return + + await self.endpoint.set_status("processing", request.participant_id) + fragments: list[str] = [] + try: + stream = await nemo_relay.llm.stream_execute( + _VLM_CALL_NAME, + self._relay_request(image_url, request.query), + self._stream_vlm, + lambda chunk: fragments.append(_stream_text(chunk)), + lambda: _openai_response("".join(fragments)), + model_name=_VLM_CALL_NAME, + codec=OpenAIChatCodec(), + response_codec=OpenAIChatCodec(), + ) + async for chunk in stream: + text = _stream_text(chunk) + if text: + yield VisionChunk(text=text) + except Exception: + _LOGGER.exception("Live VLM stream failed") + yield VisionChunk(text="VLM server unavailable — please retry.") + finally: + await self.endpoint.set_status("idle", request.participant_id) + + async def _current_image(self, participant_id: str) -> str: + frame = await self.frames.get(participant_id) + return await asyncio.to_thread(lambda: encode_image(frame_to_pil(frame))) + + def _relay_request(self, image_url: str, query: str) -> nemo_relay.LLMRequest: + return nemo_relay.LLMRequest( + {}, + { + "model": _VLM_CALL_NAME, + "messages": [ + {"role": "system", "content": self.system_prompt}, + { + "role": "user", + "content": [ + {"type": "text", "text": query}, + {"type": "image_url", "image_url": {"url": image_url}}, + ], + }, + ], + }, + ) + + async def _stream_vlm( + self, + request: nemo_relay.LLMRequest, + ) -> AsyncIterator[dict[str, Any]]: + image_url, query, system_prompt = _vision_inputs(request.content) + async for token in self.vlm.stream( + image_url, + query, + system_prompt=system_prompt, + ): + yield {"choices": [{"delta": {"content": token}}]} + + +def _vision_inputs(content: Mapping[str, object]) -> tuple[str, str, str]: + messages = content.get("messages") + if not isinstance(messages, list): + raise TypeError("Relay VLM request must contain a message array") + + system_prompt = "" + image_url: str | None = None + query: str | None = None + for message in messages: + if not isinstance(message, Mapping): + raise TypeError("Relay VLM messages must be objects") + role = message.get("role") + raw_content = message.get("content") + if role == "system": + if not isinstance(raw_content, str): + raise TypeError("Relay VLM system content must be text") + system_prompt = raw_content + elif role == "user": + candidate_image, candidate_query = _image_and_text(raw_content) + if candidate_image is not None: + image_url = candidate_image + if candidate_query is not None: + query = candidate_query + + if image_url is None or query is None: + raise ValueError("Relay VLM request needs one image URL and one text question") + return image_url, query, system_prompt + + +def _image_and_text(content: object) -> tuple[str | None, str | None]: + if not isinstance(content, list): + raise TypeError("Relay VLM user content must be a multimodal array") + image_url: str | None = None + query: str | None = None + for part in content: + if not isinstance(part, Mapping): + raise TypeError("Relay VLM content parts must be objects") + if part.get("type") == "text" and isinstance(part.get("text"), str): + query = part["text"] + elif part.get("type") == "image_url": + image = part.get("image_url") + if isinstance(image, Mapping) and isinstance(image.get("url"), str): + image_url = image["url"] + return image_url, query + + +def _openai_response(text: str) -> dict[str, Any]: + return { + "model": _VLM_CALL_NAME, + "choices": [ + { + "message": {"role": "assistant", "content": text}, + "finish_reason": "stop", + } + ], + } + + +def _stream_text(raw_chunk: object) -> str: + if not isinstance(raw_chunk, Mapping): + raise TypeError("Relay VLM stream chunk must be an object") + choices = raw_chunk.get("choices") + if not isinstance(choices, list) or not choices or not isinstance(choices[0], Mapping): + raise ValueError("Relay VLM stream chunk must contain one choice") + delta = choices[0].get("delta") + if not isinstance(delta, Mapping): + raise TypeError("Relay VLM stream choice must contain a delta") + content = delta.get("content", "") + if not isinstance(content, str): + raise TypeError("Relay VLM stream content must be text") + return content + + +__all__ = ["LiveVisionTool", "VisionChunk", "VisionRequest"] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py b/agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py new file mode 100644 index 000000000..2bb5d8d42 --- /dev/null +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py @@ -0,0 +1,72 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Typed streaming tools for trigger-driven native application paths.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Awaitable, Callable +from inspect import isawaitable +from typing import Any, Generic, TypeVar, cast + +import nemo_relay +from pydantic import BaseModel + +RequestT = TypeVar("RequestT", bound=BaseModel) +ChunkT = TypeVar("ChunkT", bound=BaseModel) +ValueT = TypeVar("ValueT") + + +async def _resolve(value: ValueT | Awaitable[ValueT]) -> ValueT: + """Normalize Relay's sync-or-async hook contract for an async tool path.""" + + if isawaitable(value): + return await cast(Awaitable[ValueT], value) + return value + + +class StreamingTool(Generic[RequestT, ChunkT]): + """A typed tool that yields Pydantic chunks while keeping one Relay tool span open.""" + + def __init__( + self, + name: str, + description: str, + request_model: type[RequestT], + chunk_model: type[ChunkT], + handler: Callable[[RequestT], AsyncIterator[ChunkT]], + ) -> None: + if not name: + raise ValueError("tool name must not be empty") + if not description: + raise ValueError(f"tool {name!r} needs a description") + self.name = name + self.description = description + self.request_model = request_model + self.chunk_model = chunk_model + self.handler = handler + + async def stream(self, request: RequestT) -> AsyncIterator[ChunkT]: + """Yield one validated tool stream under a Relay lifecycle span.""" + + raw_request = request.model_dump(mode="json") + await _resolve(nemo_relay.tools.conditional_execution(self.name, raw_request)) + intercepted = await _resolve(nemo_relay.tools.request_intercepts(self.name, raw_request)) + if not isinstance(intercepted, dict): + raise TypeError("Relay tool request intercepts must return an object") + request = self.request_model.model_validate(intercepted) + handle = nemo_relay.tools.call(self.name, intercepted) + chunks: list[dict[str, Any]] = [] + try: + async for chunk in self.handler(request): + value = self.chunk_model.model_validate(chunk) + chunks.append(value.model_dump(mode="json")) + yield value + except BaseException: + nemo_relay.tools.call_end(handle, {"status": "interrupted", "chunks": chunks}) + raise + else: + nemo_relay.tools.call_end(handle, {"status": "ok", "chunks": chunks}) + + +__all__ = ["StreamingTool"] diff --git a/docs/source/components/agent-sdk.md b/docs/source/components/agent-sdk.md index bd480079a..f9a552c72 100644 --- a/docs/source/components/agent-sdk.md +++ b/docs/source/components/agent-sdk.md @@ -218,6 +218,13 @@ Applications use `as_agent_tool(...)` to expose any `AgentRunner` as a registered tool; foreground selection, workflow state, and background work stay explicit in application code. +`StreamingTool` keeps one Relay tool lifecycle open while a capability yields +typed response chunks. `xr_ai_nat.live_vision.LiveVisionTool` is one such tool: +it acquires a participant-scoped hub frame and calls its injected `VLMService` +through a nested Relay LLM stream. Vision is therefore not a runtime special +case or a model-control path; it is a capability implementation behind the same +tool interface used by deterministic tools and agent tools. + ## xr-ai-nat model bridge Unmigrated workflows install `xr-ai-nat[agents]` when they use NAT's built-in diff --git a/tests/test_native_tools.py b/tests/test_native_tools.py index 52335e291..3cabdf5d2 100644 --- a/tests/test_native_tools.py +++ b/tests/test_native_tools.py @@ -13,7 +13,7 @@ from nemo_relay.codecs import OpenAIChatCodec from pydantic import BaseModel from xr_ai_models import Capabilities, ChatMessage, ChatResponse, ToolCall, ToolDef -from xr_ai_nat import AgentRunner, Tool, ToolSet, as_agent_tool +from xr_ai_nat import AgentRunner, StreamingTool, Tool, ToolSet, as_agent_tool from xr_ai_nat.agents import Agent, ToolLoopLimitError, _response_from_openai @@ -42,10 +42,27 @@ class AskResult(BaseModel): text: str +class CountRequest(BaseModel): + """One requested stream count.""" + + limit: int + + +class CountChunk(BaseModel): + """One streamed count value.""" + + value: int + + async def add(request: AddRequest) -> AddResult: return AddResult(total=request.left + request.right) +async def count(request: CountRequest) -> AsyncIterator[CountChunk]: + for value in range(request.limit): + yield CountChunk(value=value) + + class _ToolCallingLLM: capabilities = Capabilities(tool_calls=True) @@ -269,6 +286,27 @@ def test_agent_accepts_null_content_for_tool_call_only_response() -> None: assert response.tool_calls == [ToolCall(id="lookup-1", name="lookup", arguments="{}")] +async def test_streaming_tools_preserve_the_native_tool_lifecycle() -> None: + tool = StreamingTool( + "count", + "Count from zero to one less than the requested limit.", + CountRequest, + CountChunk, + count, + ) + events = [] + subscriber = "xr-ai-native-streaming-tool" + nemo_relay.subscribers.register(subscriber, events.append) + try: + chunks = [chunk async for chunk in tool.stream(CountRequest(limit=3))] + await nemo_relay.subscribers.flush_async() + finally: + nemo_relay.subscribers.deregister(subscriber) + + assert chunks == [CountChunk(value=0), CountChunk(value=1), CountChunk(value=2)] + assert "tool" in {getattr(event, "category", None) for event in events} + + async def test_invalid_tool_arguments_are_returned_to_the_model_for_repair() -> None: tool = Tool("add", "Add two integers.", AddRequest, AddResult, add) diff --git a/tests/test_simple_vlm_example_worker.py b/tests/test_simple_vlm_example_worker.py index 6440e0b28..6548cf532 100644 --- a/tests/test_simple_vlm_example_worker.py +++ b/tests/test_simple_vlm_example_worker.py @@ -9,13 +9,14 @@ import time from pathlib import Path from types import SimpleNamespace +from typing import cast +import nemo_relay import pytest import tomllib import yaml -from nat.builder.workflow_builder import WorkflowBuilder -from xr_ai_hub import FrameData, FrameSignal, PixelFormat -from xr_ai_nat.functions.vision import StreamingVisionConfig +from xr_ai_hub import FrameData, FrameSignal, PixelFormat, ProcessorEndpoint +from xr_ai_models import VLMService from xr_ai_voice import VoiceQuery, VoiceSession from xr_ai_voicegate import VoiceGateConfig @@ -24,9 +25,10 @@ _WORKER_DIR = _SAMPLE_DIR / "worker" sys.path.insert(0, str(_WORKER_DIR)) -from simple_vlm_example_worker import __main__ as worker_main # noqa: E402 -from simple_vlm_example_worker import app # noqa: E402 -from simple_vlm_example_worker.config import load_config # noqa: E402 +from simple_vlm_example_worker import __main__ as worker_main # noqa: E402 # pyright: ignore[reportMissingImports] +from simple_vlm_example_worker import app # noqa: E402 # pyright: ignore[reportMissingImports] +from simple_vlm_example_worker.config import load_config # noqa: E402 # pyright: ignore[reportMissingImports] +from xr_ai_nat.live_vision import LiveVisionTool # noqa: E402 class _Service: @@ -51,44 +53,24 @@ def shutdown(self) -> None: self.shutdown_calls += 1 -class _VisionFunction: - def __init__(self) -> None: - self.requests = [] - - async def astream(self, request): - self.requests.append(request) - for text in ("a ", "blue square"): - yield SimpleNamespace(text=text) - - -class _VisionConfig: - instances: list["_VisionConfig"] = [] +class _LiveVisionTool: + instances: list["_LiveVisionTool"] = [] def __init__(self, **kwargs) -> None: self.kwargs = kwargs + self.requests = [] self.released: list[str] = [] self.instances.append(self) + async def stream(self, request): + self.requests.append(request) + for text in ("a ", "blue square"): + yield SimpleNamespace(text=text) + def release(self, participant_id: str) -> None: self.released.append(participant_id) -class _Builder: - def __init__(self, function: _VisionFunction) -> None: - self.function = function - self.added: list[tuple[str, object]] = [] - - async def __aenter__(self): - return self - - async def __aexit__(self, *_exc) -> None: - return None - - async def add_function(self, name: str, config: object): - self.added.append((name, config)) - return self.function - - class _LiveEndpoint: def __init__(self) -> None: self.frame_callback = None @@ -138,7 +120,8 @@ def test_worker_is_a_package_with_module_and_console_entry_points() -> None: assert project["tool"]["uv"]["sources"]["xr-ai-hub-client"]["path"] == ( "../../../agent-sdk/xr-ai-hub-client" ) - assert "xr-ai-nat[vision,voice]" in dependencies + assert "xr-ai-nat[relay,live-vision]" in dependencies + assert all("[vision" not in dependency and "[voice" not in dependency for dependency in dependencies) assert "xr-ai-voice" in dependencies assert "xr-ai-pipecat" not in dependencies assert all("mcp" not in dependency.lower() for dependency in dependencies) @@ -288,8 +271,6 @@ async def test_app_wires_text_voice_cleanup_readiness_and_shutdown( vlm = _Service() tts = _Service() transport = _Transport() - function = _VisionFunction() - builder = _Builder(function) sessions: list[VoiceSession] = [] text_inputs = [] run_options = {} @@ -301,8 +282,7 @@ async def test_app_wires_text_voice_cleanup_readiness_and_shutdown( monkeypatch.setattr(app, "make_stt", lambda _models, _name: stt) monkeypatch.setattr(app, "make_vlm", lambda _models, _name: vlm) monkeypatch.setattr(app, "make_tts", lambda _models, _name: tts) - monkeypatch.setattr(app, "WorkflowBuilder", lambda: builder) - monkeypatch.setattr(app, "StreamingVisionConfig", _VisionConfig) + monkeypatch.setattr(app, "LiveVisionTool", _LiveVisionTool) def make_session(**kwargs): session = VoiceSession(transport=transport, **kwargs) # type: ignore[arg-type] @@ -332,7 +312,7 @@ def __init__(self, **kwargs) -> None: monkeypatch.setattr(app, "VoiceSession", make_session) monkeypatch.setattr(app, "TextMessageInput", CaptureTextInput) - _VisionConfig.instances.clear() + _LiveVisionTool.instances.clear() await app.run_app(config, ready_file=ready_file) @@ -341,18 +321,17 @@ def __init__(self, **kwargs) -> None: assert stt.close_calls == tts.close_calls == vlm.close_calls == 1 assert transport.shutdown_calls == 1 assert sessions[0].text_topic == "vlm.response" - assert builder.added == [("perception", _VisionConfig.instances[0])] - assert _VisionConfig.instances[0].kwargs["endpoint"] is transport.endpoint - assert _VisionConfig.instances[0].kwargs["system_prompt"] == config.system_prompt - assert _VisionConfig.instances[0].kwargs["frame_max_age_s"] == ( + assert _LiveVisionTool.instances[0].kwargs["endpoint"] is transport.endpoint + assert _LiveVisionTool.instances[0].kwargs["system_prompt"] == config.system_prompt + assert _LiveVisionTool.instances[0].kwargs["frame_max_age_s"] == ( config.frame_max_age_s ) - assert _VisionConfig.instances[0].kwargs["frame_timeout_s"] == ( + assert _LiveVisionTool.instances[0].kwargs["frame_timeout_s"] == ( config.frame_timeout_s ) - assert _VisionConfig.instances[0].released == ["alice"] - assert function.requests[0].participant_id == "alice" - assert function.requests[0].query == "What is in front of me?" + assert _LiveVisionTool.instances[0].released == ["alice"] + assert _LiveVisionTool.instances[0].requests[0].participant_id == "alice" + assert _LiveVisionTool.instances[0].requests[0].query == "What is in front of me?" assert streamed == ["a ", "blue square"] assert run_options["interrupt_on_supersede"] is True assert text_inputs[0]["session"] is sessions[0] @@ -364,29 +343,31 @@ def __init__(self, **kwargs) -> None: async def test_sample_handler_streams_a_live_frame_question() -> None: endpoint = _LiveEndpoint() vlm = _StreamingVlm() - vision_config = StreamingVisionConfig( - endpoint=endpoint, - vlm=vlm, + vision = LiveVisionTool( + endpoint=cast(ProcessorEndpoint, endpoint), + vlm=cast(VLMService, vlm), system_prompt="Answer briefly.", ) - async with WorkflowBuilder() as builder: - vision = await builder.add_function("perception", vision_config) - handler = app._make_vision_handler(vision) - assert endpoint.frame_callback is not None - await endpoint.frame_callback( - FrameSignal( - slot=0, - seq=1, - pts_us=time.time_ns() // 1_000, - width=2, - height=2, - fmt=PixelFormat.RGB24, - data_sz=12, - participant_id="alice", - track_id="camera", - ) + handler = app._make_vision_handler(vision) + assert endpoint.frame_callback is not None + await endpoint.frame_callback( + FrameSignal( + slot=0, + seq=1, + pts_us=time.time_ns() // 1_000, + width=2, + height=2, + fmt=PixelFormat.RGB24, + data_sz=12, + participant_id="alice", + track_id="camera", ) + ) + events = [] + subscriber = "simple-vlm-live-vision" + nemo_relay.subscribers.register(subscriber, events.append) + try: response = await handler( VoiceQuery( participant_id="alice", @@ -396,6 +377,9 @@ async def test_sample_handler_streams_a_live_frame_question() -> None: ) ) tokens = [token async for token in response] + await nemo_relay.subscribers.flush_async() + finally: + nemo_relay.subscribers.deregister(subscriber) assert tokens == ["a ", "blue ", "square"] image, question, system_prompt = vlm.calls[0] @@ -403,3 +387,4 @@ async def test_sample_handler_streams_a_live_frame_question() -> None: assert question == "What is shown?" assert system_prompt == "Answer briefly." assert endpoint.statuses == [("processing", "alice"), ("idle", "alice")] + assert {"llm", "tool"} <= {getattr(event, "category", None) for event in events} From c3a7a38e3acc76f168b1c4ecab2870a6b71cd872 Mon Sep 17 00:00:00 2001 From: Devdeep Ray Date: Wed, 12 Aug 2026 00:08:06 +0000 Subject: [PATCH 2/3] fix(simple-vlm): honor Relay streaming boundaries Signed-off-by: Devdeep Ray --- DEPENDENCIES.md | 10 +-- agent-samples/simple-vlm-example/README.md | 5 +- .../worker/simple_vlm_example_worker/app.py | 6 +- agent-sdk/xr-ai-models/README.md | 13 ++-- .../xr_ai_models/_openai_compat.py | 6 ++ .../xr-ai-models/xr_ai_models/_protocols.py | 3 + agent-sdk/xr-ai-nat/README.md | 23 +++--- agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py | 2 - agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py | 22 ++++++ agent-sdk/xr-ai-nat/xr_ai_nat/agents.py | 14 +--- agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py | 60 +++++++++++++--- agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py | 72 ------------------- docs/changelog.md | 26 +++++-- docs/source/components/agent-sdk.md | 21 +++--- tests/test_models_openai_compat.py | 32 +++++++++ tests/test_native_tools.py | 40 +---------- tests/test_simple_vlm_example_worker.py | 62 +++++++++++----- 17 files changed, 228 insertions(+), 189 deletions(-) create mode 100644 agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py delete mode 100644 agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md index 0a3b9b6ae..6a2d73357 100644 --- a/DEPENDENCIES.md +++ b/DEPENDENCIES.md @@ -556,10 +556,12 @@ the latest video frame via streaming VLM and replies with both | Orchestrator | `simple-vlm-example` | `xr-ai-launcher` | — | | Worker | `simple-vlm-example-worker` | `xr-ai-hub-client [editable]`, `xr-ai-logging [editable]`, `xr-ai-models [editable]`, `xr-ai-nat[relay,live-vision] [editable]`, `xr-ai-voice [editable]`, `xr-ai-voicegate [editable]` | loguru >=0.7, pyyaml >=6.0 (`xr-ai-voice` pulls in VAD, pipecat-ai, numpy, and scipy; `xr-ai-nat[live-vision]` pulls in numpy and Pillow) | -The packaged worker invokes `LiveVisionTool` directly and maps its streamed -participant-scoped response to `VoiceSession`. The tool acquires a current frame -through `xr-ai-hub-client` and runs both its tool lifecycle and nested VLM stream -through NeMo Relay. `VoiceSession` owns readiness, hub transport, signals, the +The packaged worker invokes `LiveVisionResponder` directly and maps its +streamed participant-scoped response to `VoiceSession`. The responder acquires +a current frame through `xr-ai-hub-client` and runs the model call through NeMo +Relay's managed streaming LLM path under an Agent scope. Camera bytes are +redacted from Relay telemetry while the provider receives the original frame. +`VoiceSession` owns readiness, hub transport, signals, the private Pipecat pipeline, and cleanup; `TextMessageInput` routes `"ping"` and ad-hoc text through the same participant-aware path as speech. Voice-gate behavior (magic phrases, follow-up grace, listening chime, stop acknowledgement), diff --git a/agent-samples/simple-vlm-example/README.md b/agent-samples/simple-vlm-example/README.md index 57e38ef90..0f066d492 100644 --- a/agent-samples/simple-vlm-example/README.md +++ b/agent-samples/simple-vlm-example/README.md @@ -19,8 +19,9 @@ The worker is a package under `worker/simple_vlm_example_worker/`: `VoiceSession` owns STT/TTS/VLM readiness, the hub voice transport, voice-gate processing, streaming TTS, signals, and cleanup. The application invokes the -native `LiveVisionTool` directly; it acquires the participant's current frame -and streams an injected VLM through Relay. Typed text uses the same +native `LiveVisionResponder` directly; it acquires the participant's current +frame and streams an injected VLM through Relay's managed LLM path. The camera +frame is redacted from Relay telemetry. Typed text uses the same participant-aware turn path as speech. Participant leave events release cached live-frame state, and a newer turn cancels and interrupts a superseded response. diff --git a/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py b/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py index b577304a1..e1d4b7497 100644 --- a/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py +++ b/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py @@ -11,14 +11,14 @@ from loguru import logger from xr_ai_logging import setup_logging from xr_ai_models import load_models_config, make_stt, make_tts, make_vlm -from xr_ai_nat.live_vision import LiveVisionTool, VisionRequest +from xr_ai_nat.live_vision import LiveVisionResponder, VisionRequest from xr_ai_voice import TextMessageInput, VadConfig, VoiceHandler, VoiceSession from xr_ai_voicegate import load_voice_gate_config from .config import WorkerConfig -def _make_vision_handler(vision: LiveVisionTool) -> VoiceHandler: +def _make_vision_handler(vision: LiveVisionResponder) -> VoiceHandler: async def handle(turn): async def response(): async for chunk in vision.stream( @@ -69,7 +69,7 @@ async def run_app( ) async with session: - vision = LiveVisionTool( + vision = LiveVisionResponder( endpoint=session.transport.endpoint, vlm=vlm, system_prompt=config.system_prompt, diff --git a/agent-sdk/xr-ai-models/README.md b/agent-sdk/xr-ai-models/README.md index 9b634f289..b1adfe9fd 100644 --- a/agent-sdk/xr-ai-models/README.md +++ b/agent-sdk/xr-ai-models/README.md @@ -150,10 +150,13 @@ class VLMService(Protocol): capabilities: Capabilities async def ask_image(self, image, question, *, system_prompt="", max_tokens=None, temperature=None, - timeout=None) -> ChatResponse: ... + timeout=None, headers=None) -> ChatResponse: ... async def ask_video(self, video, question, *, system_prompt="", max_tokens=None, temperature=None, - timeout=None) -> ChatResponse: ... + timeout=None, headers=None) -> ChatResponse: ... + def stream(self, image, question, *, system_prompt="", + max_tokens=None, temperature=None, + timeout=None, headers=None) -> AsyncIterator[str]: ... async def health(self) -> bool: ... class STTService(Protocol): @@ -175,9 +178,9 @@ class EmbeddingService(Protocol): `reasoning_field` knob normalizes `reasoning_content` (nemotron_v3 parser) into the same surface. -`LLMService.chat` and `LLMService.stream` accept optional string-valued -per-call headers for execution context such as Relay session lineage. The -model profile remains the authority for credentials: callers cannot supply an +`LLMService` and `VLMService` request methods accept optional string-valued +per-call headers for execution context such as Relay session lineage. The model +profile remains the authority for credentials: callers cannot supply an `Authorization` header. ## Remote / hosted-NIM endpoints diff --git a/agent-sdk/xr-ai-models/xr_ai_models/_openai_compat.py b/agent-sdk/xr-ai-models/xr_ai_models/_openai_compat.py index d066308ae..d137eff0f 100644 --- a/agent-sdk/xr-ai-models/xr_ai_models/_openai_compat.py +++ b/agent-sdk/xr-ai-models/xr_ai_models/_openai_compat.py @@ -471,10 +471,12 @@ async def ask_image( max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, + headers: Mapping[str, str] | None = None, ) -> ChatResponse: return await self._llm.chat( self._build_messages(image, question, system_prompt), max_tokens=max_tokens, temperature=temperature, timeout=timeout, + headers=headers, ) def _build_video_messages( @@ -501,6 +503,7 @@ async def ask_video( max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, + headers: Mapping[str, str] | None = None, ) -> ChatResponse: if not self.capabilities.video: raise ValueError( @@ -511,6 +514,7 @@ async def ask_video( return await self._llm.chat( self._build_video_messages(video, question, system_prompt), max_tokens=max_tokens, temperature=temperature, timeout=timeout, + headers=headers, ) async def stream( @@ -522,10 +526,12 @@ async def stream( max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, + headers: Mapping[str, str] | None = None, ) -> AsyncIterator[str]: async for chunk in self._llm.stream( self._build_messages(image, question, system_prompt), max_tokens=max_tokens, temperature=temperature, timeout=timeout, + headers=headers, ): yield chunk diff --git a/agent-sdk/xr-ai-models/xr_ai_models/_protocols.py b/agent-sdk/xr-ai-models/xr_ai_models/_protocols.py index 05ac730ce..b1c93469f 100644 --- a/agent-sdk/xr-ai-models/xr_ai_models/_protocols.py +++ b/agent-sdk/xr-ai-models/xr_ai_models/_protocols.py @@ -143,6 +143,7 @@ async def ask_image( max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, + headers: Mapping[str, str] | None = None, ) -> ChatResponse: pass async def ask_video( @@ -154,6 +155,7 @@ async def ask_video( max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, + headers: Mapping[str, str] | None = None, ) -> ChatResponse: pass def stream( @@ -165,6 +167,7 @@ def stream( max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, + headers: Mapping[str, str] | None = None, ) -> AsyncIterator[str]: pass async def health(self) -> bool: pass diff --git a/agent-sdk/xr-ai-nat/README.md b/agent-sdk/xr-ai-nat/README.md index 097862107..7a952d3d9 100644 --- a/agent-sdk/xr-ai-nat/README.md +++ b/agent-sdk/xr-ai-nat/README.md @@ -60,15 +60,20 @@ custom, Fabric-backed, or framework-backed runner through the same registered invocation path. Relay observes model calls inside a tool-backed runner; the application never calls an LLM client as a separate control path. -## Live vision tool - -Install `xr-ai-nat[relay,live-vision]` for `LiveVisionTool`. It is a normal -streaming tool: its request carries a participant ID and question, it acquires a -fresh frame through `xr-ai-hub-client`, and it calls an injected `VLMService` -through a nested Relay LLM stream. `release(participant_id)` clears cached frame -state when the participant leaves. Voice applications can map its chunks to a -`VoiceHandler`; the event dispatcher follow-up invokes this same tool from voice -and autonomous triggers. +## Live vision responder + +Install `xr-ai-nat[relay,live-vision]` for `LiveVisionResponder`. Its request +carries a participant ID and question, it acquires a fresh frame through +`xr-ai-hub-client`, and it calls an injected `VLMService` through Relay's +managed streaming LLM path. Each response has an Agent scope; the inline camera +frame is redacted from Relay events while the unmodified frame reaches the +provider. `release(participant_id)` clears cached frame state when the +participant leaves. Voice applications map its chunks to a `VoiceHandler`. + +Relay's managed tool API accepts completed JSON results, while its managed LLM +API supports streaming. Finite native tools therefore use `Tool.execute()`; +real-time model responses remain application-owned streams with Relay managing +the nested model call. ## Legacy NAT compatibility diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py b/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py index 3e06eaf08..7938b7866 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py @@ -4,12 +4,10 @@ """Toolkit-independent native XR tools with legacy NAT compatibility.""" from .agent_runner import AgentRunner, as_agent_tool -from .streaming import StreamingTool from .tools import Tool, ToolInvocationResult, ToolSet __all__ = [ "AgentRunner", - "StreamingTool", "Tool", "ToolInvocationResult", "ToolSet", diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py b/agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py new file mode 100644 index 000000000..a01333bb9 --- /dev/null +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared validation at the NeMo Relay service boundary.""" + +from __future__ import annotations + + +def headers_from_relay(raw: object) -> dict[str, str]: + """Return string-only headers supplied by a Relay request intercept.""" + + if not isinstance(raw, dict): + raise TypeError("Relay LLM request headers must be an object") + headers: dict[str, str] = {} + for name, value in raw.items(): + if not isinstance(name, str) or not isinstance(value, str): + raise TypeError("Relay LLM request headers must be strings") + headers[name] = value + return headers + + +__all__ = ["headers_from_relay"] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py b/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py index c98b8cf65..82331ccd7 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py @@ -14,6 +14,7 @@ from nemo_relay.codecs import OpenAIChatCodec from xr_ai_models import ChatMessage, ChatResponse, LLMService, ToolCall, ToolDef +from ._relay import headers_from_relay from .agent_runner import AgentRunner, as_agent_tool from .tools import Tool, ToolSet @@ -137,7 +138,7 @@ async def invoke(request: nemo_relay.LLMRequest) -> dict[str, Any]: temperature=_optional_float(content.get("temperature")), enable_thinking=bool(content.get("enable_thinking", False)), thinking_budget=_optional_int(content.get("thinking_budget")), - headers=_headers_from_relay(request.headers), + headers=headers_from_relay(request.headers), ) return _response_to_openai(response) @@ -243,17 +244,6 @@ def _tools_from_openai(raw: object) -> list[ToolDef] | None: return definitions -def _headers_from_relay(raw: object) -> dict[str, str]: - if not isinstance(raw, dict): - raise TypeError("Relay LLM request headers must be an object") - headers: dict[str, str] = {} - for name, value in raw.items(): - if not isinstance(name, str) or not isinstance(value, str): - raise TypeError("Relay LLM request headers must be strings") - headers[name] = value - return headers - - def _response_to_openai(response: ChatResponse) -> dict[str, Any]: message = _message_to_openai( ChatMessage(role="assistant", content=response.content, tool_calls=response.tool_calls), diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py index c1a11edcc..bd1cffea7 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py @@ -1,11 +1,12 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""A normal streaming tool for questions about a participant's current frame.""" +"""A Relay-observed streaming responder for a participant's current frame.""" from __future__ import annotations import asyncio +import copy import logging from collections.abc import AsyncIterator, Mapping from typing import Any @@ -17,10 +18,11 @@ from xr_ai_models import VLMService from ._pixels import encode_image, frame_to_pil -from .streaming import StreamingTool +from ._relay import headers_from_relay _LOGGER = logging.getLogger(__name__) _VLM_CALL_NAME = "xr-ai-vlm" +_FRAME_REDACTION = "" class VisionRequest(BaseModel): @@ -38,8 +40,8 @@ class VisionChunk(BaseModel): text: str = Field(description="A partial fragment of the streamed answer text.") -class LiveVisionTool(StreamingTool[VisionRequest, VisionChunk]): - """A participant-scoped current-frame VLM tool with an injected model service.""" +class LiveVisionResponder: + """Stream one participant-scoped answer through Relay's managed LLM path.""" def __init__( self, @@ -62,13 +64,24 @@ def __init__( max_age_s=frame_max_age_s, timeout_s=frame_timeout_s, ) - super().__init__( + + async def stream(self, request: VisionRequest) -> AsyncIterator[VisionChunk]: + """Run one live-vision turn without exposing camera bytes to telemetry.""" + + request = VisionRequest.model_validate(request) + with nemo_relay.scope.scope( "look_at_current_frame", - "Answer a question about a participant's current live camera view.", - VisionRequest, - VisionChunk, - self._stream_current, - ) + nemo_relay.ScopeType.Agent, + input=request.model_dump(mode="json"), + ) as handle: + nemo_relay.scope_local.register_llm_sanitize_request( + handle, + "xr-ai-live-frame", + 0, + _sanitize_live_frame, + ) + async for chunk in self._stream_current(request): + yield VisionChunk.model_validate(chunk) def release(self, participant_id: str) -> None: """Forget cached frame state after a participant disconnects.""" @@ -140,10 +153,35 @@ async def _stream_vlm( image_url, query, system_prompt=system_prompt, + headers=headers_from_relay(request.headers), ): yield {"choices": [{"delta": {"content": token}}]} +def _sanitize_live_frame( + request: nemo_relay.LLMRequest, + _context: nemo_relay.LlmSanitizeRequestContext, +) -> nemo_relay.LLMRequest: + """Redact inline camera data from events without changing provider input.""" + + content = copy.deepcopy(request.content) + messages = content.get("messages") + if isinstance(messages, list): + for message in messages: + if not isinstance(message, dict): + continue + parts = message.get("content") + if not isinstance(parts, list): + continue + for part in parts: + if not isinstance(part, dict) or part.get("type") != "image_url": + continue + image = part.get("image_url") + if isinstance(image, dict) and isinstance(image.get("url"), str): + image["url"] = _FRAME_REDACTION + return nemo_relay.LLMRequest(dict(request.headers), content) + + def _vision_inputs(content: Mapping[str, object]) -> tuple[str, str, str]: messages = content.get("messages") if not isinstance(messages, list): @@ -217,4 +255,4 @@ def _stream_text(raw_chunk: object) -> str: return content -__all__ = ["LiveVisionTool", "VisionChunk", "VisionRequest"] +__all__ = ["LiveVisionResponder", "VisionChunk", "VisionRequest"] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py b/agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py deleted file mode 100644 index 2bb5d8d42..000000000 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/streaming.py +++ /dev/null @@ -1,72 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -"""Typed streaming tools for trigger-driven native application paths.""" - -from __future__ import annotations - -from collections.abc import AsyncIterator, Awaitable, Callable -from inspect import isawaitable -from typing import Any, Generic, TypeVar, cast - -import nemo_relay -from pydantic import BaseModel - -RequestT = TypeVar("RequestT", bound=BaseModel) -ChunkT = TypeVar("ChunkT", bound=BaseModel) -ValueT = TypeVar("ValueT") - - -async def _resolve(value: ValueT | Awaitable[ValueT]) -> ValueT: - """Normalize Relay's sync-or-async hook contract for an async tool path.""" - - if isawaitable(value): - return await cast(Awaitable[ValueT], value) - return value - - -class StreamingTool(Generic[RequestT, ChunkT]): - """A typed tool that yields Pydantic chunks while keeping one Relay tool span open.""" - - def __init__( - self, - name: str, - description: str, - request_model: type[RequestT], - chunk_model: type[ChunkT], - handler: Callable[[RequestT], AsyncIterator[ChunkT]], - ) -> None: - if not name: - raise ValueError("tool name must not be empty") - if not description: - raise ValueError(f"tool {name!r} needs a description") - self.name = name - self.description = description - self.request_model = request_model - self.chunk_model = chunk_model - self.handler = handler - - async def stream(self, request: RequestT) -> AsyncIterator[ChunkT]: - """Yield one validated tool stream under a Relay lifecycle span.""" - - raw_request = request.model_dump(mode="json") - await _resolve(nemo_relay.tools.conditional_execution(self.name, raw_request)) - intercepted = await _resolve(nemo_relay.tools.request_intercepts(self.name, raw_request)) - if not isinstance(intercepted, dict): - raise TypeError("Relay tool request intercepts must return an object") - request = self.request_model.model_validate(intercepted) - handle = nemo_relay.tools.call(self.name, intercepted) - chunks: list[dict[str, Any]] = [] - try: - async for chunk in self.handler(request): - value = self.chunk_model.model_validate(chunk) - chunks.append(value.model_dump(mode="json")) - yield value - except BaseException: - nemo_relay.tools.call_end(handle, {"status": "interrupted", "chunks": chunks}) - raise - else: - nemo_relay.tools.call_end(handle, {"status": "ok", "chunks": chunks}) - - -__all__ = ["StreamingTool"] diff --git a/docs/changelog.md b/docs/changelog.md index b7230b29d..a7e3613da 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -9,6 +9,22 @@ Significant decisions, in reverse-chronological order. Update this whenever a non-trivial architectural or design decision is made so the rationale is preserved and not re-litigated. +### 2026-08-12 — Real-time responders use Relay's managed streaming boundary + +Relay's managed tool API accepts completed JSON results; it does not define a +streaming tool execution contract. XR AI does not reproduce Relay's guardrail +and intercept pipeline around an async generator. Finite tools continue through +`Tool.execute()`, while real-time model responses remain application-owned +streams under an Agent scope and send the actual provider call through Relay's +managed streaming LLM API. This preserves low-latency TTS and gives configured +LLM request, streaming-execution, sanitization, and observability middleware one +authoritative execution path. + +Live camera frames remain provider input but are replaced in emitted Relay +events by a scope-local sanitizer. Relay request-intercept headers cross the +`VLMService` boundary as controlled per-call headers; model-profile credentials +remain non-overridable. + ### 2026-08-11 — Native tools own agent composition; Relay owns their execution NeMo Agent Toolkit is being retired from XR AI in focused migrations rather @@ -16,11 +32,11 @@ than a framework-wide swap. `xr-ai-nat` is the public, toolkit-independent native tools layer: typed tools, an `AgentRunner` async-turn protocol, and a bounded default tool loop. A model is a private implementation dependency of a model-backed tool or agent, reached only through `xr-ai-models`; voice, text, -and background triggers invoke registered tools rather than model clients. NeMo -Relay runs the tool and model lifecycles, supplying middleware, guardrails, and -telemetry. `as_agent_tool` lets a custom or future Fabric-backed runner use the -same registered-tool path without making a framework part of the public trigger -boundary. +and background triggers invoke native tools or responders rather than model +clients. NeMo Relay runs finite tool and model lifecycles, supplying middleware, +guardrails, and telemetry. `as_agent_tool` lets a custom or future Fabric-backed +runner use the same registered-tool path without making a framework part of the +public trigger boundary. The existing NeMo Agent Toolkit function groups remain behind legacy extras while they migrate. Relay does not own XR application routing, participant diff --git a/docs/source/components/agent-sdk.md b/docs/source/components/agent-sdk.md index f9a552c72..0e70b4003 100644 --- a/docs/source/components/agent-sdk.md +++ b/docs/source/components/agent-sdk.md @@ -149,10 +149,13 @@ class VLMService(Protocol): capabilities: Capabilities async def ask_image(self, image, question, *, system_prompt="", max_tokens=None, temperature=None, - timeout=None) -> ChatResponse: ... + timeout=None, headers=None) -> ChatResponse: ... async def ask_video(self, video, question, *, system_prompt="", max_tokens=None, temperature=None, - timeout=None) -> ChatResponse: ... + timeout=None, headers=None) -> ChatResponse: ... + def stream(self, image, question, *, system_prompt="", + max_tokens=None, temperature=None, + timeout=None, headers=None) -> AsyncIterator[str]: ... async def health(self) -> bool: ... class STTService(Protocol): @@ -218,12 +221,14 @@ Applications use `as_agent_tool(...)` to expose any `AgentRunner` as a registered tool; foreground selection, workflow state, and background work stay explicit in application code. -`StreamingTool` keeps one Relay tool lifecycle open while a capability yields -typed response chunks. `xr_ai_nat.live_vision.LiveVisionTool` is one such tool: -it acquires a participant-scoped hub frame and calls its injected `VLMService` -through a nested Relay LLM stream. Vision is therefore not a runtime special -case or a model-control path; it is a capability implementation behind the same -tool interface used by deterministic tools and agent tools. +`xr_ai_nat.live_vision.LiveVisionResponder` acquires a participant-scoped hub +frame and calls its injected `VLMService` through Relay's managed streaming LLM +path. Each response runs under an Agent scope, and a scope-local sanitizer +replaces the inline camera frame in Relay events without changing provider +input. Relay's managed tool API accepts completed JSON values, so finite tools +use `Tool.execute()` while real-time model output remains an application-owned +stream. This keeps Relay middleware on a supported managed boundary rather than +reimplementing a partial streaming-tool lifecycle. ## xr-ai-nat model bridge diff --git a/tests/test_models_openai_compat.py b/tests/test_models_openai_compat.py index 7cc50424e..8e55f1a72 100644 --- a/tests/test_models_openai_compat.py +++ b/tests/test_models_openai_compat.py @@ -448,6 +448,38 @@ async def test_vlm_default_extras_propagate() -> None: assert body["chat_template_kwargs"] == {"enable_thinking": False} +async def test_vlm_stream_forwards_controlled_per_call_headers() -> None: + stub = StubOpenAI() + stub.set_stream_tokens(["visible"]) + async with OpenAICompatVLM( + "http://stub", "vlm", client=stub.client(), + ) as vlm: + chunks = [ + chunk + async for chunk in vlm.stream( + _PNG_HEADER, + "?", + headers={"X-Relay-Session": "turn-7"}, + ) + ] + + assert chunks == ["visible"] + assert stub.last_request().headers["X-Relay-Session"] == "turn-7" + + +async def test_vlm_rejects_per_call_authorization_header() -> None: + stub = StubOpenAI() + async with OpenAICompatVLM( + "http://stub", "vlm", client=stub.client(), + ) as vlm: + with pytest.raises(ValueError, match="cannot override Authorization"): + await vlm.ask_image( + _PNG_HEADER, + "?", + headers={"Authorization": "Bearer untrusted"}, + ) + + # ── VLM: video ──────────────────────────────────────────────────────────── diff --git a/tests/test_native_tools.py b/tests/test_native_tools.py index 3cabdf5d2..52335e291 100644 --- a/tests/test_native_tools.py +++ b/tests/test_native_tools.py @@ -13,7 +13,7 @@ from nemo_relay.codecs import OpenAIChatCodec from pydantic import BaseModel from xr_ai_models import Capabilities, ChatMessage, ChatResponse, ToolCall, ToolDef -from xr_ai_nat import AgentRunner, StreamingTool, Tool, ToolSet, as_agent_tool +from xr_ai_nat import AgentRunner, Tool, ToolSet, as_agent_tool from xr_ai_nat.agents import Agent, ToolLoopLimitError, _response_from_openai @@ -42,27 +42,10 @@ class AskResult(BaseModel): text: str -class CountRequest(BaseModel): - """One requested stream count.""" - - limit: int - - -class CountChunk(BaseModel): - """One streamed count value.""" - - value: int - - async def add(request: AddRequest) -> AddResult: return AddResult(total=request.left + request.right) -async def count(request: CountRequest) -> AsyncIterator[CountChunk]: - for value in range(request.limit): - yield CountChunk(value=value) - - class _ToolCallingLLM: capabilities = Capabilities(tool_calls=True) @@ -286,27 +269,6 @@ def test_agent_accepts_null_content_for_tool_call_only_response() -> None: assert response.tool_calls == [ToolCall(id="lookup-1", name="lookup", arguments="{}")] -async def test_streaming_tools_preserve_the_native_tool_lifecycle() -> None: - tool = StreamingTool( - "count", - "Count from zero to one less than the requested limit.", - CountRequest, - CountChunk, - count, - ) - events = [] - subscriber = "xr-ai-native-streaming-tool" - nemo_relay.subscribers.register(subscriber, events.append) - try: - chunks = [chunk async for chunk in tool.stream(CountRequest(limit=3))] - await nemo_relay.subscribers.flush_async() - finally: - nemo_relay.subscribers.deregister(subscriber) - - assert chunks == [CountChunk(value=0), CountChunk(value=1), CountChunk(value=2)] - assert "tool" in {getattr(event, "category", None) for event in events} - - async def test_invalid_tool_arguments_are_returned_to_the_model_for_repair() -> None: tool = Tool("add", "Add two integers.", AddRequest, AddResult, add) diff --git a/tests/test_simple_vlm_example_worker.py b/tests/test_simple_vlm_example_worker.py index 6548cf532..13750bb22 100644 --- a/tests/test_simple_vlm_example_worker.py +++ b/tests/test_simple_vlm_example_worker.py @@ -28,7 +28,7 @@ from simple_vlm_example_worker import __main__ as worker_main # noqa: E402 # pyright: ignore[reportMissingImports] from simple_vlm_example_worker import app # noqa: E402 # pyright: ignore[reportMissingImports] from simple_vlm_example_worker.config import load_config # noqa: E402 # pyright: ignore[reportMissingImports] -from xr_ai_nat.live_vision import LiveVisionTool # noqa: E402 +from xr_ai_nat.live_vision import LiveVisionResponder # noqa: E402 class _Service: @@ -53,8 +53,8 @@ def shutdown(self) -> None: self.shutdown_calls += 1 -class _LiveVisionTool: - instances: list["_LiveVisionTool"] = [] +class _LiveVisionResponder: + instances: list["_LiveVisionResponder"] = [] def __init__(self, **kwargs) -> None: self.kwargs = kwargs @@ -102,8 +102,15 @@ class _StreamingVlm: def __init__(self) -> None: self.calls = [] - async def stream(self, image, question: str, *, system_prompt: str = ""): - self.calls.append((image, question, system_prompt)) + async def stream( + self, + image, + question: str, + *, + system_prompt: str = "", + headers=None, + ): + self.calls.append((image, question, system_prompt, dict(headers or {}))) for token in ("a ", "blue ", "square"): yield token @@ -282,7 +289,7 @@ async def test_app_wires_text_voice_cleanup_readiness_and_shutdown( monkeypatch.setattr(app, "make_stt", lambda _models, _name: stt) monkeypatch.setattr(app, "make_vlm", lambda _models, _name: vlm) monkeypatch.setattr(app, "make_tts", lambda _models, _name: tts) - monkeypatch.setattr(app, "LiveVisionTool", _LiveVisionTool) + monkeypatch.setattr(app, "LiveVisionResponder", _LiveVisionResponder) def make_session(**kwargs): session = VoiceSession(transport=transport, **kwargs) # type: ignore[arg-type] @@ -312,7 +319,7 @@ def __init__(self, **kwargs) -> None: monkeypatch.setattr(app, "VoiceSession", make_session) monkeypatch.setattr(app, "TextMessageInput", CaptureTextInput) - _LiveVisionTool.instances.clear() + _LiveVisionResponder.instances.clear() await app.run_app(config, ready_file=ready_file) @@ -321,17 +328,17 @@ def __init__(self, **kwargs) -> None: assert stt.close_calls == tts.close_calls == vlm.close_calls == 1 assert transport.shutdown_calls == 1 assert sessions[0].text_topic == "vlm.response" - assert _LiveVisionTool.instances[0].kwargs["endpoint"] is transport.endpoint - assert _LiveVisionTool.instances[0].kwargs["system_prompt"] == config.system_prompt - assert _LiveVisionTool.instances[0].kwargs["frame_max_age_s"] == ( + assert _LiveVisionResponder.instances[0].kwargs["endpoint"] is transport.endpoint + assert _LiveVisionResponder.instances[0].kwargs["system_prompt"] == config.system_prompt + assert _LiveVisionResponder.instances[0].kwargs["frame_max_age_s"] == ( config.frame_max_age_s ) - assert _LiveVisionTool.instances[0].kwargs["frame_timeout_s"] == ( + assert _LiveVisionResponder.instances[0].kwargs["frame_timeout_s"] == ( config.frame_timeout_s ) - assert _LiveVisionTool.instances[0].released == ["alice"] - assert _LiveVisionTool.instances[0].requests[0].participant_id == "alice" - assert _LiveVisionTool.instances[0].requests[0].query == "What is in front of me?" + assert _LiveVisionResponder.instances[0].released == ["alice"] + assert _LiveVisionResponder.instances[0].requests[0].participant_id == "alice" + assert _LiveVisionResponder.instances[0].requests[0].query == "What is in front of me?" assert streamed == ["a ", "blue square"] assert run_options["interrupt_on_supersede"] is True assert text_inputs[0]["session"] is sessions[0] @@ -343,7 +350,7 @@ def __init__(self, **kwargs) -> None: async def test_sample_handler_streams_a_live_frame_question() -> None: endpoint = _LiveEndpoint() vlm = _StreamingVlm() - vision = LiveVisionTool( + vision = LiveVisionResponder( endpoint=cast(ProcessorEndpoint, endpoint), vlm=cast(VLMService, vlm), system_prompt="Answer briefly.", @@ -366,7 +373,18 @@ async def test_sample_handler_streams_a_live_frame_question() -> None: ) events = [] subscriber = "simple-vlm-live-vision" + intercept = "simple-vlm-live-vision-header" + + def add_header(_name, request, annotated): + headers = dict(request.headers) + headers["X-Relay-Session"] = "turn-7" + return nemo_relay.LLMRequestInterceptOutcome( + nemo_relay.LLMRequest(headers, request.content), + annotated, + ) + nemo_relay.subscribers.register(subscriber, events.append) + nemo_relay.intercepts.register_llm_request(intercept, 0, False, add_header) try: response = await handler( VoiceQuery( @@ -379,12 +397,22 @@ async def test_sample_handler_streams_a_live_frame_question() -> None: tokens = [token async for token in response] await nemo_relay.subscribers.flush_async() finally: + nemo_relay.intercepts.deregister_llm_request(intercept) nemo_relay.subscribers.deregister(subscriber) assert tokens == ["a ", "blue ", "square"] - image, question, system_prompt = vlm.calls[0] + image, question, system_prompt, headers = vlm.calls[0] assert image.startswith("data:image/jpeg;base64,") assert question == "What is shown?" assert system_prompt == "Answer briefly." + assert headers["X-Relay-Session"] == "turn-7" assert endpoint.statuses == [("processing", "alice"), ("idle", "alice")] - assert {"llm", "tool"} <= {getattr(event, "category", None) for event in events} + assert {"agent", "llm"} <= {getattr(event, "category", None) for event in events} + llm_events = [ + event.to_json() + for event in events + if getattr(event, "category", None) == "llm" + ] + assert llm_events + assert all(image not in event for event in llm_events) + assert any("" in event for event in llm_events) From fcefc3994ccf0c14de51828f671e558f1dcf1c04 Mon Sep 17 00:00:00 2001 From: Devdeep Ray Date: Wed, 12 Aug 2026 00:32:28 +0000 Subject: [PATCH 3/3] refactor(vision): separate agent and voice paths Signed-off-by: Devdeep Ray --- DEPENDENCIES.md | 11 +- agent-samples/simple-vlm-example/README.md | 9 +- .../worker/simple_vlm_example_worker/app.py | 7 +- agent-sdk/xr-ai-nat/README.md | 23 ++-- agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py | 130 ++++++++++++++---- docs/changelog.md | 14 +- docs/nemo-agent-toolkit-migration.md | 5 +- docs/source/components/agent-sdk.md | 15 +- tests/test_simple_vlm_example_worker.py | 123 ++++++++++++++--- 9 files changed, 256 insertions(+), 81 deletions(-) diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md index 6a2d73357..4b1b65ed0 100644 --- a/DEPENDENCIES.md +++ b/DEPENDENCIES.md @@ -556,11 +556,12 @@ the latest video frame via streaming VLM and replies with both | Orchestrator | `simple-vlm-example` | `xr-ai-launcher` | — | | Worker | `simple-vlm-example-worker` | `xr-ai-hub-client [editable]`, `xr-ai-logging [editable]`, `xr-ai-models [editable]`, `xr-ai-nat[relay,live-vision] [editable]`, `xr-ai-voice [editable]`, `xr-ai-voicegate [editable]` | loguru >=0.7, pyyaml >=6.0 (`xr-ai-voice` pulls in VAD, pipecat-ai, numpy, and scipy; `xr-ai-nat[live-vision]` pulls in numpy and Pillow) | -The packaged worker invokes `LiveVisionResponder` directly and maps its -streamed participant-scoped response to `VoiceSession`. The responder acquires -a current frame through `xr-ai-hub-client` and runs the model call through NeMo -Relay's managed streaming LLM path under an Agent scope. Camera bytes are -redacted from Relay telemetry while the provider receives the original frame. +The packaged worker constructs the finite `LiveVisionTool` used by agentic +flows, then maps its separate direct-voice `LiveVisionResponder` to +`VoiceSession`. They share current-frame acquisition through +`xr-ai-hub-client`; only the direct voice response uses NeMo Relay's managed +streaming LLM path under an Agent scope. Camera bytes are redacted from Relay +telemetry while the provider receives the original frame. `VoiceSession` owns readiness, hub transport, signals, the private Pipecat pipeline, and cleanup; `TextMessageInput` routes `"ping"` and ad-hoc text through the same participant-aware path as speech. Voice-gate diff --git a/agent-samples/simple-vlm-example/README.md b/agent-samples/simple-vlm-example/README.md index 0f066d492..0e6a3ede5 100644 --- a/agent-samples/simple-vlm-example/README.md +++ b/agent-samples/simple-vlm-example/README.md @@ -18,10 +18,11 @@ The worker is a package under `worker/simple_vlm_example_worker/`: - `prompts/system.txt` owns the VLM system prompt. `VoiceSession` owns STT/TTS/VLM readiness, the hub voice transport, voice-gate -processing, streaming TTS, signals, and cleanup. The application invokes the -native `LiveVisionResponder` directly; it acquires the participant's current -frame and streams an injected VLM through Relay's managed LLM path. The camera -frame is redacted from Relay telemetry. Typed text uses the same +processing, streaming TTS, signals, and cleanup. The application constructs a +finite `LiveVisionTool` and maps its separate `LiveVisionResponder` to voice; +both share participant frame acquisition while only the direct voice response +streams through Relay's managed LLM path. The camera frame is redacted from +Relay telemetry. Typed text uses the same participant-aware turn path as speech. Participant leave events release cached live-frame state, and a newer turn cancels and interrupts a superseded response. diff --git a/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py b/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py index e1d4b7497..f2b2517a5 100644 --- a/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py +++ b/agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py @@ -11,7 +11,7 @@ from loguru import logger from xr_ai_logging import setup_logging from xr_ai_models import load_models_config, make_stt, make_tts, make_vlm -from xr_ai_nat.live_vision import LiveVisionResponder, VisionRequest +from xr_ai_nat.live_vision import LiveVisionResponder, LiveVisionTool, VisionRequest from xr_ai_voice import TextMessageInput, VadConfig, VoiceHandler, VoiceSession from xr_ai_voicegate import load_voice_gate_config @@ -69,13 +69,14 @@ async def run_app( ) async with session: - vision = LiveVisionResponder( + vision = LiveVisionTool( endpoint=session.transport.endpoint, vlm=vlm, system_prompt=config.system_prompt, frame_max_age_s=config.frame_max_age_s, frame_timeout_s=config.frame_timeout_s, ) + voice = LiveVisionResponder(vision) TextMessageInput( session=session, transform=_text_transform(config.default_prompt), @@ -84,7 +85,7 @@ async def run_app( logger.info("simple-vlm-example starting") await session.run( - _make_vision_handler(vision), + _make_vision_handler(voice), on_participant_left=vision.release, interrupt_on_supersede=True, ) diff --git a/agent-sdk/xr-ai-nat/README.md b/agent-sdk/xr-ai-nat/README.md index 7a952d3d9..9b8992047 100644 --- a/agent-sdk/xr-ai-nat/README.md +++ b/agent-sdk/xr-ai-nat/README.md @@ -60,20 +60,21 @@ custom, Fabric-backed, or framework-backed runner through the same registered invocation path. Relay observes model calls inside a tool-backed runner; the application never calls an LLM client as a separate control path. -## Live vision responder +## Live vision tool and direct voice responder -Install `xr-ai-nat[relay,live-vision]` for `LiveVisionResponder`. Its request -carries a participant ID and question, it acquires a fresh frame through -`xr-ai-hub-client`, and it calls an injected `VLMService` through Relay's -managed streaming LLM path. Each response has an Agent scope; the inline camera -frame is redacted from Relay events while the unmodified frame reaches the -provider. `release(participant_id)` clears cached frame state when the -participant leaves. Voice applications map its chunks to a `VoiceHandler`. +Install `xr-ai-nat[relay,live-vision]` for `LiveVisionTool`. The finite +`look_at_current_frame` tool acquires a participant's current frame and returns +one complete `VisionResponse` for agentic planning. `LiveVisionResponder` +shares that tool's frame source and streams only the direct voice path. Both +call an injected `VLMService` through Relay's matching managed LLM boundary, +forward controlled Relay headers, and redact the inline camera frame from +events while the provider receives the original. `LiveVisionTool.release()` +clears participant frame state. Relay's managed tool API accepts completed JSON results, while its managed LLM -API supports streaming. Finite native tools therefore use `Tool.execute()`; -real-time model responses remain application-owned streams with Relay managing -the nested model call. +API supports streaming. Agentic vision therefore uses `Tool.execute()` and a +complete result; direct voice remains an application-owned response stream +with Relay managing the nested model call. ## Legacy NAT compatibility diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py index bd1cffea7..002b9ca5c 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""A Relay-observed streaming responder for a participant's current frame.""" +"""Current-frame vision for agent tools and direct voice responses.""" from __future__ import annotations @@ -19,6 +19,7 @@ from ._pixels import encode_image, frame_to_pil from ._relay import headers_from_relay +from .tools import Tool _LOGGER = logging.getLogger(__name__) _VLM_CALL_NAME = "xr-ai-vlm" @@ -34,14 +35,20 @@ class VisionRequest(BaseModel): query: str = Field(min_length=1, description="Question to answer from the camera frame.") +class VisionResponse(BaseModel): + """A complete answer about the current frame.""" + + text: str = Field(description="Complete answer text.") + + class VisionChunk(BaseModel): - """One streamed text fragment from a current-frame answer.""" + """One text fragment from a direct voice answer.""" - text: str = Field(description="A partial fragment of the streamed answer text.") + text: str = Field(description="A partial fragment of the answer text.") -class LiveVisionResponder: - """Stream one participant-scoped answer through Relay's managed LLM path.""" +class LiveVisionTool(Tool[VisionRequest, VisionResponse]): + """A finite current-frame tool for agentic planning and tool loops.""" def __init__( self, @@ -64,30 +71,47 @@ def __init__( max_age_s=frame_max_age_s, timeout_s=frame_timeout_s, ) - - async def stream(self, request: VisionRequest) -> AsyncIterator[VisionChunk]: - """Run one live-vision turn without exposing camera bytes to telemetry.""" - - request = VisionRequest.model_validate(request) - with nemo_relay.scope.scope( + super().__init__( "look_at_current_frame", - nemo_relay.ScopeType.Agent, - input=request.model_dump(mode="json"), - ) as handle: - nemo_relay.scope_local.register_llm_sanitize_request( - handle, - "xr-ai-live-frame", - 0, - _sanitize_live_frame, - ) - async for chunk in self._stream_current(request): - yield VisionChunk.model_validate(chunk) + "Answer a question about a participant's current live camera view.", + VisionRequest, + VisionResponse, + self._answer_current, + render_result=lambda result: result.text, + ) def release(self, participant_id: str) -> None: """Forget cached frame state after a participant disconnects.""" self.frames.release(participant_id) + async def _answer_current(self, request: VisionRequest) -> VisionResponse: + try: + image_url = await self._current_image(request.participant_id) + except FrameUnavailable as exc: + return VisionResponse(text=str(exc)) + except Exception: + _LOGGER.exception("Live frame conversion failed") + return VisionResponse(text="VLM server unavailable — please retry.") + + await self.endpoint.set_status("processing", request.participant_id) + try: + _register_frame_sanitizer() + response = await nemo_relay.llm.execute( + _VLM_CALL_NAME, + self._relay_request(image_url, request.query), + self._ask_vlm, + model_name=_VLM_CALL_NAME, + codec=OpenAIChatCodec(), + response_codec=OpenAIChatCodec(), + ) + return VisionResponse(text=_response_text(response)) + except Exception: + _LOGGER.exception("Live VLM request failed") + return VisionResponse(text="VLM server unavailable — please retry.") + finally: + await self.endpoint.set_status("idle", request.participant_id) + async def _stream_current(self, request: VisionRequest) -> AsyncIterator[VisionChunk]: try: image_url = await self._current_image(request.participant_id) @@ -102,6 +126,7 @@ async def _stream_current(self, request: VisionRequest) -> AsyncIterator[VisionC await self.endpoint.set_status("processing", request.participant_id) fragments: list[str] = [] try: + _register_frame_sanitizer() stream = await nemo_relay.llm.stream_execute( _VLM_CALL_NAME, self._relay_request(image_url, request.query), @@ -144,6 +169,16 @@ def _relay_request(self, image_url: str, query: str) -> nemo_relay.LLMRequest: }, ) + async def _ask_vlm(self, request: nemo_relay.LLMRequest) -> dict[str, Any]: + image_url, query, system_prompt = _vision_inputs(request.content) + text = await self.vlm.ask_image( + image_url, + query, + system_prompt=system_prompt, + headers=headers_from_relay(request.headers), + ) + return _openai_response(text.content) + async def _stream_vlm( self, request: nemo_relay.LLMRequest, @@ -158,6 +193,34 @@ async def _stream_vlm( yield {"choices": [{"delta": {"content": token}}]} +class LiveVisionResponder: + """Stream a direct voice answer without turning the stream into a tool result.""" + + def __init__(self, tool: LiveVisionTool) -> None: + self.tool = tool + + async def stream(self, request: VisionRequest) -> AsyncIterator[VisionChunk]: + """Run one participant-scoped direct voice response.""" + + request = VisionRequest.model_validate(request) + with nemo_relay.scope.scope( + "speak_about_current_frame", + nemo_relay.ScopeType.Agent, + input=request.model_dump(mode="json"), + ): + async for chunk in self.tool._stream_current(request): + yield chunk + + +def _register_frame_sanitizer() -> None: + nemo_relay.scope_local.register_llm_sanitize_request( + nemo_relay.scope.get_handle(), + "xr-ai-live-frame", + 0, + _sanitize_live_frame, + ) + + def _sanitize_live_frame( request: nemo_relay.LLMRequest, _context: nemo_relay.LlmSanitizeRequestContext, @@ -240,6 +303,21 @@ def _openai_response(text: str) -> dict[str, Any]: } +def _response_text(raw_response: object) -> str: + if not isinstance(raw_response, Mapping): + raise TypeError("Relay VLM response must be an object") + choices = raw_response.get("choices") + if not isinstance(choices, list) or not choices or not isinstance(choices[0], Mapping): + raise ValueError("Relay VLM response must contain one choice") + message = choices[0].get("message") + if not isinstance(message, Mapping): + raise TypeError("Relay VLM response choice must contain a message") + content = message.get("content", "") + if not isinstance(content, str): + raise TypeError("Relay VLM response content must be text") + return content + + def _stream_text(raw_chunk: object) -> str: if not isinstance(raw_chunk, Mapping): raise TypeError("Relay VLM stream chunk must be an object") @@ -255,4 +333,10 @@ def _stream_text(raw_chunk: object) -> str: return content -__all__ = ["LiveVisionResponder", "VisionChunk", "VisionRequest"] +__all__ = [ + "LiveVisionResponder", + "LiveVisionTool", + "VisionChunk", + "VisionRequest", + "VisionResponse", +] diff --git a/docs/changelog.md b/docs/changelog.md index a7e3613da..04059bf4f 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -9,16 +9,16 @@ Significant decisions, in reverse-chronological order. Update this whenever a non-trivial architectural or design decision is made so the rationale is preserved and not re-litigated. -### 2026-08-12 — Real-time responders use Relay's managed streaming boundary +### 2026-08-12 — Agentic vision is finite; direct voice may stream Relay's managed tool API accepts completed JSON results; it does not define a streaming tool execution contract. XR AI does not reproduce Relay's guardrail -and intercept pipeline around an async generator. Finite tools continue through -`Tool.execute()`, while real-time model responses remain application-owned -streams under an Agent scope and send the actual provider call through Relay's -managed streaming LLM API. This preserves low-latency TTS and gives configured -LLM request, streaming-execution, sanitization, and observability middleware one -authoritative execution path. +and intercept pipeline around an async generator. `LiveVisionTool` therefore +returns one complete observation through `Tool.execute()` for normal agentic +planning. Its shared `LiveVisionResponder` is reserved for direct voice, where +an application-owned stream under an Agent scope sends the provider call +through Relay's managed streaming LLM API. This keeps streaming out of ordinary +tool flows without delaying direct speech. Live camera frames remain provider input but are replaced in emitted Relay events by a scope-local sanitizer. Relay request-intercept headers cross the diff --git a/docs/nemo-agent-toolkit-migration.md b/docs/nemo-agent-toolkit-migration.md index 409de963c..802f5e14a 100644 --- a/docs/nemo-agent-toolkit-migration.md +++ b/docs/nemo-agent-toolkit-migration.md @@ -53,8 +53,9 @@ acceptance behavior rather than an implementation dependency. Toolkit by default, add Relay-managed tools, the generic `AgentRunner` seam, and a bounded default tool loop, and retain existing function groups behind legacy extras. -2. **Simple VLM tool** — move the single-turn streaming-vision path to a normal - native tool and prove the lightweight voice sample selects no legacy extra. +2. **Simple VLM tool** — add a finite current-frame tool for agentic flows, + retain streaming only for its direct-voice responder, and prove the + lightweight sample selects no legacy extra. 3. **Native event dispatcher** — port tea-making's typed participant-scoped subscriptions and periodic background sources so voice and autonomous work invoke the same registered tools. diff --git a/docs/source/components/agent-sdk.md b/docs/source/components/agent-sdk.md index 0e70b4003..fad378cd0 100644 --- a/docs/source/components/agent-sdk.md +++ b/docs/source/components/agent-sdk.md @@ -221,14 +221,13 @@ Applications use `as_agent_tool(...)` to expose any `AgentRunner` as a registered tool; foreground selection, workflow state, and background work stay explicit in application code. -`xr_ai_nat.live_vision.LiveVisionResponder` acquires a participant-scoped hub -frame and calls its injected `VLMService` through Relay's managed streaming LLM -path. Each response runs under an Agent scope, and a scope-local sanitizer -replaces the inline camera frame in Relay events without changing provider -input. Relay's managed tool API accepts completed JSON values, so finite tools -use `Tool.execute()` while real-time model output remains an application-owned -stream. This keeps Relay middleware on a supported managed boundary rather than -reimplementing a partial streaming-tool lifecycle. +`xr_ai_nat.live_vision.LiveVisionTool` acquires a participant-scoped hub frame +and returns one complete observation through Relay's finite tool and LLM +boundaries. Direct voice can wrap the same tool with `LiveVisionResponder`, +whose provider response uses Relay's managed streaming LLM path under an Agent +scope. Both paths replace the inline camera frame in Relay events without +changing provider input. The split keeps token streaming out of ordinary +agentic tool calls without reimplementing a partial streaming-tool lifecycle. ## xr-ai-nat model bridge diff --git a/tests/test_simple_vlm_example_worker.py b/tests/test_simple_vlm_example_worker.py index 13750bb22..4f09bd623 100644 --- a/tests/test_simple_vlm_example_worker.py +++ b/tests/test_simple_vlm_example_worker.py @@ -16,7 +16,7 @@ import tomllib import yaml from xr_ai_hub import FrameData, FrameSignal, PixelFormat, ProcessorEndpoint -from xr_ai_models import VLMService +from xr_ai_models import ChatResponse, VLMService from xr_ai_voice import VoiceQuery, VoiceSession from xr_ai_voicegate import VoiceGateConfig @@ -28,7 +28,12 @@ from simple_vlm_example_worker import __main__ as worker_main # noqa: E402 # pyright: ignore[reportMissingImports] from simple_vlm_example_worker import app # noqa: E402 # pyright: ignore[reportMissingImports] from simple_vlm_example_worker.config import load_config # noqa: E402 # pyright: ignore[reportMissingImports] -from xr_ai_nat.live_vision import LiveVisionResponder # noqa: E402 +from xr_ai_nat.live_vision import ( # noqa: E402 + LiveVisionResponder, + LiveVisionTool, + VisionRequest, + VisionResponse, +) class _Service: @@ -53,8 +58,8 @@ def shutdown(self) -> None: self.shutdown_calls += 1 -class _LiveVisionResponder: - instances: list["_LiveVisionResponder"] = [] +class _LiveVisionTool: + instances: list["_LiveVisionTool"] = [] def __init__(self, **kwargs) -> None: self.kwargs = kwargs @@ -62,14 +67,19 @@ def __init__(self, **kwargs) -> None: self.released: list[str] = [] self.instances.append(self) + def release(self, participant_id: str) -> None: + self.released.append(participant_id) + + +class _LiveVisionResponder: + def __init__(self, tool: _LiveVisionTool) -> None: + self.tool = tool + async def stream(self, request): - self.requests.append(request) + self.tool.requests.append(request) for text in ("a ", "blue square"): yield SimpleNamespace(text=text) - def release(self, participant_id: str) -> None: - self.released.append(participant_id) - class _LiveEndpoint: def __init__(self) -> None: @@ -101,6 +111,18 @@ async def set_status(self, status: str, participant_id: str) -> None: class _StreamingVlm: def __init__(self) -> None: self.calls = [] + self.ask_calls = [] + + async def ask_image( + self, + image, + question: str, + *, + system_prompt: str = "", + headers=None, + ) -> ChatResponse: + self.ask_calls.append((image, question, system_prompt, dict(headers or {}))) + return ChatResponse("a blue square", None, None, "stop", {}) async def stream( self, @@ -289,6 +311,7 @@ async def test_app_wires_text_voice_cleanup_readiness_and_shutdown( monkeypatch.setattr(app, "make_stt", lambda _models, _name: stt) monkeypatch.setattr(app, "make_vlm", lambda _models, _name: vlm) monkeypatch.setattr(app, "make_tts", lambda _models, _name: tts) + monkeypatch.setattr(app, "LiveVisionTool", _LiveVisionTool) monkeypatch.setattr(app, "LiveVisionResponder", _LiveVisionResponder) def make_session(**kwargs): @@ -319,7 +342,7 @@ def __init__(self, **kwargs) -> None: monkeypatch.setattr(app, "VoiceSession", make_session) monkeypatch.setattr(app, "TextMessageInput", CaptureTextInput) - _LiveVisionResponder.instances.clear() + _LiveVisionTool.instances.clear() await app.run_app(config, ready_file=ready_file) @@ -328,17 +351,17 @@ def __init__(self, **kwargs) -> None: assert stt.close_calls == tts.close_calls == vlm.close_calls == 1 assert transport.shutdown_calls == 1 assert sessions[0].text_topic == "vlm.response" - assert _LiveVisionResponder.instances[0].kwargs["endpoint"] is transport.endpoint - assert _LiveVisionResponder.instances[0].kwargs["system_prompt"] == config.system_prompt - assert _LiveVisionResponder.instances[0].kwargs["frame_max_age_s"] == ( + assert _LiveVisionTool.instances[0].kwargs["endpoint"] is transport.endpoint + assert _LiveVisionTool.instances[0].kwargs["system_prompt"] == config.system_prompt + assert _LiveVisionTool.instances[0].kwargs["frame_max_age_s"] == ( config.frame_max_age_s ) - assert _LiveVisionResponder.instances[0].kwargs["frame_timeout_s"] == ( + assert _LiveVisionTool.instances[0].kwargs["frame_timeout_s"] == ( config.frame_timeout_s ) - assert _LiveVisionResponder.instances[0].released == ["alice"] - assert _LiveVisionResponder.instances[0].requests[0].participant_id == "alice" - assert _LiveVisionResponder.instances[0].requests[0].query == "What is in front of me?" + assert _LiveVisionTool.instances[0].released == ["alice"] + assert _LiveVisionTool.instances[0].requests[0].participant_id == "alice" + assert _LiveVisionTool.instances[0].requests[0].query == "What is in front of me?" assert streamed == ["a ", "blue square"] assert run_options["interrupt_on_supersede"] is True assert text_inputs[0]["session"] is sessions[0] @@ -347,16 +370,79 @@ def __init__(self, **kwargs) -> None: assert text_inputs[0]["transform"]("What is this?") == "What is this?" +async def test_live_vision_tool_returns_a_complete_agent_observation() -> None: + endpoint = _LiveEndpoint() + vlm = _StreamingVlm() + vision = LiveVisionTool( + endpoint=cast(ProcessorEndpoint, endpoint), + vlm=cast(VLMService, vlm), + system_prompt="Answer briefly.", + ) + assert endpoint.frame_callback is not None + await endpoint.frame_callback( + FrameSignal( + slot=0, + seq=1, + pts_us=time.time_ns() // 1_000, + width=2, + height=2, + fmt=PixelFormat.RGB24, + data_sz=12, + participant_id="alice", + track_id="camera", + ) + ) + events = [] + subscriber = "simple-vlm-finite-vision" + intercept = "simple-vlm-finite-vision-header" + + def add_header(_name, request, annotated): + headers = dict(request.headers) + headers["X-Relay-Session"] = "turn-8" + return nemo_relay.LLMRequestInterceptOutcome( + nemo_relay.LLMRequest(headers, request.content), + annotated, + ) + + nemo_relay.subscribers.register(subscriber, events.append) + nemo_relay.intercepts.register_llm_request(intercept, 0, False, add_header) + try: + result = await vision.execute( + VisionRequest(participant_id="alice", query="What is shown?"), + ) + await nemo_relay.subscribers.flush_async() + finally: + nemo_relay.intercepts.deregister_llm_request(intercept) + nemo_relay.subscribers.deregister(subscriber) + + assert result == VisionResponse(text="a blue square") + image, question, system_prompt, headers = vlm.ask_calls[0] + assert image.startswith("data:image/jpeg;base64,") + assert question == "What is shown?" + assert system_prompt == "Answer briefly." + assert headers["X-Relay-Session"] == "turn-8" + assert endpoint.statuses == [("processing", "alice"), ("idle", "alice")] + assert {"tool", "llm"} <= {getattr(event, "category", None) for event in events} + llm_events = [ + event.to_json() + for event in events + if getattr(event, "category", None) == "llm" + ] + assert llm_events + assert all(image not in event for event in llm_events) + assert any("" in event for event in llm_events) + + async def test_sample_handler_streams_a_live_frame_question() -> None: endpoint = _LiveEndpoint() vlm = _StreamingVlm() - vision = LiveVisionResponder( + vision = LiveVisionTool( endpoint=cast(ProcessorEndpoint, endpoint), vlm=cast(VLMService, vlm), system_prompt="Answer briefly.", ) - handler = app._make_vision_handler(vision) + handler = app._make_vision_handler(LiveVisionResponder(vision)) assert endpoint.frame_callback is not None await endpoint.frame_callback( FrameSignal( @@ -394,6 +480,7 @@ def add_header(_name, request, annotated): timestamp_us=123, ) ) + assert not isinstance(response, str) tokens = [token async for token in response] await nemo_relay.subscribers.flush_async() finally: