diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md index adc26350..4b1b65ed 100644 --- a/DEPENDENCIES.md +++ b/DEPENDENCIES.md @@ -554,16 +554,21 @@ 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 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 +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 98f7ad11..0e6a3ede 100644 --- a/agent-samples/simple-vlm-example/README.md +++ b/agent-samples/simple-vlm-example/README.md @@ -18,11 +18,13 @@ 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 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. 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 ed825e72..95d78d2c 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 ce134aa2..f2b2517a 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 LiveVisionResponder, 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: LiveVisionResponder) -> 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,15 @@ 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) + voice = LiveVisionResponder(vision) TextMessageInput( session=session, transform=_text_transform(config.default_prompt), @@ -87,8 +85,8 @@ async def run_app( logger.info("simple-vlm-example starting") await session.run( - _make_vision_handler(vision), - on_participant_left=vision_config.release, + _make_vision_handler(voice), + on_participant_left=vision.release, interrupt_on_supersede=True, ) logger.info("simple-vlm-example stopped") diff --git a/agent-sdk/xr-ai-models/README.md b/agent-sdk/xr-ai-models/README.md index 9b634f28..b1adfe9f 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 d066308a..d137eff0 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 05ac730c..b1c93469 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 cf882279..9b899204 100644 --- a/agent-sdk/xr-ai-nat/README.md +++ b/agent-sdk/xr-ai-nat/README.md @@ -60,6 +60,22 @@ 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 and direct voice responder + +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. 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 ## 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 dfcdb78c..7938b786 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py @@ -6,4 +6,10 @@ from .agent_runner import AgentRunner, as_agent_tool from .tools import Tool, ToolInvocationResult, ToolSet -__all__ = ["AgentRunner", "Tool", "ToolInvocationResult", "ToolSet", "as_agent_tool"] +__all__ = [ + "AgentRunner", + "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 00000000..05ac34c4 --- /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/_relay.py b/agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py new file mode 100644 index 00000000..a01333bb --- /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 c98b8cf6..82331ccd 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 new file mode 100644 index 00000000..002b9ca5 --- /dev/null +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py @@ -0,0 +1,342 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Current-frame vision for agent tools and direct voice responses.""" + +from __future__ import annotations + +import asyncio +import copy +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 ._relay import headers_from_relay +from .tools import Tool + +_LOGGER = logging.getLogger(__name__) +_VLM_CALL_NAME = "xr-ai-vlm" +_FRAME_REDACTION = "" + + +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 VisionResponse(BaseModel): + """A complete answer about the current frame.""" + + text: str = Field(description="Complete answer text.") + + +class VisionChunk(BaseModel): + """One text fragment from a direct voice answer.""" + + text: str = Field(description="A partial fragment of the answer text.") + + +class LiveVisionTool(Tool[VisionRequest, VisionResponse]): + """A finite current-frame tool for agentic planning and tool loops.""" + + 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, + 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) + 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: + _register_frame_sanitizer() + 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 _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, + ) -> 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, + headers=headers_from_relay(request.headers), + ): + 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, +) -> 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): + 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 _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") + 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__ = [ + "LiveVisionResponder", + "LiveVisionTool", + "VisionChunk", + "VisionRequest", + "VisionResponse", +] diff --git a/docs/changelog.md b/docs/changelog.md index b7230b29..04059bf4 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 — 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. `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 +`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/nemo-agent-toolkit-migration.md b/docs/nemo-agent-toolkit-migration.md index 409de963..802f5e14 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 bd480079..fad378cd 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,6 +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. +`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 Unmigrated workflows install `xr-ai-nat[agents]` when they use NAT's built-in diff --git a/tests/test_models_openai_compat.py b/tests/test_models_openai_compat.py index 7cc50424..8e55f1a7 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_simple_vlm_example_worker.py b/tests/test_simple_vlm_example_worker.py index 6440e0b2..4f09bd62 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 ChatResponse, VLMService from xr_ai_voice import VoiceQuery, VoiceSession from xr_ai_voicegate import VoiceGateConfig @@ -24,9 +25,15 @@ _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 ( # noqa: E402 + LiveVisionResponder, + LiveVisionTool, + VisionRequest, + VisionResponse, +) class _Service: @@ -51,21 +58,12 @@ 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) @@ -73,20 +71,14 @@ 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 +class _LiveVisionResponder: + def __init__(self, tool: _LiveVisionTool) -> None: + self.tool = tool - async def add_function(self, name: str, config: object): - self.added.append((name, config)) - return self.function + async def stream(self, request): + self.tool.requests.append(request) + for text in ("a ", "blue square"): + yield SimpleNamespace(text=text) class _LiveEndpoint: @@ -119,9 +111,28 @@ async def set_status(self, status: str, participant_id: str) -> None: 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)) + 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, + 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 @@ -138,7 +149,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 +300,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 +311,8 @@ 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) + monkeypatch.setattr(app, "LiveVisionResponder", _LiveVisionResponder) def make_session(**kwargs): session = VoiceSession(transport=transport, **kwargs) # type: ignore[arg-type] @@ -332,7 +342,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 +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 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] @@ -361,32 +370,108 @@ 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_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(LiveVisionResponder(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" + 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( participant_id="alice", @@ -395,11 +480,26 @@ async def test_sample_handler_streams_a_live_frame_question() -> None: timestamp_us=123, ) ) + assert not isinstance(response, str) 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 {"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)