diff --git a/AGENTS.md b/AGENTS.md index df901d05c..6a209170e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,12 +12,13 @@ historical decisions in `docs/changelog.md`. ``` client-samples/ # Platform clients (Android, iOS/visionOS, Web) -agent-sdk/ # Five packages: +agent-sdk/ # Six packages: # xr-ai-hub-client — IPC client library (pyzmq + msgpack only) # xr-ai-models — LLM/VLM/STT/TTS service protocols + OpenAI-compat clients # xr-ai-pipecat — optional Pipecat transport bridge (heavier deps) # xr-ai-voice — voice runtime (VoiceSession); introduced alongside xr-ai-pipecat - # xr-ai-nat — native Relay-managed tools; legacy NAT compatibility during migration + # xr-ai-tools — toolkit-independent Relay-managed tools + # xr-ai-nat — legacy NeMo Agent Toolkit compatibility during migration utils/ # Shared infra: launcher, logging, vad, vllm, voicegate services/ # XR hub, CloudXR, model-serving, and typed capability services agent-mcp-servers/ # MCP adapters: oxr, render, transcript, vec, video, vlm @@ -58,12 +59,12 @@ deps/ # Gitignored downloaded binaries (e.g. LOVR AppImage) `anthropic`, no `litellm`); all in-tree backends speak OpenAI-compatible HTTP. - **Workers never import from `xr_media_hub` or `xr_ai_launcher`.** Use the - public `xr_ai_hub`, `xr_ai_models`, `xr_ai_nat`, and `xr_ai_voice` SDK - surfaces plus task-specific libraries (numpy, torch, …). + public `xr_ai_hub`, `xr_ai_models`, `xr_ai_tools`, `xr_ai_nat`, and + `xr_ai_voice` SDK surfaces plus task-specific libraries (numpy, torch, …). - **Agentic functions are native and in-process.** New and migrated tools live - in `xr-ai-nat`; every tool and tool-driven agent lifecycle passes through - NeMo Relay, and all model I/O remains in `xr-ai-models`. Its legacy extras - retain NeMo Agent Toolkit compatibility only while existing function groups + in `xr-ai-tools`; every tool execution passes through NeMo Relay, and all + model I/O remains in `xr-ai-models`. `xr-ai-nat` retains + NeMo Agent Toolkit compatibility only while existing function groups migrate. Existing MCP servers remain compatibility surfaces while their capabilities migrate. - **RAG is a native typed capability.** `rag-service` owns document chunking, @@ -172,9 +173,8 @@ itself. `XRTrackingFunctionsConfig` exposes the current user frame through the typed OpenXR service without routing native agents through MCP. `VideoMemoryFunctionsConfig` exposes recorded-video discovery, queries, and frame extraction through a typed service while keeping MCP optional; callers -obtain current frames through the hub client. `StreamingVisionConfig` composes -raw frame acquisition with VLM streaming behind one native function for voice -workflows. `ModelsLLMConfig` adapts the `xr-ai-models` service boundary to +obtain current frames through the hub client. `StreamingVisionTool` in `xr-ai-tools` composes raw frame +acquisition with VLM streaming and stays independent of voice. `ModelsLLMConfig` adapts the `xr-ai-models` service boundary to NAT's built-in LangChain-backed agent types; applications install `xr-ai-nat[agents]` rather than calling LangChain model clients directly. diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md index 4b1b65ed0..eb500df04 100644 --- a/DEPENDENCIES.md +++ b/DEPENDENCIES.md @@ -119,23 +119,25 @@ xr-ai-models (agent-sdk/xr-ai-models/) own httpx wrappers. Profiles may separate adapter, endpoint, and deployment metadata while the existing flat YAML schema remains valid. -xr-ai-nat (agent-sdk/xr-ai-nat/) +xr-ai-tools (agent-sdk/xr-ai-tools/) └── nemo-relay >=0.7.2,<0.8 └── pydantic >=2.10 └── [relay] xr-ai-models [editable: ../xr-ai-models] └── [live-vision] numpy >=1.24, Pillow >=10.0, xr-ai-hub-client [editable: ../xr-ai-hub-client], xr-ai-models [editable: ../xr-ai-models] - └── [agents] nvidia-nat-core ==1.8.0, nvidia-nat-langchain ==1.8.0, xr-ai-models [editable: ../xr-ai-models] - └── [mcp] nvidia-nat-core ==1.8.0, fastmcp >=3.4,<4 - └── [services] nvidia-nat-core ==1.8.0, msgpack >=1.0, pyzmq >=27.0 - └── [vision] nvidia-nat-core ==1.8.0, httpx >=0.27, numpy >=1.24, Pillow >=10.0, xr-ai-hub-client [editable: ../xr-ai-hub-client], xr-ai-models [editable: ../xr-ai-models] - └── [voice] nvidia-nat-core ==1.8.0, xr-ai-voice [editable: ../xr-ai-voice] - The base package is the toolkit-independent native tools layer: Pydantic - request and response models, Relay-managed execution, the generic - ``AgentRunner`` protocol, and a bounded default tool loop over - `xr-ai-models`. The ``[relay]`` and - ``[live-vision]`` extras add model-backed tools without selecting NeMo Agent - Toolkit. The existing function groups remain behind legacy compatibility - extras while they migrate. The ``xr_spatial_math`` function group accepts + Toolkit-independent native tools: Pydantic request and response models, + Relay-managed finite and async execution, model tool-call workflow helpers, + and participant-scoped live vision. + +xr-ai-nat (agent-sdk/xr-ai-nat/) + └── nvidia-nat-core ==1.8.0 + └── pydantic >=2.10 + └── [agents] nvidia-nat-langchain ==1.8.0, xr-ai-models [editable: ../xr-ai-models] + └── [mcp] fastmcp >=3.4,<4 + └── [services] msgpack >=1.0, pyzmq >=27.0 + └── [vision] httpx >=0.27, numpy >=1.24, Pillow >=10.0, xr-ai-hub-client [editable: ../xr-ai-hub-client], xr-ai-models [editable: ../xr-ai-models] + └── [voice] xr-ai-voice [editable: ../xr-ai-voice] + Typed, in-process NeMo Agent Toolkit functions retained while their + concrete capabilities migrate. The ``xr_spatial_math`` function group accepts explicit coordinate frames and performs deterministic spatial calculations without OpenXR, model, or MCP dependencies. ``xr_text_memory`` owns persistent per-source JSONL text @@ -153,9 +155,8 @@ xr-ai-nat (agent-sdk/xr-ai-nat/) ``xr_vision_tools`` exposes ``look_at_current_frame`` and ``look_at_past_frame`` over the always-on live-frame source, acquiring the frame itself and calling an injected xr-ai-models VLM; recorded lookups - resolve through the ``xr_video_memory`` group. A separate - ``xr_streaming_vision`` function composes current-frame acquisition with - complete or streaming VLM invocation. ``xr_tracking`` calls + resolve through the ``xr_video_memory`` group. The replaced NAT + streaming function has moved to ``xr-ai-tools``. ``xr_tracking`` calls the typed OpenXR service and returns a complete user coordinate frame. ``xr_video_memory`` calls the typed video-memory service for recorded-video discovery, queries, and frame extraction. ``xr_rag`` calls the typed RAG @@ -334,7 +335,8 @@ vec-mcp-server (agent-mcp-servers/vec-mcp/) xr-ai-tests (tests/) └── xr-ai-hub-client [editable: ../agent-sdk/xr-ai-hub-client] └── xr-ai-models [editable: ../agent-sdk/xr-ai-models] - └── xr-ai-nat[agents,relay,services,vision] [editable: ../agent-sdk/xr-ai-nat] + └── xr-ai-nat[agents,services,vision] [editable: ../agent-sdk/xr-ai-nat] + └── xr-ai-tools[live-vision] [editable: ../agent-sdk/xr-ai-tools] └── xr-rag-service [editable: ../services/rag-service] └── xr-ai-pipecat [editable: ../agent-sdk/xr-ai-pipecat] └── xr-ai-voice [editable: ../agent-sdk/xr-ai-voice] @@ -554,14 +556,13 @@ 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[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. +| Worker | `simple-vlm-example-worker` | `xr-ai-hub-client [editable]`, `xr-ai-logging [editable]`, `xr-ai-models [editable]`, `xr-ai-tools[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-tools[live-vision]` pulls in numpy and Pillow) | + +The packaged worker constructs a transport-independent `StreamingVisionTool` +and adapts its typed async chunks to `VoiceSession` locally. The tool owns +current-frame acquisition through `xr-ai-hub-client`, has no voice dependency, +and uses NeMo Relay's managed streaming LLM path. 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/README.md b/README.md index 288fae02e..7c1b8838f 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,8 @@ frames are dropped if it is closed. | Hub service | `services/xr-media-hub/` | XR-Media-Hub + LiveKit internal transport | | Launcher | `utils/xr-ai-launcher/` | stdlib-only process manager used by samples | | Logging | `utils/xr-ai-logging/` | shared loguru sink + stdlib bridge for every process | -| Agent tools | `agent-sdk/xr-ai-nat/` | Relay-managed native tools and legacy NAT compatibility during migration | +| Agent tools | `agent-sdk/xr-ai-tools/` | Toolkit-independent Relay-managed native tools | +| Legacy NAT | `agent-sdk/xr-ai-nat/` | NeMo Agent Toolkit compatibility during migration | | Reusable services | `services/` | Model-serving and typed capability processes | | Agent interfaces | `agent-mcp-servers/` | MCP compatibility processes for XR data & rendering | | Agent demos | `agent-samples/` | End-to-end agent pipelines | @@ -195,7 +196,7 @@ channel, or send the literal text `"ping"` — all routes go through the same VLM pipeline against the latest video frame. Replies arrive as streaming Piper TTS audio plus a `vlm.response` text message. -The packaged worker composes the NAT-native streaming vision function with +The packaged worker adapts the transport-independent `StreamingVisionTool` to `xr-ai-voice`'s `VoiceSession`; Pipecat remains private to that runtime and no MCP client is involved. See the [sample README](agent-samples/simple-vlm-example/README.md) for the worker diff --git a/agent-samples/simple-vlm-example/README.md b/agent-samples/simple-vlm-example/README.md index 0e6a3ede5..c4606b851 100644 --- a/agent-samples/simple-vlm-example/README.md +++ b/agent-samples/simple-vlm-example/README.md @@ -19,9 +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 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 +transport-independent `StreamingVisionTool` and adapts its async chunks to the +voice handler locally. The tool has no voice dependency and sends its provider +stream 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/pyproject.toml b/agent-samples/simple-vlm-example/worker/pyproject.toml index 95d78d2c6..dc4900542 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[relay,live-vision]", + "xr-ai-tools[live-vision]", "xr-ai-voice", "xr-ai-voicegate", "loguru>=0.7", @@ -24,7 +24,7 @@ dependencies = [ xr-ai-hub-client = { path = "../../../agent-sdk/xr-ai-hub-client", editable = true } xr-ai-logging = { path = "../../../utils/xr-ai-logging", editable = true } xr-ai-models = { path = "../../../agent-sdk/xr-ai-models", editable = true } -xr-ai-nat = { path = "../../../agent-sdk/xr-ai-nat", editable = true } +xr-ai-tools = { path = "../../../agent-sdk/xr-ai-tools", editable = true } xr-ai-voice = { path = "../../../agent-sdk/xr-ai-voice", editable = true } xr-ai-voicegate = { path = "../../../utils/xr-ai-voicegate", editable = true } 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 f2b2517a5..d41e6a585 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 LiveVisionResponder, LiveVisionTool, VisionRequest +from xr_ai_tools.streaming_vision import StreamingVisionTool, 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: LiveVisionResponder) -> VoiceHandler: +def _make_vision_handler(vision: StreamingVisionTool) -> VoiceHandler: async def handle(turn): async def response(): async for chunk in vision.stream( @@ -69,14 +69,13 @@ async def run_app( ) async with session: - vision = LiveVisionTool( + vision = StreamingVisionTool( 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), @@ -85,7 +84,7 @@ async def run_app( logger.info("simple-vlm-example starting") await session.run( - _make_vision_handler(voice), + _make_vision_handler(vision), on_participant_left=vision.release, interrupt_on_supersede=True, ) diff --git a/agent-sdk/xr-ai-hub-client/README.md b/agent-sdk/xr-ai-hub-client/README.md index a6d0b820c..30e28b4e3 100644 --- a/agent-sdk/xr-ai-hub-client/README.md +++ b/agent-sdk/xr-ai-hub-client/README.md @@ -25,8 +25,8 @@ await endpoint.run() ``` `LiveFrameSource` adds raw frame acquisition without adding image conversion or -model dependencies. Use `xr_ai_nat.functions.vision.StreamingVisionConfig` when the desired -interface is a model-facing vision function rather than raw pixels. +model dependencies. Use `xr_ai_tools.streaming_vision.StreamingVisionTool` when the +desired interface is a model-facing async vision tool rather than raw pixels. ## Migrating from `xr-ai-agent` diff --git a/agent-sdk/xr-ai-nat/README.md b/agent-sdk/xr-ai-nat/README.md index 9b8992047..538b4719d 100644 --- a/agent-sdk/xr-ai-nat/README.md +++ b/agent-sdk/xr-ai-nat/README.md @@ -3,80 +3,11 @@ SPDX-License-Identifier: Apache-2.0 --> -# XR AI native tools +# XR AI functions for NeMo Agent Toolkit -`xr-ai-nat` is the toolkit-independent native tools layer for XR AI. -`Tool` gives voice, background triggers, and model-driven agents one typed -Pydantic invocation interface. NeMo Relay manages every new tool execution; -model-backed tools use injected `xr-ai-models` services rather than exposing a -model client to an application trigger. - -The existing NeMo Agent Toolkit function groups remain available through legacy -extras while their concrete capabilities migrate. They are compatibility code, -not the destination for new tools. - -## Native tools and tool-driven agents - -The base install supplies `Tool`, `AgentRunner`, and `as_agent_tool`. Install -`xr-ai-nat[relay]` for the bundled bounded tool-driven `Agent`: - -```python -from pydantic import BaseModel -from xr_ai_nat import Tool -from xr_ai_nat.agents import Agent - - -class LookupRequest(BaseModel): - query: str - - -class LookupResult(BaseModel): - answer: str - - -async def lookup(request: LookupRequest) -> LookupResult: - return LookupResult(answer=request.query) - - -lookup_tool = Tool( - "lookup", - "Look up one answer.", - LookupRequest, - LookupResult, - lookup, -) -agent = Agent( - name="assistant", - llm=llm, - system_prompt="Use the available tools.", - tools=(lookup_tool,), -) -``` - -`AgentRunner` is the small async turn protocol behind `as_agent_tool(...)`. -The bundled `Agent` is the basic stateless tool loop; applications can expose a -custom, Fabric-backed, or framework-backed runner through the same registered -`Tool`. That keeps voice, text, and autonomous background work on one -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 +`xr-ai-nat` provides typed, in-process XR functions for NVIDIA NeMo Agent +Toolkit (NAT). Applications compose these functions directly; process-backed +or MCP compatibility adapters remain separate boundaries. ## Shared value models and the service boundary @@ -238,15 +169,6 @@ The `video_memory` reference is resolved lazily — only on the first `look_at_past_frame` call. A **live-only** consumer may omit `video_memory` (and need not register that group) as long as it never calls `look_at_past_frame`. -For live voice workflows, `StreamingVisionConfig` (`xr_streaming_vision`) accepts -a hub `ProcessorEndpoint` and exposes one native function with complete and -streaming invocation modes. It owns fresh-frame acquisition and VLM invocation; -Pipecat continues to own audio framing, interruption, and TTS. - -Its complete invocation returns a `VisionResult` with `status` set to `ok` or -`unavailable`; callers must handle an unavailable result without treating its -text as an answer about the scene. - MCP-only agents that already hold a local image path can still reach the legacy file-path `ask_image` tool through the vlm-mcp compatibility server (`agent-mcp-servers/vlm-mcp/`), which now owns that path-based surface directly. diff --git a/agent-sdk/xr-ai-nat/pyproject.toml b/agent-sdk/xr-ai-nat/pyproject.toml index 8c3545982..ea9575182 100644 --- a/agent-sdk/xr-ai-nat/pyproject.toml +++ b/agent-sdk/xr-ai-nat/pyproject.toml @@ -8,21 +8,19 @@ build-backend = "hatchling.build" [project] name = "xr-ai-nat" version = "0.1.0" -description = "Native Relay-managed tools and legacy NeMo Agent Toolkit compatibility for XR AI." +description = "NVIDIA NeMo Agent Toolkit functions for XR AI." requires-python = ">=3.11,<3.13" dependencies = [ - "nemo-relay>=0.7.2,<0.8", + "nvidia-nat-core==1.8.0", "pydantic>=2.10", ] [project.optional-dependencies] -relay = ["xr-ai-models"] -live-vision = ["numpy>=1.24", "Pillow>=10.0", "xr-ai-hub-client", "xr-ai-models"] -agents = ["nvidia-nat-core==1.8.0", "nvidia-nat-langchain==1.8.0", "xr-ai-models"] -mcp = ["nvidia-nat-core==1.8.0", "fastmcp>=3.4,<4"] -services = ["nvidia-nat-core==1.8.0", "msgpack>=1.0", "pyzmq>=27.0"] -vision = ["nvidia-nat-core==1.8.0", "httpx>=0.27", "numpy>=1.24", "Pillow>=10.0", "xr-ai-hub-client", "xr-ai-models"] -voice = ["nvidia-nat-core==1.8.0", "xr-ai-voice"] +agents = ["nvidia-nat-langchain==1.8.0", "xr-ai-models"] +mcp = ["fastmcp>=3.4,<4"] +services = ["msgpack>=1.0", "pyzmq>=27.0"] +vision = ["httpx>=0.27", "numpy>=1.24", "Pillow>=10.0", "xr-ai-hub-client", "xr-ai-models"] +voice = ["xr-ai-voice"] [project.entry-points."nat.plugins"] xr_ai_nat_llm = "xr_ai_nat.llm.config" 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 7938b7866..282ee8956 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/__init__.py @@ -1,15 +1,6 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Toolkit-independent native XR tools with legacy NAT compatibility.""" +"""NAT-native XR functions.""" -from .agent_runner import AgentRunner, as_agent_tool -from .tools import Tool, ToolInvocationResult, ToolSet - -__all__ = [ - "AgentRunner", - "Tool", - "ToolInvocationResult", - "ToolSet", - "as_agent_tool", -] +__all__: list[str] = [] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/agent_runner.py b/agent-sdk/xr-ai-nat/xr_ai_nat/agent_runner.py deleted file mode 100644 index 6c65dc0d0..000000000 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/agent_runner.py +++ /dev/null @@ -1,55 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -"""Framework-neutral agent runners exposed through native tools.""" - -from __future__ import annotations - -from collections.abc import Callable -from typing import Protocol, TypeVar - -from pydantic import BaseModel - -from .tools import Tool - -RunnerRequestT = TypeVar("RunnerRequestT", contravariant=True) -RunnerResultT = TypeVar("RunnerResultT", covariant=True) -ToolRequestT = TypeVar("ToolRequestT", bound=BaseModel) -ToolResultT = TypeVar("ToolResultT", bound=BaseModel) - - -class AgentRunner(Protocol[RunnerRequestT, RunnerResultT]): - """An application-owned agent implementation that completes one asynchronous turn.""" - - async def run(self, request: RunnerRequestT) -> RunnerResultT: - """Run one turn and return the implementation-specific result.""" - raise NotImplementedError - - -def as_agent_tool( - *, - name: str, - description: str, - agent: AgentRunner[RunnerRequestT, RunnerResultT], - request_model: type[ToolRequestT], - result_model: type[ToolResultT], - request: Callable[[ToolRequestT], RunnerRequestT], - response: Callable[[RunnerResultT], ToolResultT], - return_direct: bool = False, -) -> Tool[ToolRequestT, ToolResultT]: - """Expose any ``AgentRunner`` through the same ``Tool`` interface as capabilities.""" - - async def invoke(value: ToolRequestT) -> ToolResultT: - return response(await agent.run(request(value))) - - return Tool( - name, - description, - request_model, - result_model, - invoke, - return_direct=return_direct, - ) - - -__all__ = ["AgentRunner", "as_agent_tool"] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py b/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py deleted file mode 100644 index 82331ccd7..000000000 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/agents.py +++ /dev/null @@ -1,306 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -"""Tool-driven agents whose model calls remain internal to the tools layer.""" - -from __future__ import annotations - -import json -from collections.abc import Sequence -from dataclasses import dataclass -from typing import Any - -import nemo_relay -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 - - -class ToolLoopLimitError(RuntimeError): - """Raised when a model has not produced a final answer within the configured budget.""" - - -@dataclass(frozen=True, slots=True) -class AgentResult: - """The final text and messages generated during one tool-driven agent turn.""" - - text: str - messages: tuple[ChatMessage, ...] - - -class Agent(AgentRunner[str, AgentResult]): - """Run one small stateless tool-calling turn over a catalog of native tools.""" - - def __init__( - self, - *, - name: str, - llm: LLMService, - system_prompt: str, - tools: Sequence[Tool[Any, Any]], - model_name: str = "xr-ai-model", - max_iterations: int = 4, - max_tokens: int = 1024, - temperature: float = 0.0, - enable_thinking: bool = False, - thinking_budget: int | None = None, - ) -> None: - if not name: - raise ValueError("agent name must not be empty") - if not system_prompt: - raise ValueError(f"agent {name!r} needs a system prompt") - if max_iterations < 1: - raise ValueError("max_iterations must be at least one") - self.name = name - self.llm = llm - self.system_prompt = system_prompt - self.tools = ToolSet(tools) - self.model_name = model_name - self.max_iterations = max_iterations - self.max_tokens = max_tokens - self.temperature = temperature - self.enable_thinking = enable_thinking - self.thinking_budget = thinking_budget - - async def run(self, request: str) -> AgentResult: - """Execute one user request without carrying hidden conversation history.""" - - if not request.strip(): - raise ValueError("agent request must not be blank") - messages = [ - ChatMessage(role="system", content=self.system_prompt), - ChatMessage(role="user", content=request), - ] - with nemo_relay.scope.scope( - self.name, - nemo_relay.ScopeType.Agent, - input={"request": request}, - ): - for _ in range(self.max_iterations): - response = await self._chat(messages) - messages.append( - ChatMessage( - role="assistant", - content=response.content, - tool_calls=response.tool_calls, - ) - ) - if not response.tool_calls: - return AgentResult(response.content.strip(), tuple(messages)) - for call in response.tool_calls: - tool = self.tools.get(call.name) - if tool is None: - outcome = json.dumps({"error": "unknown_tool", "tool": call.name}) - return_direct = False - else: - invocation = await tool.invoke(call.arguments) - outcome = invocation.content - return_direct = invocation.return_direct - messages.append( - ChatMessage( - role="tool", - content=outcome, - tool_call_id=call.id, - ) - ) - if return_direct: - return AgentResult(outcome, tuple(messages)) - raise ToolLoopLimitError( - f"agent {self.name!r} exhausted {self.max_iterations} model iterations", - ) - - async def _chat(self, messages: Sequence[ChatMessage]) -> ChatResponse: - tool_definitions = list(self.tools.definitions) - content: dict[str, Any] = { - "model": self.model_name, - "messages": [_message_to_openai(message) for message in messages], - "max_tokens": self.max_tokens, - "temperature": self.temperature, - "enable_thinking": self.enable_thinking, - "thinking_budget": self.thinking_budget, - } - if tool_definitions: - content["tools"] = tool_definitions - relay_request = nemo_relay.LLMRequest( - {}, - content, - ) - - async def invoke(request: nemo_relay.LLMRequest) -> dict[str, Any]: - content = request.content - response = await self.llm.chat( - _messages_from_openai(content.get("messages")), - tools=_tools_from_openai(content.get("tools")), - max_tokens=_optional_int(content.get("max_tokens")), - 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), - ) - return _response_to_openai(response) - - raw_response = await nemo_relay.llm.execute( - self.name, - relay_request, - invoke, - model_name=self.model_name, - codec=OpenAIChatCodec(), - response_codec=OpenAIChatCodec(), - ) - return _response_from_openai(raw_response) - - -def _message_to_openai(message: ChatMessage) -> dict[str, Any]: - if not isinstance(message.content, str): - raise TypeError("the native agent accepts text-only LLM messages") - content: str | None = message.content - if message.role == "assistant" and not content and message.tool_calls: - content = None - result: dict[str, Any] = {"role": message.role, "content": content} - if message.tool_calls: - result["tool_calls"] = [ - { - "id": call.id, - "type": "function", - "function": {"name": call.name, "arguments": call.arguments}, - } - for call in message.tool_calls - ] - if message.tool_call_id: - result["tool_call_id"] = message.tool_call_id - return result - - -def _messages_from_openai(raw: object) -> list[ChatMessage]: - if not isinstance(raw, list): - raise TypeError("Relay LLM request must contain a message array") - messages: list[ChatMessage] = [] - for item in raw: - if not isinstance(item, dict): - raise TypeError("Relay LLM messages must be objects") - role = item.get("role") - if role not in {"system", "user", "assistant", "tool"}: - raise ValueError(f"unsupported Relay LLM role: {role!r}") - content = item.get("content", "") - tool_calls = _tool_calls_from_openai(item.get("tool_calls")) or None - if content is None and tool_calls: - content = "" - if not isinstance(content, str): - raise TypeError("the native agent accepts text-only LLM messages") - tool_call_id = item.get("tool_call_id") - if tool_call_id is not None and not isinstance(tool_call_id, str): - raise TypeError("tool_call_id must be a string") - messages.append( - ChatMessage( - role=role, - content=content, - tool_calls=tool_calls, - tool_call_id=tool_call_id, - ) - ) - return messages - - -def _tool_calls_from_openai(raw: object) -> list[ToolCall]: - if raw is None: - return [] - if not isinstance(raw, list): - raise TypeError("tool_calls must be an array") - calls: list[ToolCall] = [] - for item in raw: - if not isinstance(item, dict) or not isinstance(item.get("function"), dict): - raise TypeError("tool call must contain a function object") - function = item["function"] - name = function.get("name") - arguments = function.get("arguments") - identifier = item.get("id") - if not isinstance(identifier, str) or not isinstance(name, str) or not isinstance(arguments, str): - raise TypeError("tool call id, name, and arguments must be strings") - calls.append(ToolCall(id=identifier, name=name, arguments=arguments)) - return calls - - -def _tools_from_openai(raw: object) -> list[ToolDef] | None: - if raw is None: - return None - if not isinstance(raw, list): - raise TypeError("tools must be an array") - definitions: list[ToolDef] = [] - for item in raw: - if not isinstance(item, dict): - raise TypeError("tool definition must be an object") - function = item.get("function", item) - if not isinstance(function, dict): - raise TypeError("tool definition function must be an object") - name = function.get("name") - description = function.get("description", "") - parameters = function.get("parameters", {"type": "object"}) - if not isinstance(name, str) or not isinstance(description, str) or not isinstance(parameters, dict): - raise TypeError("tool definition has invalid fields") - definitions.append(ToolDef(name=name, description=description, parameters=parameters)) - return definitions - - -def _response_to_openai(response: ChatResponse) -> dict[str, Any]: - message = _message_to_openai( - ChatMessage(role="assistant", content=response.content, tool_calls=response.tool_calls), - ) - return { - "model": response.raw.get("model", "xr-ai-model") if isinstance(response.raw, dict) else "xr-ai-model", - "choices": [{"message": message, "finish_reason": response.finish_reason}], - "usage": response.raw.get("usage", {}) if isinstance(response.raw, dict) else {}, - "xr_ai_reasoning": response.reasoning, - } - - -def _response_from_openai(raw: object) -> ChatResponse: - if not isinstance(raw, dict): - raise TypeError("Relay LLM response must be an object") - choices = raw.get("choices") - if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict): - raise ValueError("Relay LLM response must contain one choice") - choice = choices[0] - message = choice.get("message") - if not isinstance(message, dict): - raise TypeError("Relay LLM response choice must contain a message") - content = message.get("content", "") - tool_calls = _tool_calls_from_openai(message.get("tool_calls")) or None - if content is None and tool_calls: - content = "" - if not isinstance(content, str): - raise TypeError("Relay LLM response content must be a string") - finish_reason = choice.get("finish_reason") - if finish_reason is not None and not isinstance(finish_reason, str): - raise TypeError("Relay LLM finish_reason must be a string") - reasoning = raw.get("xr_ai_reasoning") - if reasoning is not None and not isinstance(reasoning, str): - raise TypeError("Relay LLM reasoning must be a string") - return ChatResponse( - content=content, - reasoning=reasoning, - tool_calls=tool_calls, - finish_reason=finish_reason, - raw=raw, - ) - - -def _optional_int(value: object) -> int | None: - if value is None: - return None - if isinstance(value, bool) or not isinstance(value, int): - raise TypeError("expected an integer or null") - return value - - -def _optional_float(value: object) -> float | None: - if value is None: - return None - if isinstance(value, bool) or not isinstance(value, (int, float)): - raise TypeError("expected a number or null") - return float(value) - - -__all__ = ["Agent", "AgentResult", "AgentRunner", "ToolLoopLimitError", "as_agent_tool"] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/__init__.py b/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/__init__.py index 6ff7a4c95..f6f2c7363 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/__init__.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/__init__.py @@ -7,10 +7,6 @@ HistoricalVisionRequest, LiveVisionRequest, LiveVisionResult, - StreamingVisionConfig, - VisionChunk, - VisionRequest, - VisionResult, VisionToolsConfig, ) @@ -18,9 +14,5 @@ "HistoricalVisionRequest", "LiveVisionRequest", "LiveVisionResult", - "StreamingVisionConfig", - "VisionChunk", - "VisionRequest", - "VisionResult", "VisionToolsConfig", ] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/functions.py b/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/functions.py index aee48c316..2dc8e1150 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/functions.py +++ b/agent-sdk/xr-ai-nat/xr_ai_nat/functions/vision/functions.py @@ -4,19 +4,14 @@ """Native NAT functions for current and recorded camera frames.""" import asyncio -import logging -from collections.abc import AsyncGenerator from pathlib import Path -from typing import Any, Literal +from typing import Any from nat.plugin_api import ( Builder, - FunctionBaseConfig, FunctionGroup, FunctionGroupBaseConfig, FunctionGroupRef, - FunctionInfo, - register_function, register_function_group, ) from pydantic import BaseModel, ConfigDict, Field, PrivateAttr @@ -24,34 +19,6 @@ from .._models import _StrictRequest -_LOGGER = logging.getLogger(__name__) - - -class VisionRequest(_StrictRequest): - """Request a VLM answer from one participant's current camera frame.""" - - 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 VisionResult(BaseModel): - """Complete answer from a live-camera vision invocation.""" - - text: str = Field(description="Answer text or a reason that vision is unavailable.") - status: Literal["ok", "unavailable"] = Field( - default="ok", - description=( - "Whether a current frame and VLM answer were available. " - "Callers must handle an unavailable result without interpreting its text as a scene answer." - ), - ) - - -class VisionChunk(BaseModel): - """One streamed text fragment from a live-camera vision invocation.""" - - text: str = Field(description="A partial fragment of the streamed answer text.") - class LiveVisionRequest(_StrictRequest): """Ask a question about a participant's present live camera frame.""" @@ -99,89 +66,6 @@ async def _ask_image( return text -class StreamingVisionConfig(FunctionBaseConfig, name="xr_streaming_vision"): - """Configure one native streaming function over a live XR camera.""" - - model_config = ConfigDict(arbitrary_types_allowed=True) - - endpoint: Any = Field(exclude=True, repr=False) - vlm: Any = Field(exclude=True, repr=False) - system_prompt: str = "" - frame_max_age_s: float = Field(default=2.0, gt=0.0) - frame_timeout_s: float = Field(default=5.0, gt=0.0) - _frames: LiveFrameSource | None = PrivateAttr(default=None) - - def release(self, participant_id: str) -> None: - """Forget cached frame state after a participant disconnects.""" - - if self._frames is not None: - self._frames.release(participant_id) - - -@register_function(config_type=StreamingVisionConfig) -async def streaming_vision(config: StreamingVisionConfig, _builder: Builder): - frames = LiveFrameSource( - config.endpoint, - max_age_s=config.frame_max_age_s, - timeout_s=config.frame_timeout_s, - ) - config._frames = frames - - async def answer(request: VisionRequest) -> VisionResult: - await config.endpoint.set_status("processing", request.participant_id) - status: Literal["ok", "unavailable"] = "ok" - try: - image_url = await _current_image(frames, request.participant_id) - response = await config.vlm.ask_image( - image_url, - request.query, - system_prompt=config.system_prompt, - ) - text = (response.content or "").strip() - if not text: - text = "I couldn't make out anything in the view." - except FrameUnavailable as exc: - text = str(exc) - status = "unavailable" - except Exception: - _LOGGER.exception("Live VLM request failed") - text = "VLM server unavailable — please retry." - status = "unavailable" - finally: - await config.endpoint.set_status("idle", request.participant_id) - return VisionResult(text=text, status=status) - - async def stream(request: VisionRequest) -> AsyncGenerator[VisionChunk, None]: - try: - image_url = await _current_image(frames, request.participant_id) - except FrameUnavailable as exc: - yield VisionChunk(text=str(exc)) - return - - await config.endpoint.set_status("processing", request.participant_id) - try: - async for token in config.vlm.stream( - image_url, - request.query, - system_prompt=config.system_prompt, - ): - yield VisionChunk(text=token) - except Exception: - _LOGGER.exception("Live VLM stream failed") - yield VisionChunk(text="VLM server unavailable — please retry.") - finally: - await config.endpoint.set_status("idle", request.participant_id) - - try: - yield FunctionInfo.create( - single_fn=answer, - stream_fn=stream, - description="Answer a question about a participant's current live camera view.", - ) - finally: - config._frames = None - - class VisionToolsConfig(FunctionGroupBaseConfig, name="xr_vision_tools"): """Configure the one-shot vision tools used by agent workflows.""" @@ -281,9 +165,5 @@ async def look_past(request: HistoricalVisionRequest) -> LiveVisionResult: "HistoricalVisionRequest", "LiveVisionRequest", "LiveVisionResult", - "StreamingVisionConfig", - "VisionChunk", - "VisionRequest", - "VisionResult", "VisionToolsConfig", ] 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 deleted file mode 100644 index 002b9ca5c..000000000 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/live_vision.py +++ /dev/null @@ -1,342 +0,0 @@ -# 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/agent-sdk/xr-ai-tools/README.md b/agent-sdk/xr-ai-tools/README.md new file mode 100644 index 000000000..f33ee8483 --- /dev/null +++ b/agent-sdk/xr-ai-tools/README.md @@ -0,0 +1,81 @@ + + +# XR AI native tools + +`xr-ai-tools` is the toolkit-independent native tools layer for XR AI. +`Tool` gives voice, background triggers, and model-driven agents one typed +Pydantic invocation interface. NeMo Relay manages every tool execution; +model-backed tools use injected `xr-ai-models` services. + +## Native tools and model tool calls + +The base install supplies finite `Tool` and streaming `AsyncTool` types. Install +`xr-ai-tools[relay]` for the small model tool-call helpers: + +```python +from pydantic import BaseModel +from xr_ai_models import ChatMessage +from xr_ai_tools import Tool, ToolSet +from xr_ai_tools.tool_calling import handle_tool_call, tool_definitions + + +class LookupRequest(BaseModel): + query: str + + +class LookupResult(BaseModel): + answer: str + + +async def lookup(request: LookupRequest) -> LookupResult: + return LookupResult(answer=request.query) + + +lookup_tool = Tool( + "lookup", + "Look up one answer.", + LookupRequest, + LookupResult, + lookup, +) +tools = (lookup_tool,) +tool_set = ToolSet(tools) + +response = await llm.chat(messages, tools=tool_definitions(tools)) +messages.append( + ChatMessage( + role="assistant", + content=response.content, + tool_calls=response.tool_calls, + ) +) +for call in response.tool_calls or (): + result = await handle_tool_call(call, tool_set) + messages.append(result.message) + if result.return_direct: + final_answer = result.message.content + break +``` + +`tool_definitions(...)` adapts native tools to `xr-ai-models` `ToolDef` values. +`handle_tool_call(...)` validates and invokes one model-produced `ToolCall`, then +returns a tool-role `ChatMessage` plus its `return_direct` hint. The application +or agent owns prompts, model calls, conversation state, iteration policy, and +whether calls run sequentially or concurrently. + +## Finite and streaming live vision tools + +Install `xr-ai-tools[live-vision]` for two independent current-frame +tools. `LiveVisionTool` is a finite `Tool` that returns one complete +`VisionResponse` for agentic planning. `StreamingVisionTool` is an `AsyncTool` +that yields typed `VisionChunk` values. It has no voice dependency or output +transport; applications decide how to consume its async stream. + +Each tool owns its own participant frame source and calls an injected +`VLMService`. Both forward controlled Relay headers and redact inline camera +frames from events while preserving provider input. The finite path uses +Relay's managed tool and LLM boundaries; the streaming path uses a typed tool +scope around Relay's managed streaming LLM boundary. diff --git a/agent-sdk/xr-ai-tools/pyproject.toml b/agent-sdk/xr-ai-tools/pyproject.toml new file mode 100644 index 000000000..f6e23a6fb --- /dev/null +++ b/agent-sdk/xr-ai-tools/pyproject.toml @@ -0,0 +1,27 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "xr-ai-tools" +version = "0.1.0" +description = "Native Relay-managed tools for XR AI." +requires-python = ">=3.11,<3.13" +dependencies = [ + "nemo-relay>=0.7.2,<0.8", + "pydantic>=2.10", +] + +[project.optional-dependencies] +relay = ["xr-ai-models"] +live-vision = ["numpy>=1.24", "Pillow>=10.0", "xr-ai-hub-client", "xr-ai-models"] + +[tool.uv.sources] +xr-ai-models = { path = "../xr-ai-models", editable = true } +xr-ai-hub-client = { path = "../xr-ai-hub-client", editable = true } + +[tool.hatch.build.targets.wheel] +packages = ["xr_ai_tools"] diff --git a/agent-sdk/xr-ai-tools/xr_ai_tools/__init__.py b/agent-sdk/xr-ai-tools/xr_ai_tools/__init__.py new file mode 100644 index 000000000..3d08374cf --- /dev/null +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/__init__.py @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Toolkit-independent native XR tools.""" + +from .async_tools import AsyncTool +from .tools import Tool, ToolInvocationResult, ToolSet + +__all__ = [ + "AsyncTool", + "Tool", + "ToolInvocationResult", + "ToolSet", +] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py b/agent-sdk/xr-ai-tools/xr_ai_tools/_pixels.py similarity index 100% rename from agent-sdk/xr-ai-nat/xr_ai_nat/_pixels.py rename to agent-sdk/xr-ai-tools/xr_ai_tools/_pixels.py diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py b/agent-sdk/xr-ai-tools/xr_ai_tools/_relay.py similarity index 100% rename from agent-sdk/xr-ai-nat/xr_ai_nat/_relay.py rename to agent-sdk/xr-ai-tools/xr_ai_tools/_relay.py diff --git a/agent-sdk/xr-ai-tools/xr_ai_tools/_vision.py b/agent-sdk/xr-ai-tools/xr_ai_tools/_vision.py new file mode 100644 index 000000000..7de0bf505 --- /dev/null +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/_vision.py @@ -0,0 +1,196 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared schemas and Relay codecs for live-vision tools.""" + +from __future__ import annotations + +import copy +from collections.abc import Mapping +from typing import Any + +import nemo_relay +from pydantic import BaseModel, ConfigDict, Field + +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 streamed current-frame answer.""" + + text: str = Field(description="A partial fragment of the streamed answer text.") + + +def register_frame_sanitizer() -> None: + """Redact the current frame from events in the active tool scope.""" + + nemo_relay.scope_local.register_llm_sanitize_request( + nemo_relay.scope.get_handle(), + "xr-ai-live-frame", + 0, + _sanitize_live_frame, + ) + + +def relay_request( + system_prompt: str, + image_url: str, + query: str, +) -> nemo_relay.LLMRequest: + """Build the shared OpenAI-compatible current-frame request.""" + + return nemo_relay.LLMRequest( + {}, + { + "model": VLM_CALL_NAME, + "messages": [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": [ + {"type": "text", "text": query}, + {"type": "image_url", "image_url": {"url": image_url}}, + ], + }, + ], + }, + ) + + +def vision_inputs(content: Mapping[str, object]) -> tuple[str, str, str]: + """Decode the image, query, and system prompt from a Relay request.""" + + 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 openai_response(text: str) -> dict[str, Any]: + """Build the complete Relay response used by both VLM paths.""" + + return { + "model": VLM_CALL_NAME, + "choices": [ + { + "message": {"role": "assistant", "content": text}, + "finish_reason": "stop", + } + ], + } + + +def response_text(raw_response: object) -> str: + """Decode complete text from a Relay VLM response.""" + + 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: + """Decode one text fragment from a Relay VLM stream chunk.""" + + 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 + + +def _sanitize_live_frame( + request: nemo_relay.LLMRequest, + _context: nemo_relay.LlmSanitizeRequestContext, +) -> nemo_relay.LLMRequest: + 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 _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 diff --git a/agent-sdk/xr-ai-tools/xr_ai_tools/async_tools.py b/agent-sdk/xr-ai-tools/xr_ai_tools/async_tools.py new file mode 100644 index 000000000..4da6a08f3 --- /dev/null +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/async_tools.py @@ -0,0 +1,55 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Typed tools that yield asynchronous result chunks.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable, Mapping +from typing import Generic, TypeVar + +import nemo_relay +from pydantic import BaseModel + +RequestT = TypeVar("RequestT", bound=BaseModel) +ChunkT = TypeVar("ChunkT", bound=BaseModel) + + +class AsyncTool(Generic[RequestT, ChunkT]): + """A validated asynchronous tool independent of any output transport.""" + + 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 | Mapping[str, object], + ) -> AsyncIterator[ChunkT]: + """Validate and yield one typed result stream under a tool scope.""" + + value = self.request_model.model_validate(request) + with nemo_relay.scope.scope( + self.name, + nemo_relay.ScopeType.Tool, + input=value.model_dump(mode="json"), + ): + async for chunk in self.handler(value): + yield self.chunk_model.model_validate(chunk) + + +__all__ = ["AsyncTool"] diff --git a/agent-sdk/xr-ai-tools/xr_ai_tools/live_vision.py b/agent-sdk/xr-ai-tools/xr_ai_tools/live_vision.py new file mode 100644 index 000000000..61ef57e2e --- /dev/null +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/live_vision.py @@ -0,0 +1,111 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Finite current-frame vision for ordinary agent tool calls.""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +import nemo_relay +from nemo_relay.codecs import OpenAIChatCodec +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 ._vision import ( + VLM_CALL_NAME, + VisionRequest, + VisionResponse, + openai_response, + register_frame_sanitizer, + relay_request, + response_text, + vision_inputs, +) +from .tools import Tool + +_LOGGER = logging.getLogger(__name__) + + +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)) + + await self.endpoint.set_status("processing", request.participant_id) + try: + register_frame_sanitizer() + response = await nemo_relay.llm.execute( + VLM_CALL_NAME, + relay_request(self.system_prompt, 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 _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))) + + 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) + + +__all__ = ["LiveVisionTool", "VisionRequest", "VisionResponse"] diff --git a/agent-sdk/xr-ai-tools/xr_ai_tools/streaming_vision.py b/agent-sdk/xr-ai-tools/xr_ai_tools/streaming_vision.py new file mode 100644 index 000000000..1eac8f9b9 --- /dev/null +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/streaming_vision.py @@ -0,0 +1,127 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Streaming current-frame vision as a standalone async tool.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator +from typing import Any + +import nemo_relay +from nemo_relay.codecs import OpenAIChatCodec +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 ._vision import ( + VLM_CALL_NAME, + VisionChunk, + VisionRequest, + openai_response, + register_frame_sanitizer, + relay_request, + stream_text, + vision_inputs, +) +from .async_tools import AsyncTool + +_LOGGER = logging.getLogger(__name__) + + +class StreamingVisionTool(AsyncTool[VisionRequest, VisionChunk]): + """A typed current-frame tool that yields answer fragments asynchronously.""" + + 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__( + "stream_current_frame", + "Stream an answer 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 + + await self.endpoint.set_status("processing", request.participant_id) + fragments: list[str] = [] + emitted_output = False + try: + register_frame_sanitizer() + stream = await nemo_relay.llm.stream_execute( + VLM_CALL_NAME, + relay_request(self.system_prompt, 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: + emitted_output = True + yield VisionChunk(text=text) + except Exception: + _LOGGER.exception("Live VLM stream failed") + if not emitted_output: + 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))) + + 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}}]} + + +__all__ = ["StreamingVisionTool", "VisionChunk", "VisionRequest"] diff --git a/agent-sdk/xr-ai-tools/xr_ai_tools/tool_calling.py b/agent-sdk/xr-ai-tools/xr_ai_tools/tool_calling.py new file mode 100644 index 000000000..5799e8232 --- /dev/null +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/tool_calling.py @@ -0,0 +1,60 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Helpers for dispatching model-selected native tool calls.""" + +from __future__ import annotations + +import json +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any + +from xr_ai_models import ChatMessage, ToolCall, ToolDef + +from .tools import Tool, ToolSet + + +@dataclass(frozen=True, slots=True) +class ToolCallResult: + """One model-ready tool response and its control-flow hint.""" + + message: ChatMessage + return_direct: bool + + +def tool_definitions(tools: Iterable[Tool[Any, Any]]) -> tuple[ToolDef, ...]: + """Return model-service definitions for native tools.""" + + return tuple( + ToolDef( + name=tool.name, + description=tool.description, + parameters=tool.request_model.model_json_schema(), + ) + for tool in tools + ) + + +async def handle_tool_call(call: ToolCall, tools: ToolSet) -> ToolCallResult: + """Invoke one model-produced call and return its tool-role message.""" + + tool = tools.get(call.name) + if tool is None: + content = json.dumps({"error": "unknown_tool", "tool": call.name}) + return_direct = False + else: + invocation = await tool.invoke(call.arguments) + content = invocation.content + return_direct = invocation.return_direct + return ToolCallResult( + message=ChatMessage( + role="tool", + content=content, + tool_call_id=call.id, + ), + return_direct=return_direct, + ) + + +__all__ = ["ToolCallResult", "handle_tool_call", "tool_definitions"] diff --git a/agent-sdk/xr-ai-nat/xr_ai_nat/tools.py b/agent-sdk/xr-ai-tools/xr_ai_tools/tools.py similarity index 86% rename from agent-sdk/xr-ai-nat/xr_ai_nat/tools.py rename to agent-sdk/xr-ai-tools/xr_ai_tools/tools.py index 37ae59cee..691eb2df5 100644 --- a/agent-sdk/xr-ai-nat/xr_ai_nat/tools.py +++ b/agent-sdk/xr-ai-tools/xr_ai_tools/tools.py @@ -54,18 +54,6 @@ def __init__( self._result_codec = typed.PydanticCodec(result_model) self._render_result = render_result or _json_result - def to_openai(self) -> dict[str, Any]: - """Return the OpenAI-compatible definition supplied to an agent model.""" - - return { - "type": "function", - "function": { - "name": self.name, - "description": self.description, - "parameters": self.request_model.model_json_schema(), - }, - } - async def execute(self, request: RequestT) -> ResultT: """Run one validated request through the shared Relay tool lifecycle.""" @@ -109,7 +97,7 @@ async def _execute_handler(self, request: RequestT) -> ResultT: class ToolSet: - """A non-overlapping tool catalog used by one native agent.""" + """A non-overlapping native tool catalog.""" def __init__(self, tools: Iterable[Tool[Any, Any]]) -> None: by_name: dict[str, Tool[Any, Any]] = {} @@ -119,12 +107,6 @@ def __init__(self, tools: Iterable[Tool[Any, Any]]) -> None: by_name[tool.name] = tool self._by_name = by_name - @property - def definitions(self) -> tuple[dict[str, Any], ...]: - """Return model-visible schemas in registration order.""" - - return tuple(tool.to_openai() for tool in self._by_name.values()) - def get(self, name: str) -> Tool[Any, Any] | None: """Return the named tool when this catalog owns it.""" diff --git a/docs/changelog.md b/docs/changelog.md index bee8891df..d76a1c6c6 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -9,16 +9,34 @@ 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 +### 2026-08-12 — Tool-call handling is not an agent runtime + +`agents.py`, `agent_runner.py`, `Agent`, and `AgentRunner` are removed. +`tool_calling.py` only adapts native schemas to model `ToolDef` values and +handles one model-produced `ToolCall` at a time. Applications own prompts, model +calls, history, iteration, and concurrency. + +The unused NAT `StreamingVisionConfig` and its schemas are also removed because +`StreamingVisionTool` replaces that surface. The still-used `VisionToolsConfig` +remains in NAT because its recorded-frame tool has no native replacement yet. + +### 2026-08-12 — Native tools live outside the NAT compatibility package + +The Relay-managed tool modules moved from `xr-ai-nat` to the +dedicated `xr-ai-tools` package. Live vision is split into independent finite +and streaming tool modules. `xr-ai-nat` now remains only as the NeMo Agent +Toolkit compatibility surface during migration. + +### 2026-08-12 — Agentic vision is finite; streaming vision is asynchronous 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. +planning. `StreamingVisionTool` is a separate, transport-independent +`AsyncTool` that yields typed chunks while its provider call uses Relay's +managed streaming LLM API. Applications may adapt that stream to voice or any +other consumer; the tool itself has no voice behavior. 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 802f5e14a..d1c5a39f4 100644 --- a/docs/nemo-agent-toolkit-migration.md +++ b/docs/nemo-agent-toolkit-migration.md @@ -11,13 +11,12 @@ for all model HTTP, and reach clients only through the Hub IPC SDK. ## Target architecture -NeMo Relay is the local execution boundary. The XR-owned `xr-ai-nat` package -is becoming a toolkit-independent tools layer: Pydantic tool schemas, trigger -dispatch, and the generic `AgentRunner` protocol. Its bundled agent is a -bounded tool loop, while `as_agent_tool(...)` lets custom or future -Fabric-backed runners use the same trigger path. Relay owns LLM and tool -lifecycles, middleware, guardrails, and telemetry. Existing NeMo Agent Toolkit -function groups remain compatibility extras until their concrete tools migrate. +NeMo Relay is the local execution boundary. The XR-owned `xr-ai-tools` package +is the toolkit-independent tools layer: Pydantic tool schemas, trigger +dispatch, and small helpers that adapt those tools to `xr-ai-models` tool-call +types. Applications own their agents, model calls, history, and loop policy. +Relay owns tool lifecycles, middleware, guardrails, and telemetry. Existing +NeMo Agent Toolkit function groups remain compatibility extras until their concrete tools migrate. NeMo Platform and NeMo Fabric are deployment and evaluation integrations, not worker dependencies. Platform currently requires Python 3.12 or 3.13 and owns @@ -28,8 +27,8 @@ remain optional launch targets after the local runtime has migrated. ```text XR worker - -> xr-ai-nat: typed tools and trigger dispatch - -> NeMo Relay: managed tool and agent execution, guardrails, telemetry + -> xr-ai-tools: typed tools and trigger dispatch + -> NeMo Relay: managed tool execution, guardrails, telemetry -> xr-ai-models: private model boundary used by model-backed tools -> Hub IPC: media and client data @@ -49,13 +48,12 @@ acceptance behavior rather than an implementation dependency. ## Focused PR sequence -1. **Native tools foundation** — make `xr-ai-nat` independent of NeMo Agent - 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. +1. **Native tools foundation** — add `xr-ai-tools` with Relay-managed + tools plus model tool-call dispatch helpers, and retain existing function + groups behind legacy extras. 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. + retain streaming in a separate transport-independent async tool, + 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. @@ -79,6 +77,6 @@ hand-rolled model HTTP client behind. The retirement is complete only when repository-wide search finds no runtime imports from `nat` and no `nvidia-nat-*` dependency, worker requirements select -only toolkit-independent `xr-ai-nat` extras, all migrated samples pass their +`xr-ai-tools` for toolkit-independent tools, all migrated samples pass their unit and evaluation suites, and `THIRD_PARTY_NOTICES.md` no longer lists NeMo Agent Toolkit. diff --git a/docs/source/components/agent-sdk.md b/docs/source/components/agent-sdk.md index fad378cd0..579c5d0bf 100644 --- a/docs/source/components/agent-sdk.md +++ b/docs/source/components/agent-sdk.md @@ -21,9 +21,10 @@ from: - **`xr-ai-hub-client`** — the minimal pyzmq + msgpack IPC library every agent uses to talk to the XR-Media-Hub (refer to {doc}`server-runtime`). No LiveKit or FastAPI dependency. -- **`xr-ai-nat`** — Relay-managed native tools, the generic `AgentRunner` - protocol, and a basic tool-driven agent. Its legacy extras retain existing - NeMo Agent Toolkit function groups while their concrete capabilities migrate. +- **`xr-ai-tools`** — Relay-managed native tools and model tool-call + workflow helpers. +- **`xr-ai-nat`** — legacy NeMo Agent Toolkit function groups retained while + their concrete capabilities migrate. --- @@ -206,28 +207,25 @@ The clients can be exercised without a GPU. --- -## Native tools and agents +## Native tools and model tool calls -`xr-ai-nat` is the native migration target for model-driven XR composition. +`xr-ai-tools` is the native migration target for model-driven XR composition. `Tool` declares Pydantic request and response boundaries and executes its -handler through NeMo Relay. `AgentRunner` is the generic async-turn protocol; -`xr_ai_nat.agents.Agent` is its bounded, stateless tool-loop implementation. -It builds OpenAI-compatible tool definitions from those schemas and sends each -model request through an injected `LLMService`. - -Relay scopes each turn and manages the LLM and tool lifecycles. The base package -does not add an HTTP client, a Hub transport, or an implicit conversation store. -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. +handler through NeMo Relay. `tool_definitions(...)` adapts a native catalog to +`xr-ai-models` definitions; `handle_tool_call(...)` invokes one model-selected +call and returns the tool-role message. + +The package does not implement an agent, model loop, Hub transport, or implicit +conversation store. Applications own prompts, LLM calls, history, iteration, +foreground selection, workflow state, and background work. + +`xr_ai_tools.live_vision.LiveVisionTool` is the finite current-frame tool; it +returns one complete observation through Relay's managed tool and LLM +boundaries. `xr_ai_tools.streaming_vision.StreamingVisionTool` is a separate +`AsyncTool` that yields typed chunks around Relay's managed streaming LLM +boundary. It has no voice or output-transport dependency. Both tools own their +own frame sources and redact inline camera data from Relay events without +changing provider input. ## xr-ai-nat model bridge diff --git a/tests/pyproject.toml b/tests/pyproject.toml index e1150b59f..a97fe75cb 100644 --- a/tests/pyproject.toml +++ b/tests/pyproject.toml @@ -14,7 +14,8 @@ dependencies = [ # Pulled in via editable installs of the workspace packages. "xr-ai-hub-client", "xr-ai-models", - "xr-ai-nat[agents,relay,services,vision]", + "xr-ai-nat[agents,services,vision]", + "xr-ai-tools[live-vision]", "xr-ai-pipecat", "xr-ai-voice", "xr-media-hub", @@ -46,6 +47,7 @@ dependencies = [ xr-ai-hub-client = { path = "../agent-sdk/xr-ai-hub-client", editable = true } xr-ai-models = { path = "../agent-sdk/xr-ai-models", editable = true } xr-ai-nat = { path = "../agent-sdk/xr-ai-nat", editable = true } +xr-ai-tools = { path = "../agent-sdk/xr-ai-tools", editable = true } xr-ai-pipecat = { path = "../agent-sdk/xr-ai-pipecat", editable = true } xr-ai-voice = { path = "../agent-sdk/xr-ai-voice", editable = true } xr-media-hub = { path = "../services/xr-media-hub", editable = true } diff --git a/tests/test_native_tools.py b/tests/test_native_tools.py index 52335e291..9732408d5 100644 --- a/tests/test_native_tools.py +++ b/tests/test_native_tools.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Contracts for Relay-managed native tools and tool-driven agents.""" +"""Contracts for Relay-managed native tools and model-selected tool calls.""" from __future__ import annotations @@ -10,11 +10,10 @@ import nemo_relay import pytest -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.agents import Agent, ToolLoopLimitError, _response_from_openai +from xr_ai_models import ChatMessage, ToolCall, ToolDef +from xr_ai_tools import AsyncTool, Tool, ToolSet +from xr_ai_tools.tool_calling import handle_tool_call, tool_definitions class AddRequest(BaseModel): @@ -30,333 +29,120 @@ class AddResult(BaseModel): total: int -class AskRequest(BaseModel): - """One text request delegated to an agent tool.""" +async def add(request: AddRequest) -> AddResult: + return AddResult(total=request.left + request.right) - text: str +async def add_stream(request: AddRequest) -> AsyncIterator[AddResult]: + yield AddResult(total=request.left) + yield AddResult(total=request.left + request.right) -class AskResult(BaseModel): - """The agent tool's text result.""" - text: str +async def test_async_tool_validates_and_yields_typed_chunks() -> None: + tool = AsyncTool( + "stream_add", + "Stream a running total.", + AddRequest, + AddResult, + add_stream, + ) + chunks = [chunk async for chunk in tool.stream({"left": 2, "right": 3})] -async def add(request: AddRequest) -> AddResult: - return AddResult(total=request.left + request.right) + assert chunks == [AddResult(total=2), AddResult(total=5)] -class _ToolCallingLLM: - capabilities = Capabilities(tool_calls=True) - - def __init__(self) -> None: - self.calls: list[tuple[list[ChatMessage], list[ToolDef] | None]] = [] - - async def chat(self, messages, *, tools=None, **_kwargs) -> ChatResponse: - self.calls.append((list(messages), list(tools) if tools else None)) - if len(self.calls) == 1: - return ChatResponse( - content="", - reasoning=None, - tool_calls=[ - ToolCall( - id="add-call", - name="add", - arguments=json.dumps({"left": 2, "right": 3}), - ) - ], - finish_reason="tool_calls", - raw={"usage": {"prompt_tokens": 10, "completion_tokens": 3}}, - ) - return ChatResponse( - content="The answer is five.", - reasoning=None, - tool_calls=None, - finish_reason="stop", - raw={"usage": {"prompt_tokens": 20, "completion_tokens": 5}}, - ) - - async def health(self) -> bool: - return True +def test_tool_definitions_adapt_native_tools_for_model_services() -> None: + tool = Tool("add", "Add two integers.", AddRequest, AddResult, add) - async def close(self) -> None: - return None + assert tool_definitions((tool,)) == ( + ToolDef( + name="add", + description="Add two integers.", + parameters=AddRequest.model_json_schema(), + ), + ) - async def stream(self, *_args, **_kwargs) -> AsyncIterator[str]: - if False: - yield "" +async def test_handle_tool_call_returns_a_model_ready_tool_message() -> None: + tool = Tool("add", "Add two integers.", AddRequest, AddResult, add) -async def test_agent_executes_a_native_tool_and_returns_the_final_model_answer() -> None: - llm = _ToolCallingLLM() - agent = Agent( - name="calculator", - llm=llm, - system_prompt="Use the add tool.", - tools=(Tool("add", "Add two integers.", AddRequest, AddResult, add),), - max_iterations=2, + result = await handle_tool_call( + ToolCall(id="add-call", name="add", arguments='{"left":2,"right":3}'), + ToolSet((tool,)), ) - result = await agent.run("What is two plus three?") - - assert result.text == "The answer is five." - assert len(llm.calls) == 2 - first_messages, first_tools = llm.calls[0] - assert [message.role for message in first_messages] == ["system", "user"] - assert first_tools is not None - assert first_tools[0].name == "add" - second_messages, _ = llm.calls[1] - assert [(message.role, message.content) for message in second_messages[-2:]] == [ - ("assistant", ""), - ("tool", '{"total":5}'), - ] + assert result.message == ChatMessage( + role="tool", + content='{"total":5}', + tool_call_id="add-call", + ) + assert result.return_direct is False -async def test_tools_are_relay_managed_for_agent_and_direct_invocation() -> None: +async def test_handled_tool_calls_use_the_relay_tool_lifecycle() -> None: tool = Tool("add", "Add two integers.", AddRequest, AddResult, add) - llm = _ToolCallingLLM() - agent = Agent( - name="calculator-lifecycle", - llm=llm, - system_prompt="Use the add tool.", - tools=(tool,), - max_iterations=2, - ) events = [] - subscriber = "xr-ai-native-tools-lifecycle" + subscriber = "xr-ai-native-tool-call-lifecycle" nemo_relay.subscribers.register(subscriber, events.append) try: - assert await tool.execute(AddRequest(left=1, right=2)) == AddResult(total=3) - await agent.run("What is two plus three?") + await handle_tool_call( + ToolCall(id="add-call", name="add", arguments='{"left":2,"right":3}'), + ToolSet((tool,)), + ) await nemo_relay.subscribers.flush_async() finally: nemo_relay.subscribers.deregister(subscriber) - categories = {getattr(event, "category", None) for event in events} - assert {"llm", "tool"} <= categories - - -async def test_agent_can_be_exposed_as_a_normal_native_tool() -> None: - class _FinalAnswerLLM(_ToolCallingLLM): - async def chat(self, messages, *, tools=None, **_kwargs) -> ChatResponse: - self.calls.append((list(messages), list(tools) if tools else None)) - return ChatResponse( - content="Handled by the agent tool.", - reasoning=None, - tool_calls=None, - finish_reason="stop", - raw={}, - ) - - agent = Agent( - name="delegate", - llm=_FinalAnswerLLM(), - system_prompt="Answer directly.", - tools=(), - ) - tool = as_agent_tool( - name="delegate", - description="Delegate one request to the agent.", - agent=agent, - request_model=AskRequest, - result_model=AskResult, - request=lambda value: value.text, - response=lambda result: AskResult(text=result.text), - ) - - assert await tool.execute(AskRequest(text="Hello")) == AskResult( - text="Handled by the agent tool.", - ) - - -async def test_custom_agent_runner_can_be_exposed_as_a_normal_native_tool() -> None: - class _CustomRunner: - async def run(self, request: str) -> AskResult: - return AskResult(text=f"Custom runner: {request}") - - runner: AgentRunner[str, AskResult] = _CustomRunner() - tool = as_agent_tool( - name="custom_delegate", - description="Delegate one request to a custom agent runner.", - agent=runner, - request_model=AskRequest, - result_model=AskResult, - request=lambda value: value.text, - response=lambda result: result, - ) - - assert await tool.execute(AskRequest(text="Hello")) == AskResult( - text="Custom runner: Hello", - ) - - -async def test_agent_forwards_relay_rewritten_request_to_the_model(monkeypatch) -> None: - class _RecordingLLM: - capabilities = Capabilities() - - def __init__(self) -> None: - self.messages: list[ChatMessage] = [] - self.headers: dict[str, str] = {} - - async def chat(self, messages, *, headers=None, **_kwargs) -> ChatResponse: - self.messages = list(messages) - self.headers = dict(headers or {}) - return ChatResponse("Rewritten.", None, None, "stop", {}) - - async def health(self) -> bool: - return True - - async def close(self) -> None: - return None - - async def stream(self, *_args, **_kwargs) -> AsyncIterator[str]: - if False: - yield "" - - llm = _RecordingLLM() - observed: dict[str, object] = {} - - async def execute(_name, request, invoke, **kwargs): - observed["content"] = request.content - observed["codec"] = kwargs["codec"] - observed["response_codec"] = kwargs["response_codec"] - rewritten = dict(request.content) - rewritten["messages"] = [ - {"role": "system", "content": "Answer directly."}, - {"role": "user", "content": "rewritten request"}, - ] - return await invoke(nemo_relay.LLMRequest({"X-Relay-Session": "turn-7"}, rewritten)) - - monkeypatch.setattr(nemo_relay.llm, "execute", execute) - result = await Agent( - name="relay-boundary", - llm=llm, - system_prompt="Answer directly.", - tools=(), - ).run("original request") - - assert result.text == "Rewritten." - initial_content = observed["content"] - assert isinstance(initial_content, dict) - assert "tools" not in initial_content - assert isinstance(observed["codec"], OpenAIChatCodec) - assert isinstance(observed["response_codec"], OpenAIChatCodec) - assert llm.headers == {"X-Relay-Session": "turn-7"} - assert llm.messages[-1].content == "rewritten request" - - -def test_agent_accepts_null_content_for_tool_call_only_response() -> None: - response = _response_from_openai( - { - "choices": [ - { - "message": { - "content": None, - "tool_calls": [ - { - "id": "lookup-1", - "function": {"name": "lookup", "arguments": "{}"}, - } - ], - }, - "finish_reason": "tool_calls", - } - ] - } - ) - - assert response.content == "" - assert response.tool_calls == [ToolCall(id="lookup-1", name="lookup", arguments="{}")] + 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) - result = await tool.invoke('{"left":"not-an-int"}') + result = await handle_tool_call( + ToolCall(id="add-call", name="add", arguments='{"left":"not-an-int"}'), + ToolSet((tool,)), + ) assert result.return_direct is False - payload = json.loads(result.content) + assert isinstance(result.message.content, str) + payload = json.loads(result.message.content) assert payload["error"] == "invalid_tool_arguments" assert "right" in payload["detail"] -async def test_return_direct_tool_finishes_without_an_extra_model_call() -> None: - llm = _ToolCallingLLM() - agent = Agent( - name="calculator", - llm=llm, - system_prompt="Use the add tool.", - tools=( - Tool( - "add", - "Add two integers.", - AddRequest, - AddResult, - add, - return_direct=True, - ), - ), +async def test_unknown_tool_is_returned_to_the_model_for_repair() -> None: + result = await handle_tool_call( + ToolCall(id="missing-call", name="missing", arguments="{}"), + ToolSet(()), ) - result = await agent.run("What is two plus three?") - - assert result.text == '{"total":5}' - assert len(llm.calls) == 1 + assert result.message == ChatMessage( + role="tool", + content='{"error": "unknown_tool", "tool": "missing"}', + tool_call_id="missing-call", + ) + assert result.return_direct is False -async def test_unknown_tool_is_returned_to_the_model_for_repair() -> None: - class _UnknownToolLLM(_ToolCallingLLM): - async def chat(self, messages, *, tools=None, **kwargs) -> ChatResponse: - response = await super().chat(messages, tools=tools, **kwargs) - if len(self.calls) == 1: - return ChatResponse( - content="", - reasoning=None, - tool_calls=[ToolCall(id="missing", name="missing", arguments="{}")], - finish_reason="tool_calls", - raw={}, - ) - return response - - llm = _UnknownToolLLM() - agent = Agent( - name="calculator", - llm=llm, - system_prompt="Use the available tools.", - tools=(Tool("add", "Add two integers.", AddRequest, AddResult, add),), - max_iterations=2, +async def test_handle_tool_call_preserves_return_direct() -> None: + tool = Tool( + "add", + "Add two integers.", + AddRequest, + AddResult, + add, + return_direct=True, ) - result = await agent.run("Calculate.") - - assert result.text == "The answer is five." - content = llm.calls[1][0][-1].content - assert isinstance(content, str) - assert json.loads(content) == {"error": "unknown_tool", "tool": "missing"} - - -async def test_agent_enforces_its_model_iteration_budget() -> None: - class _LoopingLLM(_ToolCallingLLM): - async def chat(self, messages, *, tools=None, **kwargs) -> ChatResponse: - self.calls.append((list(messages), list(tools) if tools else None)) - return ChatResponse( - content="", - reasoning=None, - tool_calls=[ToolCall(id=str(len(self.calls)), name="add", arguments='{"left":1,"right":1}')], - finish_reason="tool_calls", - raw={}, - ) - - agent = Agent( - name="calculator", - llm=_LoopingLLM(), - system_prompt="Use the add tool.", - tools=(Tool("add", "Add two integers.", AddRequest, AddResult, add),), - max_iterations=2, + result = await handle_tool_call( + ToolCall(id="add-call", name="add", arguments='{"left":2,"right":3}'), + ToolSet((tool,)), ) - with pytest.raises(ToolLoopLimitError, match="exhausted 2"): - await agent.run("Loop forever.") + assert result.message.content == '{"total":5}' + assert result.return_direct is True def test_tool_sets_reject_duplicate_names() -> None: diff --git a/tests/test_simple_vlm_example_worker.py b/tests/test_simple_vlm_example_worker.py index 4f09bd623..abfa06a70 100644 --- a/tests/test_simple_vlm_example_worker.py +++ b/tests/test_simple_vlm_example_worker.py @@ -28,12 +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 ( # noqa: E402 - LiveVisionResponder, +from xr_ai_tools.live_vision import ( # noqa: E402 LiveVisionTool, VisionRequest, VisionResponse, ) +from xr_ai_tools.streaming_vision import StreamingVisionTool # noqa: E402 class _Service: @@ -58,8 +58,8 @@ def shutdown(self) -> None: self.shutdown_calls += 1 -class _LiveVisionTool: - instances: list["_LiveVisionTool"] = [] +class _StreamingVisionTool: + instances: list["_StreamingVisionTool"] = [] def __init__(self, **kwargs) -> None: self.kwargs = kwargs @@ -70,13 +70,8 @@ def __init__(self, **kwargs) -> None: 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.tool.requests.append(request) + self.requests.append(request) for text in ("a ", "blue square"): yield SimpleNamespace(text=text) @@ -149,7 +144,7 @@ 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[relay,live-vision]" in dependencies + assert "xr-ai-tools[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 @@ -311,8 +306,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) + monkeypatch.setattr(app, "StreamingVisionTool", _StreamingVisionTool) def make_session(**kwargs): session = VoiceSession(transport=transport, **kwargs) # type: ignore[arg-type] @@ -342,7 +336,7 @@ def __init__(self, **kwargs) -> None: monkeypatch.setattr(app, "VoiceSession", make_session) monkeypatch.setattr(app, "TextMessageInput", CaptureTextInput) - _LiveVisionTool.instances.clear() + _StreamingVisionTool.instances.clear() await app.run_app(config, ready_file=ready_file) @@ -351,17 +345,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 _StreamingVisionTool.instances[0].kwargs["endpoint"] is transport.endpoint + assert _StreamingVisionTool.instances[0].kwargs["system_prompt"] == config.system_prompt + assert _StreamingVisionTool.instances[0].kwargs["frame_max_age_s"] == ( config.frame_max_age_s ) - assert _LiveVisionTool.instances[0].kwargs["frame_timeout_s"] == ( + assert _StreamingVisionTool.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 _StreamingVisionTool.instances[0].released == ["alice"] + assert _StreamingVisionTool.instances[0].requests[0].participant_id == "alice" + assert _StreamingVisionTool.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] @@ -433,16 +427,14 @@ def add_header(_name, request, annotated): assert any("" in event for event in llm_events) -async def test_sample_handler_streams_a_live_frame_question() -> None: +async def test_streaming_vision_tool_yields_typed_chunks() -> None: endpoint = _LiveEndpoint() vlm = _StreamingVlm() - vision = LiveVisionTool( + vision = StreamingVisionTool( endpoint=cast(ProcessorEndpoint, endpoint), vlm=cast(VLMService, vlm), system_prompt="Answer briefly.", ) - - handler = app._make_vision_handler(LiveVisionResponder(vision)) assert endpoint.frame_callback is not None await endpoint.frame_callback( FrameSignal( @@ -472,29 +464,28 @@ def add_header(_name, request, 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", - text="What is shown?", - fresh_match=True, - timestamp_us=123, + chunks = [ + chunk + async for chunk in vision.stream( + VisionRequest( + participant_id="alice", + query="What is shown?", + ) ) - ) - 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"] + assert [chunk.text for chunk in chunks] == ["a ", "blue ", "square"] 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} + assert {"tool", "llm"} <= {getattr(event, "category", None) for event in events} llm_events = [ event.to_json() for event in events @@ -503,3 +494,94 @@ def add_header(_name, request, annotated): 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_streaming_vision_stops_after_a_partial_stream_failure(monkeypatch) -> None: + class _PartialFailureVlm: + def __init__(self) -> None: + self.calls = [] + + async def stream( + self, + image, + question: str, + *, + system_prompt: str = "", + headers=None, + ): + self.calls.append((image, question, system_prompt, dict(headers or {}))) + yield "The object is " + raise RuntimeError("stream disconnected") + + endpoint = _LiveEndpoint() + vision = StreamingVisionTool( + endpoint=cast(ProcessorEndpoint, endpoint), + vlm=cast(VLMService, _PartialFailureVlm()), + ) + + async def current_image(_participant_id: str) -> str: + return "data:image/jpeg;base64,frame" + + monkeypatch.setattr(vision, "_current_image", current_image) + + chunks = [ + chunk.text + async for chunk in vision.stream( + VisionRequest(participant_id="alice", query="What is shown?"), + ) + ] + + assert chunks == ["The object is "] + assert endpoint.statuses == [("processing", "alice"), ("idle", "alice")] + + +async def test_streaming_vision_reports_failure_before_any_output(monkeypatch) -> None: + class _ImmediateFailureVlm: + async def stream(self, *_args, **_kwargs): + if False: + yield "" + raise RuntimeError("stream unavailable") + + endpoint = _LiveEndpoint() + vision = StreamingVisionTool( + endpoint=cast(ProcessorEndpoint, endpoint), + vlm=cast(VLMService, _ImmediateFailureVlm()), + ) + + async def current_image(_participant_id: str) -> str: + return "data:image/jpeg;base64,frame" + + monkeypatch.setattr(vision, "_current_image", current_image) + + chunks = [ + chunk.text + async for chunk in vision.stream( + VisionRequest(participant_id="alice", query="What is shown?"), + ) + ] + + assert chunks == ["VLM server unavailable — please retry."] + assert endpoint.statuses == [("processing", "alice"), ("idle", "alice")] + + +async def test_vision_tools_propagate_frame_conversion_errors(monkeypatch) -> None: + endpoint = _LiveEndpoint() + vlm = _StreamingVlm() + finite = LiveVisionTool( + endpoint=cast(ProcessorEndpoint, endpoint), + vlm=cast(VLMService, vlm), + ) + streaming = StreamingVisionTool( + endpoint=cast(ProcessorEndpoint, endpoint), + vlm=cast(VLMService, vlm), + ) + + async def malformed_frame(_participant_id: str) -> str: + raise ValueError("malformed pixels") + + monkeypatch.setattr(finite, "_current_image", malformed_frame) + monkeypatch.setattr(streaming, "_current_image", malformed_frame) + request = VisionRequest(participant_id="alice", query="What is shown?") + + with pytest.raises(RuntimeError, match="malformed pixels"): + await finite.execute(request) diff --git a/tests/test_vision_functions.py b/tests/test_vision_functions.py index 3a84671f2..d89e08bf8 100644 --- a/tests/test_vision_functions.py +++ b/tests/test_vision_functions.py @@ -26,8 +26,6 @@ from xr_ai_nat.functions.vision import ( HistoricalVisionRequest, LiveVisionRequest, - StreamingVisionConfig, - VisionRequest, VisionToolsConfig, ) from xr_ai_nat.functions.vision._pixels import encode_image, frame_to_pil @@ -42,11 +40,6 @@ async def ask_image(self, image: Any, question: str, *, system_prompt: str = "") self.calls.append((image, question, system_prompt)) return SimpleNamespace(content=self.content) - async def stream(self, image: Any, question: str, *, system_prompt: str = ""): - self.calls.append((image, question, system_prompt)) - for token in ("a ", "blue ", "square"): - yield token - class _Endpoint: def __init__(self) -> None: @@ -159,61 +152,6 @@ def test_frame_to_pil_supports_non_rgb_frame_formats(pixel_format, data) -> None assert image.size == (2, 2) -# ── StreamingVisionConfig (live-camera streaming) ───────────────────────────── - - -async def test_streaming_vision_function_uses_current_participant_frame() -> None: - endpoint = _Endpoint() - vlm = _Vlm("a blue square") - config = StreamingVisionConfig(endpoint=endpoint, vlm=vlm, system_prompt="Answer briefly.") - - async with WorkflowBuilder() as builder: - function = await builder.add_function("perception", config) - assert endpoint.frame_callback is not None - await endpoint.frame_callback(_seed_signal()) - chunks = [ - chunk.text - async for chunk in function.astream(VisionRequest(participant_id="alice", query="What is shown?")) - ] - answer = await function.ainvoke(VisionRequest(participant_id="alice", query="What is shown?")) - - assert chunks == ["a ", "blue ", "square"] - assert answer.text == "a blue square" - assert answer.status == "ok" - assert endpoint.statuses == [ - ("processing", "alice"), - ("idle", "alice"), - ("processing", "alice"), - ("idle", "alice"), - ] - assert vlm.calls[0][1:] == ("What is shown?", "Answer briefly.") - assert vlm.calls[0][0].startswith("data:image/jpeg;base64,") - - -async def test_streaming_vision_function_reports_unavailable_frame(monkeypatch) -> None: - endpoint = _Endpoint() - vlm = _Vlm("unused") - - async def unavailable_frame(*_args) -> str: - raise FrameUnavailable("No camera frame available — please try again.") - - monkeypatch.setattr("xr_ai_nat.functions.vision.functions._current_image", unavailable_frame) - - async with WorkflowBuilder() as builder: - function = await builder.add_function("perception", StreamingVisionConfig(endpoint=endpoint, vlm=vlm)) - chunks = [ - chunk.text - async for chunk in function.astream(VisionRequest(participant_id="alice", query="What is shown?")) - ] - answer = await function.ainvoke(VisionRequest(participant_id="alice", query="What is shown?")) - - assert chunks == ["No camera frame available — please try again."] - assert answer.text == "No camera frame available — please try again." - assert answer.status == "unavailable" - assert endpoint.statuses == [("processing", "alice"), ("idle", "alice")] - assert vlm.calls == [] - - # ── VisionToolsConfig — look_at_current_frame ───────────────────────────────── @@ -324,10 +262,6 @@ def test_historical_vision_request_requires_a_positive_reference_time() -> None: ) -def test_vision_request_rejects_unknown_arguments() -> None: - with pytest.raises(ValidationError): - VisionRequest(participant_id="alice", query="What is shown?", unsupported=True) - def test_live_vision_request_rejects_unknown_arguments() -> None: with pytest.raises(ValidationError):