xr_ai_models#

Unified service protocols and OpenAI-compatible clients for XR AI models.

Repository code talks to the typed *Service protocols. The concrete OpenAICompat* clients cover every in-tree backend and external OpenAI-compatible endpoints.

Submodules#

xr_ai_models.presets

Attributes#

`ContentPart`	A supported text, image, or video part in a chat message.
`ImageInput`	Image bytes, a filesystem path, a `data:` URL, or an HTTP(S) URL.
`VideoInput`	Video bytes, a filesystem path, a `data:` URL, or an HTTP(S) URL.
`Category`	A model role supported by `ModelsConfig`.
`KIND_OPENAI_COMPAT`	The adapter kind for OpenAI-compatible HTTP endpoints.
`ModelKind`	A supported model-service adapter implementation.
`Spec`	Any typed model-role specification stored in `ModelsConfig`.

Classes#

`Capabilities`	Features supported by a configured model endpoint.
`ChatMessage`	One input turn in an LLM or VLM conversation.
`ChatResponse`	A normalized non-streaming response from a chat model.
`EmbeddingService`	Structural interface for text-embedding services.
`ImagePart`	An image reference in a multimodal chat message.
`LLMService`	Structural interface for text chat-completion services.
`STTService`	Structural interface for speech-to-text services.
`TextPart`	Text content in a multimodal chat message.
`ToolCall`	A function-tool invocation requested by a model.
`ToolDef`	A function tool definition exposed to a chat model.
`TTSService`	Structural interface for text-to-speech services.
`VideoPart`	A video reference in a multimodal chat message.
`VLMService`	Structural interface for visual chat-completion services.
`OpenAICompatLLM`	OpenAI-compatible `/v1/chat/completions` client.
`OpenAICompatEmbedding`	OpenAI-compatible `/v1/embeddings` client.
`OpenAICompatSTT`	OpenAI-compatible `/v1/audio/transcriptions` client.
`OpenAICompatTTS`	OpenAI-compatible `/v1/audio/speech` client.
`OpenAICompatVLM`	OpenAI-compatible client for image- and video-bearing chat requests.
`AdapterSpec`	API dialect and model-specific request/response behavior.
`DeploymentSpec`	Process ownership for the endpoint that serves a model role.
`EmbeddingSpec`	Configuration for a text-embedding model role.
`EndpointSpec`	Connectivity, authentication, timeout, and readiness for an adapter.
`LLMSpec`	Configuration for a text chat-completion model role.
`ModelsConfig`	Logical-name to typed model-role specifications.
`STTSpec`	Configuration for a speech-to-text model role.
`TTSSpec`	Configuration for a text-to-speech model role.
`VLMSpec`	Configuration for an image- or video-language model role.

Functions#

`load_models_config`(→ ModelsConfig)	Load a YAML or JSON model profile and resolve its adapter presets.
`load_models_config_from_dict`(→ ModelsConfig)	Build a model configuration from a parsed direct or `models` mapping.
`make_embedding`(→ xr_ai_models.EmbeddingService)	Construct the embedding service for the named configuration entry.
`make_llm`(→ xr_ai_models.LLMService)	Construct the text chat service for the named configuration entry.
`make_stt`(→ xr_ai_models.STTService)	Construct the speech-to-text service for the named configuration entry.
`make_tts`(→ xr_ai_models.TTSService)	Construct the text-to-speech service for the named configuration entry.
`make_vlm`(→ xr_ai_models.VLMService)	Construct the visual chat service for the named configuration entry.

Package Contents#

class xr_ai_models.Capabilities#

Features supported by a configured model endpoint.

streaming: bool = True#: Whether the endpoint supports token streaming.

tool_calls: bool = False#: Whether the endpoint supports function-tool calls.

vision: bool = False#: Whether the endpoint accepts image inputs.

video: bool = False#: Whether the endpoint accepts video inputs.

reasoning: bool = False#: Whether the endpoint can return model reasoning.

class xr_ai_models.ChatMessage#

One input turn in an LLM or VLM conversation.

role: Literal['system', 'user', 'assistant', 'tool']#: The participant role associated with the message.

content: str | list[ContentPart]#: Plain text or ordered multimodal content supplied by the participant.

tool_calls: list[ToolCall] | None = None#: Tool calls emitted by an assistant turn, when present.

tool_call_id: str | None = None#: The call identifier answered by a tool-result message, when applicable.

class xr_ai_models.ChatResponse#

A normalized non-streaming response from a chat model.

content: str#: The assistant’s user-visible response text.

reasoning: str | None#: Normalized model reasoning, when the endpoint returns it.

tool_calls: list[ToolCall] | None#: Function-tool invocations requested by the model, when present.

finish_reason: str | None#: The provider’s reason for ending generation, when supplied.

raw: dict[str, Any]#: The unmodified provider response object.

xr_ai_models.ContentPart#: A supported text, image, or video part in a chat message.

class xr_ai_models.EmbeddingService#

Structural interface for text-embedding services.

async embed( texts: Sequence[str], *, timeout: float | None = None, ) → list[list[float]]#: Embed text strings, returning one vector for each input in order.

async health() → bool#: Return whether the configured endpoint is ready for requests.

async close() → None#: Release resources owned by the service.

xr_ai_models.ImageInput#: Image bytes, a filesystem path, a data: URL, or an HTTP(S) URL.

class xr_ai_models.ImagePart#

An image reference in a multimodal chat message.

url: str#: An HTTP(S) or data: URL containing the image.

type: Literal['image_url'] = 'image_url'#: The OpenAI-compatible discriminator for this content part.

class xr_ai_models.LLMService#

Structural interface for text chat-completion services.

capabilities: Capabilities#: Features supported by this service.

async chat( messages: Sequence[ChatMessage], *, tools: Sequence[ToolDef] | None = None, max_tokens: int | None = None, temperature: float | None = None, enable_thinking: bool = False, thinking_budget: int | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → ChatResponse#: Generate one complete response for a sequence of chat messages.

stream( messages: Sequence[ChatMessage], *, tools: Sequence[ToolDef] | None = None, max_tokens: int | None = None, temperature: float | None = None, enable_thinking: bool = False, thinking_budget: int | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → AsyncIterator[str]#: Stream user-visible response text for a sequence of chat messages.

async health() → bool#: Return whether the configured endpoint is ready for requests.

async close() → None#: Release resources owned by the service.

class xr_ai_models.STTService#

Structural interface for speech-to-text services.

async transcribe( audio: bytes, *, sample_rate: int | None = None, channels: int = 1, timeout: float | None = None, ) → str#: Transcribe WAV data or 16-bit PCM audio into text.

async health() → bool#: Return whether the configured endpoint is ready for requests.

async close() → None#: Release resources owned by the service.

class xr_ai_models.TextPart#

Text content in a multimodal chat message.

text: str#: The text presented to the model.

type: Literal['text'] = 'text'#: The OpenAI-compatible discriminator for this content part.

class xr_ai_models.ToolCall#

A function-tool invocation requested by a model.

id: str#: The provider-assigned identifier used to submit the tool result.

name: str#: The function name requested by the model.

arguments: str#: JSON-encoded arguments string, per the OpenAI tool-call contract.

class xr_ai_models.ToolDef#

A function tool definition exposed to a chat model.

name: str#: The function name the model uses in a ToolCall.

description: str#: A natural-language description of what the function does.

parameters: dict[str, Any]#: The function parameters as a JSON Schema object.

to_openai() → dict[str, Any]#: Return this definition in OpenAI’s function-tool wire format.

class xr_ai_models.TTSService#

Structural interface for text-to-speech services.

async synthesize( text: str, *, response_format: str = 'wav', timeout: float | None = None, ) → bytes#: Synthesize text and return audio in the requested format.

async health() → bool#: Return whether the configured endpoint is ready for requests.

async close() → None#: Release resources owned by the service.

xr_ai_models.VideoInput#: Video bytes, a filesystem path, a data: URL, or an HTTP(S) URL.

class xr_ai_models.VideoPart#

A video reference in a multimodal chat message.

url: str#: An HTTP(S) or data: URL containing the video.

type: Literal['video_url'] = 'video_url'#: The OpenAI-compatible discriminator for this content part.

class xr_ai_models.VLMService#

Structural interface for visual chat-completion services.

capabilities: Capabilities#: Features supported by this service.

async ask_image( image: ImageInput, question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → ChatResponse#: Generate one response to a question about an image.

async ask_images( images: Sequence[ImageInput], question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → ChatResponse#: Generate one response to a question about multiple images.

async ask_video( video: VideoInput, question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → ChatResponse#: Generate one response to a question about a video.

stream( image: ImageInput, question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → AsyncIterator[str]#: Stream response text for a question about an image.

stream_images( images: Sequence[ImageInput], question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → AsyncIterator[str]#: Stream response text for a question about multiple images.

async health() → bool#: Return whether the configured endpoint is ready for requests.

async close() → None#: Release resources owned by the service.

class xr_ai_models.OpenAICompatLLM( base_url: str, model_name: str, *, capabilities: xr_ai_models.Capabilities | None = None, reasoning_field: str | None = None, default_extras: dict[str, Any] | None = None, api_key_env: str | None = None, timeout: float = 60.0, health_check: bool = True, client: httpx.AsyncClient | None = None, )#

OpenAI-compatible /v1/chat/completions client.

Used directly for plain LLMs (Llama-Nemotron, Nemotron3-Nano, Nemotron-Omni) and indirectly via OpenAICompatVLM for VLMs. The API key is read once from api_key_env during construction. An injected HTTP client remains owned by the caller and is not closed here.

health_url#: The URL used to check endpoint readiness.

capabilities#: Features declared for this model endpoint.

async chat( messages: Sequence[xr_ai_models.ChatMessage], *, tools: Sequence[xr_ai_models.ToolDef] | None = None, max_tokens: int | None = None, temperature: float | None = None, enable_thinking: bool = False, thinking_budget: int | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → xr_ai_models.ChatResponse#

Generate and return one normalized chat response.

Per-call generation values override endpoint defaults. headers may supply request context but cannot override the configured authorization header.

async stream( messages: Sequence[xr_ai_models.ChatMessage], *, tools: Sequence[xr_ai_models.ToolDef] | None = None, max_tokens: int | None = None, temperature: float | None = None, enable_thinking: bool = False, thinking_budget: int | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → AsyncIterator[str]#

Stream user-visible text deltas from a chat completion.

Malformed server-sent event lines and deltas without content are skipped. Tool-call and reasoning deltas are not yielded.

async health() → bool#

Return whether the endpoint health check succeeds.

Returns True without an HTTP request when health checks were disabled at construction time.

async close() → None#: Close the internally created HTTP client, if one is owned.

class xr_ai_models.OpenAICompatEmbedding( base_url: str, model_name: str, *, api_key_env: str | None = None, timeout: float = 60.0, health_check: bool = True, client: httpx.AsyncClient | None = None, )#

OpenAI-compatible /v1/embeddings client.

The API key is read once from api_key_env during construction. An injected HTTP client remains owned by the caller and is not closed here.

health_url#: The URL used to check endpoint readiness.

async embed( texts: Sequence[str], *, timeout: float | None = None, ) → list[list[float]]#

Embed texts and return one vector per input in input order.

An empty input is handled locally. A response containing a different number of vectors raises ValueError.

async health() → bool#

Return whether the endpoint health check succeeds.

Returns True without an HTTP request when health checks were disabled at construction time.

async close() → None#: Close the internally created HTTP client, if one is owned.

class xr_ai_models.OpenAICompatSTT( base_url: str, *, api_key_env: str | None = None, timeout: float = 30.0, health_check: bool = True, client: httpx.AsyncClient | None = None, )#

OpenAI-compatible /v1/audio/transcriptions client.

The API key is read once from api_key_env during construction. An injected HTTP client remains owned by the caller and is not closed here.

health_url#: The URL used to check endpoint readiness.

async transcribe( audio: bytes, *, sample_rate: int | None = None, channels: int = 1, timeout: float | None = None, ) → str#

Transcribe WAV data or 16-bit PCM audio into text.

When sample_rate is provided, audio is interpreted as raw signed 16-bit PCM with the given channel count and wrapped in a WAV container. Otherwise, audio must already contain a server-supported audio file.

async health() → bool#

Return whether the endpoint health check succeeds.

Returns True without an HTTP request when health checks were disabled at construction time.

async close() → None#: Close the internally created HTTP client, if one is owned.

class xr_ai_models.OpenAICompatTTS( base_url: str, *, api_key_env: str | None = None, timeout: float = 30.0, health_check: bool = True, client: httpx.AsyncClient | None = None, )#

OpenAI-compatible /v1/audio/speech client.

The API key is read once from api_key_env during construction. An injected HTTP client remains owned by the caller and is not closed here.

health_url#: The URL used to check endpoint readiness.

async synthesize( text: str, *, response_format: str = 'wav', timeout: float | None = None, ) → bytes#: Synthesize text and return audio bytes in response_format.

async health() → bool#

Return whether the endpoint health check succeeds.

Returns True without an HTTP request when health checks were disabled at construction time.

async close() → None#: Close the internally created HTTP client, if one is owned.

class xr_ai_models.OpenAICompatVLM( base_url: str, model_name: str, *, capabilities: xr_ai_models.Capabilities | None = None, default_extras: dict[str, Any] | None = None, api_key_env: str | None = None, timeout: float = 60.0, health_check: bool = True, client: httpx.AsyncClient | None = None, )#

OpenAI-compatible client for image- and video-bearing chat requests.

Image and video paths or bytes are converted to data: URLs before submission. An injected HTTP client remains owned by the caller and is not closed here.

property capabilities: xr_ai_models.Capabilities#: Return the features declared for the underlying chat endpoint.

property health_url: str#: Return the URL used to check endpoint readiness.

async ask_image( image: xr_ai_models.ImageInput, question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → xr_ai_models.ChatResponse#: Generate one response to question about image.

async ask_images( images: Sequence[xr_ai_models.ImageInput], question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → xr_ai_models.ChatResponse#

Generate one response to question about an ordered image sequence.

Raises ValueError when images is empty.

async ask_video( video: xr_ai_models.VideoInput, question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → xr_ai_models.ChatResponse#

Generate one response to question about video.

Raises ValueError unless video support was declared in the configured capabilities.

async stream( image: xr_ai_models.ImageInput, question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → AsyncIterator[str]#: Stream response text for question about image.

async stream_images( images: Sequence[xr_ai_models.ImageInput], question: str, *, system_prompt: str = '', max_tokens: int | None = None, temperature: float | None = None, timeout: float | None = None, headers: Mapping[str, str] | None = None, ) → AsyncIterator[str]#

Stream response text for a question about an ordered image sequence.

Raises ValueError when images is empty.

async health() → bool#: Return whether the underlying chat endpoint is ready.

async close() → None#: Release resources owned by the underlying chat client.

class xr_ai_models.AdapterSpec#

API dialect and model-specific request/response behavior.

kind: ModelKind = 'openai_compat'#: The concrete client implementation used for this role.

model_name: str = ''#: The model identifier sent to endpoints that require one.

reasoning_field: str | None = None#: The response field containing reasoning, or None for auto-detection.

capabilities: dict[str, Any]#: Overrides used to construct the service’s capability flags.

default_extras: dict[str, Any]#: Model-specific fields merged into every request payload.

xr_ai_models.Category#: A model role supported by ModelsConfig.

class xr_ai_models.DeploymentSpec#

Process ownership for the endpoint that serves a model role.

ownership: Ownership = 'external'#: Whether the launcher starts, reuses, or does not manage the service.

service: str | None = None#: The launcher service name for managed or reused deployments.

class xr_ai_models.EmbeddingSpec( kind: ModelKind = KIND_OPENAI_COMPAT, base_url: str = '', model_name: str = '', api_key_env: str | None = None, timeout: float = 60.0, health_check: bool = True, deployment: DeploymentSpec | None = None, *, adapter: AdapterSpec | None = None, endpoint: EndpointSpec | None = None, )#

Configuration for a text-embedding model role.

adapter: AdapterSpec#: Model-specific request and response behavior.

endpoint: EndpointSpec#: Endpoint connectivity, authentication, and readiness settings.

deployment: DeploymentSpec#: Launcher ownership metadata for the serving process.

class xr_ai_models.EndpointSpec#

Connectivity, authentication, timeout, and readiness for an adapter.

base_url: str = ''#: The endpoint root URL, without a model-specific route.

api_key_env: str | None = None#: Environment variable containing the bearer token, when required.

timeout: float = 60.0#: Default request timeout in seconds.

readiness: Readiness = 'health'#: How the client determines whether the endpoint is ready.

property health_check: bool#: Whether readiness requires a successful endpoint health check.

xr_ai_models.KIND_OPENAI_COMPAT: ModelKind = 'openai_compat'#: The adapter kind for OpenAI-compatible HTTP endpoints.

class xr_ai_models.LLMSpec( kind: ModelKind = KIND_OPENAI_COMPAT, base_url: str = '', model_name: str = '', api_key_env: str | None = None, reasoning_field: str | None = None, capabilities: dict[str, Any] | None = None, default_extras: dict[str, Any] | None = None, timeout: float = 60.0, health_check: bool = True, deployment: DeploymentSpec | None = None, *, adapter: AdapterSpec | None = None, endpoint: EndpointSpec | None = None, )#

Configuration for a text chat-completion model role.

adapter: AdapterSpec#: Model-specific request and response behavior.

endpoint: EndpointSpec#: Endpoint connectivity, authentication, and readiness settings.

deployment: DeploymentSpec#: Launcher ownership metadata for the serving process.

xr_ai_models.ModelKind#: A supported model-service adapter implementation.

class xr_ai_models.ModelsConfig#

Logical-name to typed model-role specifications.

entries: dict[str, Spec]#: Model-role specifications keyed by application-defined logical name.

llm(name: str) → LLMSpec#: Return the LLM specification named name.

vlm(name: str) → VLMSpec#: Return the VLM specification named name.

stt(name: str) → STTSpec#: Return the speech-to-text specification named name.

tts(name: str) → TTSSpec#: Return the text-to-speech specification named name.

embedding(name: str) → EmbeddingSpec#: Return the embedding specification named name.

property required_credentials: tuple[str, Ellipsis]#: Return the sorted environment-variable names required by all roles.

xr_ai_models.Spec#: Any typed model-role specification stored in ModelsConfig.

class xr_ai_models.STTSpec( kind: ModelKind = KIND_OPENAI_COMPAT, base_url: str = '', api_key_env: str | None = None, timeout: float = 30.0, health_check: bool = True, deployment: DeploymentSpec | None = None, *, adapter: AdapterSpec | None = None, endpoint: EndpointSpec | None = None, )#

Configuration for a speech-to-text model role.

adapter: AdapterSpec#: Model-specific request and response behavior.

endpoint: EndpointSpec#: Endpoint connectivity, authentication, and readiness settings.

deployment: DeploymentSpec#: Launcher ownership metadata for the serving process.

class xr_ai_models.TTSSpec( kind: ModelKind = KIND_OPENAI_COMPAT, base_url: str = '', api_key_env: str | None = None, timeout: float = 30.0, health_check: bool = True, deployment: DeploymentSpec | None = None, *, adapter: AdapterSpec | None = None, endpoint: EndpointSpec | None = None, )#

Configuration for a text-to-speech model role.

adapter: AdapterSpec#: Model-specific request and response behavior.

endpoint: EndpointSpec#: Endpoint connectivity, authentication, and readiness settings.

deployment: DeploymentSpec#: Launcher ownership metadata for the serving process.

class xr_ai_models.VLMSpec( kind: ModelKind = KIND_OPENAI_COMPAT, base_url: str = '', model_name: str = '', api_key_env: str | None = None, capabilities: dict[str, Any] | None = None, default_extras: dict[str, Any] | None = None, timeout: float = 60.0, health_check: bool = True, deployment: DeploymentSpec | None = None, *, adapter: AdapterSpec | None = None, endpoint: EndpointSpec | None = None, )#

Configuration for an image- or video-language model role.

adapter: AdapterSpec#: Model-specific request and response behavior.

endpoint: EndpointSpec#: Endpoint connectivity, authentication, and readiness settings.

deployment: DeploymentSpec#: Launcher ownership metadata for the serving process.

xr_ai_models.load_models_config(path: pathlib.Path | str) → ModelsConfig#

Load a YAML or JSON model profile and resolve its adapter presets.

The file may contain model entries directly or nested below a models key. Invalid entries raise ValueError with the file path and role name included in the message.

xr_ai_models.load_models_config_from_dict( raw: dict[str, Any], *, source: str = '<dict>', ) → ModelsConfig#

Build a model configuration from a parsed direct or models mapping.

source labels validation errors and is useful when the mapping did not originate from a file.

xr_ai_models.make_embedding( config: xr_ai_models.ModelsConfig, name: str, ) → xr_ai_models.EmbeddingService#

Construct the embedding service for the named configuration entry.