# Model Context Protocol (MCP) Source: https://visionagents.ai/ai-technologies/model-context-protocol The [Model Context Protocol (MCP)](https://modelcontextprotocol.io/overview) is an open standard that defines how LLMs interact with external tools and services. Vision Agents supports MCP for function calling and connecting to external tool servers. ## How Vision Agents Uses MCP Vision Agents provides two ways to give your agent access to tools: 1. **Function Registration** — Decorate Python functions to make them callable by the LLM 2. **MCP Servers** — Connect to local or remote servers that expose tools Both approaches work with any LLM that supports function calling (OpenAI, Gemini, etc.) and with Realtime models. ## Function Registration Use `@llm.register_function()` to make any Python function available to your agent: ```python theme={null} from vision_agents.plugins import openai llm = openai.LLM(model="gpt-5.4") @llm.register_function(description="Get current weather for a location") async def get_weather(location: str) -> dict: return {"location": location, "temperature": "22°C", "condition": "Sunny"} # The LLM will automatically call get_weather() when relevant response = await llm.simple_response("What's the weather like in London?") ``` ## MCP Servers Connect to MCP servers for access to external services: ```python theme={null} from vision_agents.core.mcp import MCPServerRemote, MCPServerLocal from vision_agents.core.agents import Agent # Remote server (HTTP) github_server = MCPServerRemote( url="https://api.githubcopilot.com/mcp/", headers={"Authorization": f"Bearer {token}"} ) # Local server (stdio) local_server = MCPServerLocal(command="uv run my_mcp_server.py") # Connect to your agent agent = Agent( edge=getstream.Edge(), llm=openai.LLM(model="gpt-5.4"), agent_user=User(name="Assistant", id="agent"), mcp_servers=[github_server, local_server] ) ``` When you add MCP servers to an agent, tools are automatically discovered and registered with the LLM. ## When to Use Each Approach | Approach | Best For | | ------------------------- | --------------------------------------------------- | | **Function Registration** | Simple tools, inline logic, prototyping | | **MCP Servers** | External APIs, shared tooling, complex integrations | You can combine both — register local functions while also connecting MCP servers. For detailed examples including parameter types, custom function names, and real-world patterns, see the [MCP and Function Calling Guide](/guides/mcp-tool-calling). # Speech To Speech (STS) Source: https://visionagents.ai/ai-technologies/speech-to-speech Speech-to-Speech (STS), also known as **Realtime** in the SDK, is a single provider that handles listening, reasoning, and speaking over one connection. Realtime models include built-in [STT](/ai-technologies/speech-to-text), [TTS](/ai-technologies/text-to-speech), and [turn detection](/ai-technologies/turn-detection) — you configure `llm=.Realtime()` without separate speech plugins. ```mermaid theme={null} flowchart TB Participants[Call participants] --> Edge[Edge transport] Edge --> Agent[Agent RealtimeInferenceFlow] Agent --> Realtime[Realtime API] Realtime --> Agent Agent --> Edge Edge --> Participants ``` You still need `edge`, `agent_user`, `instructions`, API keys, and call lifecycle setup. See [Voice Agents](/introduction/voice-agents) for the full guide. ## How It Works When the LLM is a Realtime model, the Agent selects `RealtimeInferenceFlow`: * Call audio is sent to the Realtime API * Model audio and transcripts are published back to the call * STT, TTS, and turn detection plugins are automatically disabled ## Quick Start ```python theme={null} from vision_agents.core import Agent, User from vision_agents.plugins import gemini, getstream agent = Agent( edge=getstream.Edge(), agent_user=User(name="Assistant", id="agent"), instructions="You are a helpful voice assistant.", llm=gemini.Realtime(), # no stt/tts/turn_detection needed ) ``` ## Realtime vs Custom Pipeline | Mode | Best for | You choose | | ------------------- | ---------------------------- | ---------------------------------- | | **Realtime** | Fastest path, lowest latency | One provider for speech in and out | | **Custom Pipeline** | Full control over each stage | STT, LLM, and TTS independently | Use a [custom pipeline](/introduction/voice-agents#custom-pipeline-mode) when you want to mix providers — for example Deepgram STT, Gemini LLM, and Inworld TTS. ## Providers | Provider | Integration page | | ----------- | ------------------------------------------------- | | OpenAI | [OpenAI Realtime](/integrations/realtime/openai) | | Gemini | [Gemini Live](/integrations/realtime/gemini) | | Qwen | [Qwen OMNI](/integrations/realtime/qwen) | | AWS Bedrock | [AWS Bedrock](/integrations/realtime/aws-bedrock) | | xAI | [xAI](/integrations/realtime/xai) | | Inworld | [Inworld](/integrations/realtime/inworld) | ## Next Steps Realtime setup and provider swapping Barge-in and turn timing Tools with realtime models Mix STT, LLM, and TTS providers # Speech To Text (STT) Source: https://visionagents.ai/ai-technologies/speech-to-text Speech-to-Text (STT) converts spoken audio into text. In a custom voice pipeline, STT sits between the call and the LLM: user speaks → **STT** transcribes → turn signal → **LLM** → **TTS**. See [Voice Agents](/introduction/voice-agents#custom-pipeline-mode) for the full wiring guide. ```mermaid theme={null} flowchart LR Audio[Spoken audio] --> STT[STT plugin] STT --> Text[Transcript text] ``` ## Quick Start ```bash theme={null} uv add "vision-agents[deepgram,inworld,getstream]" ``` Add keys to your `.env`: ```bash theme={null} STREAM_API_KEY=... STREAM_API_SECRET=... DEEPGRAM_API_KEY=... ``` ```python theme={null} from dotenv import load_dotenv from vision_agents.core import Agent, User from vision_agents.plugins import deepgram, gemini, getstream, inworld load_dotenv() agent = Agent( edge=getstream.Edge(), agent_user=User(name="Assistant", id="agent"), instructions="You are a helpful voice assistant.", llm=gemini.LLM(), stt=deepgram.STT(eager_turn_detection=True), tts=inworld.TTS(), ) ``` The SDK captures audio from your Stream call and feeds it to the STT provider in real time. As people speak, you receive live transcripts. ```mermaid theme={null} flowchart TB Participants[Call participants] --> Edge[Edge transport] Edge --> Agent[Agent audio input] Agent --> STT[STT plugin] STT --> Partial[Partial transcripts] STT --> Final[Final transcripts] STT --> Turns[TurnStarted / TurnEnded] Final --> Events[UserTranscriptEvent] Turns --> Pipeline[LLM turn timing] ``` ## Providers | Plugin | Turn detection | Latency model | Integration | | ------------ | -------------------- | ------------------------- | ---------------------------------------------- | | Deepgram | Yes (eager optional) | Streaming | [Deepgram](/integrations/stt/deepgram) | | ElevenLabs | Yes (VAD commit) | Streaming | [ElevenLabs](/integrations/stt/elevenlabs) | | Cartesia | Yes (eager default) | Streaming | [Cartesia](/integrations/stt/cartesia) | | AssemblyAI | Yes | Streaming | [AssemblyAI](/integrations/stt/assemblyai) | | Sarvam | Yes | Streaming | [Sarvam](/integrations/stt/sarvam) | | Fast-Whisper | No | Local batch (\~2s buffer) | [Fast-Whisper](/integrations/stt/fast-whisper) | | Wizper | No | Cloud (Fal.ai) | [Wizper](/integrations/stt/wizper) | | Fish | No | Batch | [Fish](/integrations/stt/fish) | | Mistral | No | Streaming | [Mistral](/integrations/stt/mistral) | **Fast-Whisper** runs locally on your machine. **Wizper** uses Fal.ai cloud — it is not a local model. STT plugins without built-in turn detection need an external [turn detection](/ai-technologies/turn-detection) plugin, or the agent falls back to using the final transcript as the end-of-turn signal. ## Transcripts STT providers emit partial, replacement, and final transcripts. Subscribe to agent events to handle them in your application: ```python theme={null} from vision_agents.core.agents.events import UserTranscriptEvent @agent.events.subscribe async def on_transcript(event: UserTranscriptEvent): print(event.text) ``` ## Next Steps Wire STT into a full voice pipeline When the agent starts and stops listening # Text To Speech (TTS) Source: https://visionagents.ai/ai-technologies/text-to-speech Text-to-Speech (TTS) transforms written words into spoken audio, allowing your applications to "speak" to users naturally. With the Vision Agents, you can easily add voice capabilities to your video calls and applications, creating experiences where text becomes lifelike speech in real-time. ## How does text to speech work? TTS converts text to PCM audio. In a voice agent, text usually comes from the LLM response; you can also call `agent.say()` to speak text directly without the LLM. ```mermaid theme={null} flowchart LR Text[Input text] --> TTS[TTS plugin] TTS --> Audio[PCM audio] ``` ## Using Text-To-Speech with Stream The Vision Agents SDK routes synthesized audio from the TTS plugin into your Stream call via the agent's outbound audio track. ```mermaid theme={null} flowchart TB LLM[LLM response text] --> TTS[TTS plugin] Say[agent.say text] --> TTS TTS --> Track[Agent audio track] Track --> Edge[Edge transport] Edge --> Participants[Call participants] ``` ## Worked example Let's walk through a real-world example to see how TTS works in your application. Imagine you're building a customer support system where callers get placed in a queue. Here's how TTS makes this experience feel personal and professional: **The Scenario**: A customer calls your support line and gets placed in a queue. **What Happens**: 1. Your system detects the caller and generates a friendly message: "Thank you for calling TechCorp Support. Your estimated wait time is 5 minutes." 2. Instead of showing this as text on screen (which the caller can't see), your TTS plugin converts it to natural speech that sounds like a real person. 3. The voice speaks directly to the caller through the Stream call, creating an immediate human connection. 4. As the queue updates, new messages are automatically spoken: "Your wait time is now 3 minutes" or "We're connecting you to an agent now." **The Result**: Instead of a silent, frustrating wait, customers get a conversational experience that feels like they're being personally attended to, even when they're waiting in line. # Turn Detection Source: https://visionagents.ai/ai-technologies/turn-detection Turn Detection identifies when a speaker has finished their conversational turn and it's appropriate for an AI to respond. It solves a critical problem in voice AI: respond too early and you interrupt the speaker; wait too long and the conversation feels awkward. ## Two Paths in Vision Agents Turn detection works differently depending on your setup: ``` Custom pipeline: Call audio → STT or TurnDetector → turn signals → LLM Realtime mode: Call audio ↔ Realtime API ``` ### 1. Provider-built-in (STT output stream) Many STT plugins emit `TurnStarted` and `TurnEnded` signals on their output stream. Turn detection happens inside the provider API — no separate plugin needed. ```python theme={null} from vision_agents.core import Agent, User from vision_agents.plugins import deepgram, gemini, getstream, inworld agent = Agent( edge=getstream.Edge(), agent_user=User(name="Assistant", id="agent"), instructions="You are a helpful voice assistant.", llm=gemini.LLM(), stt=deepgram.STT(eager_turn_detection=True), # built-in turns; eager = lower latency tts=inworld.TTS(), ) ``` ### 2. External plugin When your STT plugin has no built-in turn detection, add a `TurnDetector` plugin: ```python theme={null} from vision_agents.plugins import smart_turn agent = Agent( ..., stt=fast_whisper.STT(), turn_detection=smart_turn.TurnDetection(), ) ``` External plugins use Silero VAD plus neural models to predict turn completion — see [Smart Turn](/integrations/turn-detection/smart-turn) and [Vogent](/integrations/turn-detection/vogent). ## Turn Detection vs VAD | | VAD | Turn Detection | | ---------------- | --------------------------- | ---------------------------------------- | | **Question** | "Is someone speaking?" | "Has the speaker finished?" | | **Output** | Speech start/end timestamps | `TurnStarted` / `TurnEnded` turn signals | | **Intelligence** | Simple audio analysis | Provider API or neural turn models | | **Best for** | Detecting presence | Knowing when to respond | ## STT Plugins with Built-in Turn Detection | STT Plugin | Turn Detection | | ------------------------------------------ | ---------------------------------------------------------------------- | | [Deepgram](/integrations/stt/deepgram) | Always on; set `eager_turn_detection=True` for speculative early turns | | [ElevenLabs](/integrations/stt/elevenlabs) | Built-in via VAD commit strategy | | [Cartesia](/integrations/stt/cartesia) | Built-in; eager by default | | [AssemblyAI](/integrations/stt/assemblyai) | Built-in | | [Sarvam](/integrations/stt/sarvam) | Built-in via VAD events | STT plugins without turn detection (Fast-Whisper, Wizper, Fish, Mistral, AWS) need an external plugin or fall back to using the final transcript as the end-of-turn signal. ## External Turn Detection Plugins | Plugin | Description | | ----------------------------------------------------- | ------------------------------------------------------------------------ | | [Smart Turn](/integrations/turn-detection/smart-turn) | Combines Silero VAD, Whisper features, and neural turn completion models | | [Vogent](/integrations/turn-detection/vogent) | Neural turn detection with high accuracy prediction | For Realtime APIs (OpenAI, Gemini, AWS Bedrock, Qwen, xAI, Inworld), turn detection is built-in at the model level — no separate plugin needed. See [Voice Agents — Realtime Mode](/introduction/voice-agents#realtime-mode). When an STT plugin provides built-in turn detection (`stt.turn_detection` is `True`), the `Agent` automatically ignores any external `TurnDetector` plugin to prevent conflicts. ## SDK Behavior * **`TurnStarted`** — triggers barge-in interrupt in the transcribing flow * **`TurnEnded(eager=True)`** — starts speculative LLM work for lower latency * **`TurnEnded(eager=False)`** — confirms the turn is complete * Pipeline signals (`TurnStarted`/`TurnEnded` on the STT output stream) are distinct from agent events (`UserTurnStartedEvent`, `UserTurnEndedEvent`) ## Next Steps Setup, tuning, and troubleshooting Wire turn detection into a pipeline Configure the Smart Turn plugin Alternative turn detection option # Agent Class Source: https://visionagents.ai/core/agent-core The `Agent` class is the central orchestrator that brings together all other components in the Vision Agents framework. It manages the conversation flow, handles real-time audio/video processing, coordinates responses, and integrates with external tools via MCP (Model Context Protocol) servers. It is the main interface for building AI-powered video and voice applications. It supports both traditional STT/TTS workflows and modern realtime speech-to-speech models, making it flexible for various use cases. Do not reuse an `Agent` instance. Create a new agent for each call. Calling `join()` twice on the same agent raises `RuntimeError`. ```python theme={null} from vision_agents.core import Agent, User from vision_agents.plugins import openai, deepgram, elevenlabs, getstream # Traditional STT/TTS mode agent = Agent( edge=getstream.Edge(), agent_user=User(name="AI Assistant", id="agent"), instructions="You're a helpful AI assistant", llm=openai.LLM(model="gpt-5.4"), stt=deepgram.STT(), tts=elevenlabs.TTS(), ) # Realtime mode agent = Agent( edge=getstream.Edge(), agent_user=User(name="AI Assistant", id="agent"), instructions="You're a helpful AI assistant", llm=openai.Realtime(model="gpt-realtime", voice="marin"), ) ``` ## Constructor Parameters * **`edge`** (`EdgeTransport`): The edge network provider for video & audio transport (you can choose any provider here) * **`llm`** (`LLM | AudioLLM | VideoLLM`): The language model — can be a text LLM, audio-capable (AudioLLM), video-capable (VideoLLM), or a combined Realtime model * **`agent_user`** (`User`): The agent's user information (name, id, etc.) **Optional Parameters** * **`instructions`** (`str`): System instructions for the agent. Supports `@file.md` references to load instructions from markdown files (default: "Keep your replies short and dont use special characters.") * **`stt`** (`Optional[STT]`): Speech-to-text service (automatically disabled in realtime mode) * **`tts`** (`Optional[TTS]`): Text-to-speech service (automatically disabled in realtime mode) * **`turn_detection`** (`Optional[TurnDetector]`): Turn detection service for managing conversation turns. Automatically disabled in realtime mode. Ignored automatically if the configured STT plugin already provides built-in turn detection (when `stt.turn_detection` is `True`, e.g. `elevenlabs.STT`, `deepgram.STT`) * **`processors`** (`Optional[List[Processor]]`): List of processors for video/audio processing * **`avatar`** (`Optional[Avatar]`): Avatar provider for lip-synced video output (for example `anam.Avatar(...)`) * **`mcp_servers`** (`Optional[List[MCPBaseServer]]`): MCP servers for external tool access * **`options`** (`Optional[AgentOptions]`): Configuration options — see [AgentOptions](#agentoptions) below * **`broadcast_metrics`** (`bool`): Broadcast agent metrics to call participants via custom events (default: `False`) * **`broadcast_metrics_interval`** (`float`): Interval in seconds between metrics broadcasts (default: `5.0`) * **`multi_speaker_filter`** (`Optional[AudioFilter]`): Filter for multi-speaker audio routing. Defaults to `FirstSpeakerWinsFilter`, which locks onto the first active speaker and drops other participants' audio until silence timeout or disconnect. Only activates when two or more participants have active audio tracks — see [Multiple Speakers](/guides/multiple-speakers) * **`tracer`** (`Tracer`): OpenTelemetry tracer for distributed tracing (default: `trace.get_tracer("agents")`) * **`profiler`** (`Optional[Profiler]`): Performance profiler for the agent and its plugins ### AgentOptions | Field | Type | Default | Description | | ----------- | ----- | --------------------- | ---------------------------------------------------------------- | | `model_dir` | `str` | System temp directory | Directory for downloaded model files (VAD, turn detection, etc.) | ### Mode Validation When an `AudioLLM` is configured (currently Realtime models), STT, TTS, and turn detection are automatically disabled with a warning log. In non-realtime mode, the agent requires at least an audio processing path (STT, TTS, or turn detection) or video processors. Video-only agents without an LLM are allowed when video processors are configured. ## Core Lifecycle Methods **`async create_call(call_type: str, call_id: str) -> Call`** Creates a call on the edge provider. Calls `authenticate()` automatically. **`async join(call: Call, participant_wait_timeout: Optional[float] = 10.0) -> AsyncIterator[None]`** Joins a video call. Must be called as an async context manager. The agent can join the call only once. Once the call ends, the agent closes itself. `join()` always calls `close()` in its `finally` block. **Parameters** * **`call`** (`Call`): the call to join. * **`participant_wait_timeout`** (`Optional[float]`): timeout in seconds to wait for other participants to join before proceeding.\ If `0`, do not wait at all. If `None`, wait forever.\ Default - `10.0`. Delegates to `wait_for_participant()`. ```python theme={null} call = await agent.create_call(call_type="default", call_id="my-call") async with agent.join(call): await agent.simple_response("Say hi.") await agent.finish() # Wait for call to end ``` **`async wait_for_participant(timeout: Optional[float] = None) -> None`** Waits for at least one participant to join. Default `None` waits forever. **`async finish()`** Waits for the call to end gracefully. Subscribes to the call ended event. Returns immediately if not joined or already closed. **`async close()`** Cleans up all connections and resources. Safe to call multiple times. Called automatically when `join()` context exits. **`async authenticate()`** Authenticates the agent user with the edge provider. Idempotent — safe to call multiple times. Called automatically from `join()` and `create_call()`, so you usually don't need to call it directly. ### Response Methods **`async simple_response(text: str, participant: Optional[Participant] = None, interrupt: bool = True)`** Sends a text prompt to the active inference flow. The LLM generates a response which is then spoken through TTS (or realtime audio output). Use `interrupt=True` when you want this request to preempt an in-flight response. ```python theme={null} await agent.simple_response("Hello, how can I help you?", interrupt=True) ``` **`async say(text: str, interrupt: bool = False)`** Makes the agent speak text directly, bypassing the LLM. Works in transcribing (STT + LLM + TTS) mode only — the text is synthesized via TTS. In Realtime mode, `say()` logs a warning and does not produce audio (conversation history may still be updated). Use `interrupt=True` to stop current output before speaking. ```python theme={null} await agent.say("Welcome to the call!", interrupt=False) ``` ### Monitoring Methods **`idle_for() -> float`** Returns the number of seconds the agent has been alone on the call (all human participants have left). Returns `0.0` while other participants are present or before join. **`on_call_for() -> float`** Returns the number of seconds since the agent joined the call. ### Observability **`metrics`** — `AgentMetrics` property for performance data. See [Telemetry](/core/telemetry). **`async send_metrics_event(event_type="agent_metrics", fields=None)`** — Broadcast metrics to call participants. **`async send_custom_event(data: dict)`** — Push custom data to call participants via the edge provider. ## MCP Integration The Agent supports Model Context Protocol (MCP) for external tool integration. MCP servers connect when the agent joins a call: ```python theme={null} from vision_agents.core import Agent, User from vision_agents.core.mcp import MCPServerRemote github_server = MCPServerRemote( url="https://api.githubcopilot.com/mcp/", headers={"Authorization": f"Bearer {github_pat}"} ) agent = Agent( edge=getstream.Edge(), agent_user=User(name="Assistant", id="agent"), llm=..., mcp_servers=[github_server], ) ``` MCP tools are automatically registered with the LLM's function registry and can be called during conversations. See the [MCP guide](/guides/mcp-tool-calling) for more. ## Event System The `Agent` makes it easy for developers to quickly subscribe and listen to events happening across all components. The event system merges all events across the plugin and core allowing you to listen to events in a single place using their respective type. ### Core Events * **Agent Lifecycle Events**: `UserTurnStartedEvent`, `UserTurnEndedEvent`, `UserTranscriptEvent`, `AgentTurnStartedEvent`, `AgentTurnEndedEvent`, `AgentJoinedCallEvent`, `AgentLeftCallEvent`, `AgentFinishEvent` (emitted only when `finish()` is cancelled, not on normal call end) * **Edge / Call Events**: `ParticipantJoinedEvent`, `ParticipantLeftEvent`, `CallEndedEvent`, `TrackAddedEvent`, `TrackRemovedEvent`, `AudioReceivedEvent` * **LLM Events**: `LLMResponseFinalEvent`, `LLMErrorEvent` * **Tool Events**: `ToolStartEvent`, `ToolEndEvent` * **Realtime Events**: `RealtimeConnectedEvent`, `RealtimeDisconnectedEvent` * **STT Events**: `STTConnectedEvent`, `STTDisconnectedEvent`, `STTErrorEvent` * **TTS Events**: `TTSSynthesisStartEvent`, `TTSSynthesisCompleteEvent`, `TTSConnectedEvent`, `TTSDisconnectedEvent`, `TTSErrorEvent` See [Events Reference](/reference/events-reference) for full field details. ### Event Subscription ```python theme={null} from vision_agents.core.edge.events import AudioReceivedEvent @agent.events.subscribe async def on_audio_received(event: AudioReceivedEvent): # Handle audio data pass ``` You can also use `agent.subscribe(function)` as an alternative to the decorator. ## Debugging with local video files For testing and debugging video processing without a live camera, you can use a local video file as the video source. This is useful for reproducible testing and development. Video override only works when `publish_video` is `True` (requires an avatar or a video publisher processor). Set the path **before** calling `join()`. ### Using the CLI Pass the `--video-track-override` option when running your agent: ```bash theme={null} uv run agent.py run --video-track-override=/path/to/video.mp4 ``` ### Using the API ```python theme={null} from vision_agents.core import Agent, User agent = Agent(...) agent.set_video_track_override_path("/path/to/video.mp4") ``` When a video override is set, the local video file plays in a loop at 30 FPS instead of any incoming video tracks from call participants. The track lifecycle remains intact (starts when a user joins, stops when they leave). # Avatar Class Source: https://visionagents.ai/core/avatar-core Avatars consume the agent's audio output and produce a synced video and audio feed of a virtual character. They run in passthrough mode: the avatar owns the agent's outbound video and audio tracks, and its output never feeds back into the LLM or any video processors. ## Class Hierarchy The `vision_agents.core.avatars` module exports two classes: | Class | Purpose | | ---------------- | ------------------------------------------------------------------------------------------------------------------------------- | | `Avatar` | Abstract base class; consumes the agent's audio output and publishes synced video and audio. | | `AVSynchronizer` | Utility that owns paired audio/video tracks and delays video frames to match the audio buffer depth, keeping lip-sync accurate. | All three built-in implementations (LiveAvatar, Anam, LemonSlice) build on `AVSynchronizer` for output, so it's the recommended building block for custom avatars too. ## Lifecycle The agent drives the avatar through a fixed lifecycle: 1. **`Agent.__init__`** queries `video_output()` and calls `attach_audio_input(stream)`, handing the avatar the inference flow's audio output stream. 2. **`Agent.join()`** calls `await avatar.start()`, which opens the provider connection and begins consuming the input stream. 3. **While running**, the avatar drains `input_audio_stream`, forwards audio to the provider, and exposes lip-synced video via `video_output()` and audio via `audio_output()`. 4. **`Agent.close()`** calls `await avatar.close()` for teardown. When an avatar is set, the agent publishes `avatar.audio_output()` as outbound audio instead of the TTS stream directly — TTS still synthesises, the avatar lip-syncs and republishes. ## Abstract Methods Subclasses must implement all four: | Method | Description | | ---------------- | ---------------------------------------------------------------------- | | `video_output()` | Return the outbound `aiortc.VideoStreamTrack` published to the call. | | `audio_output()` | Return the outbound `AudioOutputStream` published to the call. | | `async start()` | Open the provider connection and begin consuming `input_audio_stream`. | | `async close()` | Tear down the provider connection and cancel any consumer tasks. | Subclasses may also implement an `interrupt()` method to stop the in-flight utterance at the provider during barge-in. ## Properties & Helpers | Member | Description | | ---------------------------- | ----------------------------------------------------------------------------------------------------------------- | | `provider_name` | Class attribute identifying the provider (used in events and metrics). | | `events` | `EventManager` for emitting avatar-specific events. | | `metrics` | `MetricsCollector` for recording avatar metrics. | | `input_audio_stream` | The agent's audio output stream attached via `attach_audio_input`. Raises `ValueError` if accessed before attach. | | `attach_audio_input(stream)` | Called by the agent to hand off its audio output stream. Override to customise how audio is consumed. | ## AVSynchronizer `AVSynchronizer` is a utility class that solves the lip-sync problem: provider video and audio arrive on separate streams, and pushing them straight onto the outbound WebRTC tracks usually drifts. It owns a paired `audio_output` and `video_output`, delays each video frame by the current audio buffer depth, and paces frames at the configured fps (overriding aiortc's hardcoded 30 fps). ```python theme={null} from vision_agents.core.avatars import AVSynchronizer sync = AVSynchronizer( width=1920, height=1080, fps=30, max_queue_size=30, # typically int(fps * buffer_seconds) ) ``` | Member | Description | | -------------------------- | -------------------------------------------------------------------------------------- | | `video_output` | The `QueuedVideoTrack` to expose from `Avatar.video_output()`. | | `audio_output` | The `AudioOutputStream` to expose from `Avatar.audio_output()`. | | `async write_video(frame)` | Queue an `av.VideoFrame` from the provider, delayed by the current audio buffer depth. | | `async write_audio(pcm)` | Write a `PcmData` chunk from the provider to the audio track. | | `async flush()` | Discard pending video frames and flush buffered audio (use on interrupt). | | `close()` | Close the underlying audio stream. | ## Building a Custom Avatar A minimal subclass wraps an `AVSynchronizer`, exposes its tracks, and pumps provider frames into it from a consumer task started in `start()`: ```python theme={null} import asyncio from vision_agents.core.avatars import Avatar, AVSynchronizer from vision_agents.core.agents.inference import AudioOutputStream from getstream.video.rtc.track_util import PcmData import av class MyAvatar(Avatar): provider_name = "my_avatar" def __init__(self, width: int = 1280, height: int = 720, fps: int = 30) -> None: super().__init__() self._sync = AVSynchronizer(width=width, height=height, fps=fps) self._task: asyncio.Task | None = None def video_output(self): return self._sync.video_output def audio_output(self) -> AudioOutputStream: return self._sync.audio_output async def start(self) -> None: # open provider connection, then pump agent audio into it self._task = asyncio.create_task(self._consume(self.input_audio_stream)) async def close(self) -> None: if self._task: self._task.cancel() self._sync.close() async def _consume(self, stream: AudioOutputStream) -> None: async for chunk in stream: # send chunk.data to the provider; for each response frame: # await self._sync.write_video(frame) # av.VideoFrame # await self._sync.write_audio(pcm) # PcmData ... ``` ## Usage Pass an avatar to the agent at initialisation: ```python theme={null} from vision_agents.core import Agent, User from vision_agents.plugins import deepgram, gemini, getstream, liveavatar agent = Agent( edge=getstream.Edge(), agent_user=User(name="Assistant", id="agent"), llm=gemini.LLM("gemini-3-flash-preview"), tts=deepgram.TTS(), stt=deepgram.STT(), avatar=liveavatar.Avatar(), ) ``` ## Available Implementations Real-time interactive avatars by HeyGen with WebSocket lip-sync. Anam's avatar SDK with configurable dimensions and frame rate. LemonSlice avatars delivered over LiveKit. # LLM Class Source: https://visionagents.ai/core/llm-core The LLM component handles text generation and conversation logic. It supports both traditional request-response patterns and real-time streaming. The base interface provides `simple_response()` for generating responses from text input, includes function calling capabilities with automatic tool execution, and manages conversation context. Multiple providers are supported including OpenAI, Anthropic, Google, and others. ## Class Hierarchy The LLM system uses a class hierarchy to support different modalities: | Class | Description | | ---------- | ------------------------------------------------------------------------------------------------------------------------ | | `LLM` | Abstract base class for text-only language models | | `AudioLLM` | Extends `LLM` with `simple_audio_response()` for speech-to-speech processing. Does not require separate STT/TTS services | | `VideoLLM` | Extends `LLM` with `watch_video_track()` / `stop_watching_video_track()` for consuming video input | | `OmniLLM` | Combines `AudioLLM` + `VideoLLM` for models supporting both audio and video | | `Realtime` | Extends `OmniLLM` for realtime streaming models (e.g., OpenAI Realtime, Gemini Live) with persistent connections | ```python theme={null} # Traditional mode — text LLM with separate STT/TTS agent = Agent( llm=openai.LLM(model="gpt-5.4"), stt=deepgram.STT(), tts=elevenlabs.TTS() ) # Realtime mode — handles speech natively, no STT/TTS needed agent = Agent( llm=openai.Realtime() ) ``` Each LLM follows our philosophy of "thin wrapping". Out of the box, developers can pass in their own client to the LLMs or interact with the native APIs directly including full support for passing native method args. The LLMs can be combined with other features such as processors to provide realtime feedback on the world around you. It can also be used in a simple voice only mode as shown in the previous example. For models running in the non-realtime mode, a TTS service and STT service must be provided. These models will convert the user's speech to text which is then passed to the model. The model response is then converted into a voice output. ## Function Calling The LLM base class includes a built-in function registry for tool use. Register functions as tools that the LLM can call during conversations: ```python theme={null} @agent.llm.register_function( name="get_weather", description="Get the current weather for a location" ) async def get_weather(location: str) -> str: return f"The weather in {location} is sunny, 72°F" ``` Registered functions are automatically provided to the LLM as available tools. When the LLM decides to call a tool, the framework executes it with concurrency control (max 8 concurrent calls) and a per-tool timeout (default 30s), then feeds the result back to the LLM. MCP servers registered via the `mcp_servers` parameter on the Agent also register their tools through the same function registry. ## Chat Completions API Support Many open-source models follow the OpenAI Chat Completions API format. Whether you're experimenting with Kimi, Deepseek or Mistral, they call can be accessed by changing the base API url of the OpenAI SDK and setting an API key obtained from their respective dashboards. To support this, Vision Agent's ships with support for both the OpenAI Response API (used by GPT 5 and the default) as well as the Chat Completions API with streaming. To use, you must have the OpenAI plugin installed in your project. **Example** ```python theme={null} from vision_agents.plugins import deepgram, elevenlabs, getstream, openai async def create_agent(**kwargs) -> Agent: # Initialize the Baseten VLM llm = openai.ChatCompletionsVLM( api_key="API-KEY", model="qwen3vl", base_url="https://model-vq0nkx7w.api.baseten.co/development/sync/v1", # Replace with your model's hosted URL ) # Create an agent with video understanding capabilities agent = Agent( edge=getstream.Edge(), agent_user=User(name="Video Assistant", id="agent"), instructions="You're a helpful video AI assistant. Analyze the video frames and respond to user questions about what you see.", llm=llm, stt=deepgram.STT(), tts=elevenlabs.TTS(), processors=[], ) return agent ``` We offer both `ChatCompletionsLLM` and `ChatCompletionsVLM` interfaces. The VLM interface will automatically forward the user's video feed as frames to the model. The above example demonstrates this using Qwen3-VL running on Baseten. ## VLM Support Models such as [Moondream](/integrations/vision/moondream), [TwelveLabs Pegasus](/integrations/vision/twelvelabs), Qwen 3 and others offer powerful APIs for visual reasoning and understanding. These models operate as a subset of `LLM` called VLM. The frames from the user's video feeds is buffered and sent to the model at a specified interval. Each VLM is unique so be sure to check the docs and model capabilities of each but generally, each VLM also requires an STT provider and in some cases an TTS provider to vocalise the response (some models like Qwen OMNI has TTS built in). ```python theme={null} llm = moondream.CloudVLM( api_key="your-api-key", # or set MOONDREAM_API_KEY env var mode="vqa", # or "caption" for image captioning ) agent = Agent( edge=getstream.Edge(), agent_user=User(name="Vision Assistant", id="agent"), llm=llm, tts=elevenlabs.TTS(), stt=deepgram.STT(), ) ``` # Overview Source: https://visionagents.ai/core/overview Modular architecture for real-time voice and video AI agents New to Vision Agents? Start with the [Overview](/introduction/overview) and [Quickstart](/introduction/quickstart). This section is API reference — read it after your first agent runs. The SDK is built around a modular architecture where each component can be independently configured and extended. At its heart is the `Agent` class, which orchestrates all other components to create intelligent, interactive experiences. The Agent class is transport agnostic, meaning developers can use Stream's low-latency Video transport by default or bring custom transports. Framework Overview ### Design Philosophy * **Modularity**: Each component (LLM, STT, TTS, Turn Detection, Processors) can be independently configured * **Provider Agnostic**: Support for multiple AI providers through standardised interfaces * **Event-Driven**: Built on an event system for loose coupling and extensibility * **Low Latency**: Optimized for real-time voice and video interactions * **Type Safety**: Rich typing throughout for better developer experience # Processors Class Source: https://visionagents.ai/core/processors-core Processors extend the agent's capabilities by analyzing and transforming audio/video streams in real-time. They can access incoming media, publish transformed streams back to the call, and (in plugin implementations) feed detection results or external data to the LLM. Common use cases: * **Video analysis** — Pose detection, object recognition, scene understanding * **Media transformation** — Video effects, avatars, filters * **State injection** — Plugin-specific patterns for feeding detection results to the LLM (not automatic in core) ## Class Hierarchy All processors inherit from the abstract `Processor` base class: | Class | Purpose | | ------------------------- | ------------------------------------------------------------------------- | | `Processor` | Abstract base class with `name`, optional `close()`, and `attach_agent()` | | `VideoProcessor` | Receives video tracks via `process_video()` | | `VideoPublisher` | Outputs video via `publish_video_track()` | | `VideoProcessorPublisher` | Receives and outputs video (e.g., annotated frames) | | `AudioProcessor` | Receives audio via `process_audio()` | | `AudioPublisher` | Outputs audio via `publish_audio_track()` | | `AudioProcessorPublisher` | Receives and outputs audio | ## Base Processor All processors must implement `name`. The `close()` method is inherited from `Component` with a no-op default — override it to release resources. The `attach_agent()` method is optional. ```python theme={null} from vision_agents.core.processors import Processor class MyProcessor(Processor): name = "my_processor" async def close(self) -> None: """Clean up resources when the agent stops.""" pass def attach_agent(self, agent: "Agent") -> None: """Called once when the processor is registered with an agent. Use this to access agent.events for custom event registration. """ self._agent = agent ``` ## Video Processor Receives video tracks from participants. The agent provides a shared `VideoForwarder` that distributes frames to all processors. ```python theme={null} from typing import Optional import aiortc from vision_agents.core.processors import VideoProcessor from vision_agents.core.utils.video_forwarder import VideoForwarder class MyVideoProcessor(VideoProcessor): name = "my_video_processor" def __init__(self): self._forwarder: Optional[VideoForwarder] = None async def process_video( self, track: aiortc.VideoStreamTrack, participant_id: Optional[str], shared_forwarder: Optional[VideoForwarder] = None, ) -> None: """Called when a participant publishes video.""" self._forwarder = shared_forwarder self._forwarder.add_frame_handler( self._on_frame, fps=5.0, name="my_handler" ) async def _on_frame(self, frame): # Process the frame pass async def stop_processing(self) -> None: """Called when video tracks are removed.""" if self._forwarder: await self._forwarder.remove_frame_handler(self._on_frame) self._forwarder = None async def close(self) -> None: await self.stop_processing() ``` ## Video Publisher Outputs a video track to the call (e.g., AI-generated video or avatars). ```python theme={null} import aiortc from vision_agents.core.processors import VideoPublisher from vision_agents.core.utils.video_track import QueuedVideoTrack class MyVideoPublisher(VideoPublisher): name = "my_video_publisher" def __init__(self): self._track = QueuedVideoTrack(width=1280, height=720, fps=30) def publish_video_track(self) -> aiortc.VideoStreamTrack: """Return the track to publish.""" return self._track async def close(self) -> None: self._track.stop() ``` ## Video Processor + Publisher For processors that receive video and output transformed frames (e.g., object detection with annotations). ```python theme={null} import av from typing import Optional import aiortc from vision_agents.core.processors import VideoProcessorPublisher from vision_agents.core.utils.video_forwarder import VideoForwarder from vision_agents.core.utils.video_track import QueuedVideoTrack class AnnotationProcessor(VideoProcessorPublisher): name = "annotation_processor" def __init__(self, fps: int = 30): self.fps = fps self._forwarder: Optional[VideoForwarder] = None self._track = QueuedVideoTrack() async def process_video( self, track: aiortc.VideoStreamTrack, participant_id: Optional[str], shared_forwarder: Optional[VideoForwarder] = None, ) -> None: self._forwarder = shared_forwarder self._forwarder.add_frame_handler( self._process_frame, fps=float(self.fps), name="annotator" ) async def _process_frame(self, frame: av.VideoFrame): # Transform the frame annotated = self._annotate(frame) await self._track.add_frame(annotated) def _annotate(self, frame: av.VideoFrame) -> av.VideoFrame: # Add annotations to frame return frame def publish_video_track(self) -> aiortc.VideoStreamTrack: return self._track async def stop_processing(self) -> None: if self._forwarder: await self._forwarder.remove_frame_handler(self._process_frame) self._forwarder = None async def close(self) -> None: await self.stop_processing() self._track.stop() ``` ## Audio Processor Receives audio data from participants. Audio is delivered as `PcmData` chunks. ```python theme={null} from getstream.video.rtc import PcmData from vision_agents.core.processors import AudioProcessor class MyAudioProcessor(AudioProcessor): name = "my_audio_processor" async def process_audio(self, audio_data: PcmData) -> None: """Process incoming audio. Args: audio_data: Contains samples, sample_rate, channels, and participant info. """ samples = audio_data.samples participant = audio_data.participant # Process audio... async def close(self) -> None: pass ``` ## Audio Publisher Outputs an audio track to the call. ```python theme={null} import aiortc from vision_agents.core.processors import AudioPublisher class MyAudioPublisher(AudioPublisher): name = "my_audio_publisher" def publish_audio_track(self) -> aiortc.AudioStreamTrack: """Return the audio track to publish.""" return self._audio_track async def close(self) -> None: pass ``` ## Usage Pass processors to the agent at initialization: ```python theme={null} from vision_agents.core import Agent, User from vision_agents.plugins import getstream, openai agent = Agent( edge=getstream.Edge(), agent_user=User(name="AI Assistant", id="agent"), llm=openai.Realtime(), processors=[your_processor], ) ``` For complete examples including YOLO pose detection and object detection, see [Building Video Processors](/guides/video-processors). # Realtime Class Source: https://visionagents.ai/core/realtime-core The Realtime component provides end-to-end speech-to-speech communication, combining STT, LLM, and TTS functionality in a single, optimized interface. It delivers ultra-low latency speech processing, direct audio streaming without intermediate text conversion, and support for multiple modalities (audio, video, text). ## When to Use Realtime Use a **Realtime** LLM when you want the lowest latency voice interactions. The model handles speech recognition, response generation, and speech synthesis natively—no separate STT or TTS services required. Use the **traditional STT → LLM → TTS** pipeline when you need custom voices (e.g., Cartesia, ElevenLabs), specific transcription providers, or models that don't support realtime audio. ## Supported Providers * [OpenAI Realtime](/integrations/realtime/openai) — WebRTC-based, supports video * [Gemini Live](/integrations/realtime/gemini) — WebSocket-based, supports video * [AWS Nova](/integrations/realtime/aws-bedrock) — WebSocket-based * [Qwen Omni](/integrations/realtime/qwen) — Native audio support ## Basic Usage ```python theme={null} from vision_agents.core import Agent, User from vision_agents.plugins import openai, getstream agent = Agent( edge=getstream.Edge(), agent_user=User(name="AI Assistant", id="agent"), instructions="You're a helpful voice assistant", llm=openai.Realtime(model="gpt-realtime", voice="marin"), processors=[] ) ``` ## Agent methods with realtime ```python theme={null} await agent.simple_response("What do you see in the video?", interrupt=True) ``` Use `agent.simple_response(...)` to inject text prompts. `agent.say()` is not supported with Realtime LLMs — use `simple_response()` instead. You usually do not call realtime audio methods directly from app code. ## Properties | Property | Type | Description | | ------------ | ------ | ---------------------------------------------------------------------------------------------------------------------------------------------- | | `connected` | `bool` | `True` if the realtime session is active | | `fps` | `int` | Video frames per second sent to the model (default: 1) | | `session_id` | `str` | UUID identifying the current session | | `epoch` | `int` | Monotonic interruption counter. Increments each time `interrupt()` is called, allowing stale audio output events to be identified and dropped. | ## Realtime methods ### `interrupt()` Increments the epoch counter so that any in-flight audio output from a previous response is detected as stale and discarded by the Agent. The Agent calls this automatically on barge-in. ## Events The Realtime class emits a small set of events for connection state: | Event | Description | | --------------------------- | --------------------------------------------------------- | | `RealtimeConnectedEvent` | Connection established with session config & capabilities | | `RealtimeDisconnectedEvent` | Connection closed (includes `reason` and `clean` flag) | For conversation events, subscribe to the agent-level events — `UserTranscriptEvent` fires in both classic STT and realtime modes: ```python theme={null} from vision_agents.core.agents.events import UserTranscriptEvent @agent.events.subscribe async def on_user_speech(event: UserTranscriptEvent): print(f"User said: {event.text}") ``` See [Events Reference](/reference/events-reference) for the full event surface, including LLM, tool, and error events. For provider-specific parameters and configuration, see the integration docs for [OpenAI](/integrations/realtime/openai), [Gemini](/integrations/realtime/gemini), [AWS Bedrock](/integrations/realtime/aws-bedrock), or [Qwen](/integrations/realtime/qwen). # Speech-to-Text and Text-to-Speech Class Source: https://visionagents.ai/core/stt-tts-core LLMs not running using a `Realtime` model requires some help to convert the user's speech and LLM responses into something the user can speak to and hear. To achieve this, the `Agent` class exposes two parameters `tts` and `stt` allowing developers to pass in any text-to-speech and speech-to-text service they like. Using this method, the output voices can be configured, the transcription rate can be adjusted and more. Internally, the Agent class handles the management between these services and things such as setting up the audio track for the STT providers as an example. ### STT (Speech-to-Text) STT components convert audio input into text for processing by the LLM. All implementations follow a standardised interface with consistent event emission. These components process real-time audio with `PcmData` objects from `getstream.video.rtc.track_util`, provide partial transcript support for responsive UI, and include comprehensive error handling and connection management. Multiple providers are supported including Deepgram, ElevenLabs, Fast Whisper, and others. The Agent starts STT automatically during `join()`. Plugin authors may override `start()` to initialize provider connections; do not call `start()` manually or twice — a second call raises `ValueError`. Some STT providers include built-in turn detection (indicated by the `turn_detection` property). When this is the case, the Agent automatically skips any separately configured `TurnDetector` to avoid conflicts. #### STT Methods | Method | Description | | -------------------------------------- | ---------------------------------------------------------------------------------------------------- | | `start()` | Initialize connections and resources. Called automatically by the Agent on join; override in plugins | | `process_audio(pcm_data, participant)` | Process an audio frame (\~20ms chunks) | | `clear()` | Clear any pending audio or internal state | | `close()` | Clean up resources | #### STT Events | Event | Description | | ---------------------- | ------------------------------------------------- | | `STTConnectedEvent` | STT connection established | | `STTDisconnectedEvent` | STT connection closed (with `reason` and `clean`) | | `STTErrorEvent` | Temporary, recoverable error | Final user transcripts surface on the agent as `UserTranscriptEvent` (from `vision_agents.core.agents.events`) — not as a separate STT event — so the same handler works in both classic STT and realtime modes. See [Events Reference](/reference/events-reference). ### TTS (Text-to-Speech) TTS components convert LLM responses into audio output. They handle audio synthesis and streaming to the output track. These components provide streaming audio synthesis for low latency, multiple voice options and customisation, audio format standardisation using `PcmData` and `AudioFormat` from `getstream.video.rtc.track_util`, and support for providers like ElevenLabs, Cartesia, and others. #### TTS Methods | Method | Description | | ----------------------------------- | --------------------------------------------------------------------- | | `stream_audio(text, ...)` | Plugin implementer API — convert text to `PcmData` or byte chunks | | `send_iter(text, participant=None)` | Convert text to speech and yield `TTSOutputChunk` items | | `stop_audio()` | Clear the audio queue and stop current playback | | `interrupt()` | Increment the interruption epoch and cancel stale in-flight synthesis | | `close()` | Clean up resources | Output is standardized as `PcmData` through the inference pipeline; the Agent re-chunks audio to 20ms frames for the outbound track. #### TTS Events | Event | Description | | --------------------------- | ------------------------------------------------------------------------------------------------- | | `TTSSynthesisStartEvent` | Synthesis has begun for a text input | | `TTSSynthesisCompleteEvent` | Synthesis finished (includes metrics like `synthesis_time_ms`, `chunk_count`, `real_time_factor`) | | `TTSConnectedEvent` | TTS connection established | | `TTSDisconnectedEvent` | TTS connection closed (with `reason` and `clean`) | | `TTSErrorEvent` | Temporary, recoverable error | #### Interruption support The TTS base class exposes an `epoch` property and an `interrupt()` method for handling barge-in scenarios: | Member | Type | Description | | ------------- | ------- | --------------------------------------------------------------------------------------------------- | | `epoch` | `int` | Monotonic counter that increments on each interruption. Used to identify stale audio events. | | `interrupt()` | `async` | Increments the epoch and stops the current audio synthesis. Stale events are automatically dropped. | You usually do not need to call `interrupt()` manually — `agent.simple_response(..., interrupt=True)` and `agent.say(..., interrupt=True)` route interruption through the active inference flow. # Telemetry & Metrics Source: https://visionagents.ai/core/telemetry Vision Agents provides built-in observability through [OpenTelemetry](https://opentelemetry.io/). Collect metrics and traces across all components to monitor performance, latency, and errors in your agents. ## Quick Start To enable metrics collection, configure OpenTelemetry: ```python theme={null} # 1. Configure OpenTelemetry from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.exporter.prometheus import PrometheusMetricReader from prometheus_client import start_http_server start_http_server(9464) reader = PrometheusMetricReader() metrics.set_meter_provider(MeterProvider(metric_readers=[reader])) # 2. Now import and create your agent from vision_agents.core import Agent, User from vision_agents.plugins import deepgram, getstream, openai agent = Agent( edge=getstream.Edge(), agent_user=User(name="Metrics Agent", id="agent"), llm=openai.LLM(model="gpt-4o"), stt=deepgram.STT(), tts=deepgram.TTS(), ) ``` Metrics are now available at `http://localhost:9464/metrics`. ## MetricsCollector The `MetricsCollector` records metrics via normalized `on_*()` hooks that plugins call directly. Each `Agent` creates a root collector and merges child collectors from STT, TTS, LLM, and other components during initialization, so metrics collection is enabled by default. If no OpenTelemetry providers are configured, metrics are no-ops and have no performance impact. For new integrations, prefer the collector's normalized `on_*` metric hooks and `agent.metrics` over provider-specific event classes. Plugins record metrics through hooks such as: * **LLM** — `on_llm_response`, `on_tool_call`, `on_llm_error` * **STT** — `on_stt_transcript`, `on_stt_error` * **TTS** — `on_tts_synthesis`, `on_tts_error` * **Turn Detection** — `on_turn_ended` * **Realtime LLM** — `on_realtime_audio_input/output`, `on_realtime_user/agent_transcription` * **VLM** — `on_vlm_inference`, `on_vlm_error` * **Video Processors** — `on_video_detection` ### Metric Attributes All metrics include contextual attributes: | Attribute | Description | | ------------ | -------------------------------------------- | | `provider` | The plugin name (e.g., `openai`, `deepgram`) | | `model` | Model identifier when available | | `error_type` | Exception class name for error metrics | | `error_code` | Error code when available | ## Metrics Reference All metrics use the `vision_agents.core` meter namespace. ### STT Metrics | Metric | Type | Unit | Description | | ----------------------- | --------- | ---- | ------------------------------------- | | `stt.latency.ms` | Histogram | ms | Processing latency for speech-to-text | | `stt.audio_duration.ms` | Histogram | ms | Duration of audio processed | | `stt.errors` | Counter | — | Total STT errors | ### TTS Metrics | Metric | Type | Unit | Description | | ----------------------- | --------- | ---- | ----------------------------- | | `tts.latency.ms` | Histogram | ms | Synthesis latency | | `tts.audio_duration.ms` | Histogram | ms | Duration of synthesized audio | | `tts.characters` | Counter | — | Characters synthesized | | `tts.errors` | Counter | — | Total TTS errors | ### LLM Metrics | Metric | Type | Unit | Description | | ---------------------------- | --------- | ---- | -------------------------------------- | | `llm.latency.ms` | Histogram | ms | Response latency (request to complete) | | `llm.time_to_first_token.ms` | Histogram | ms | Time to first token (streaming) | | `llm.tokens.input` | Counter | — | Input/prompt tokens consumed | | `llm.tokens.output` | Counter | — | Output/completion tokens generated | | `llm.tool_calls` | Counter | — | Tool/function calls executed | | `llm.tool_latency.ms` | Histogram | ms | Tool execution latency | | `llm.errors` | Counter | — | Total LLM errors | ### Turn Detection Metrics | Metric | Type | Unit | Description | | -------------------------- | --------- | ---- | --------------------------------- | | `turn.duration.ms` | Histogram | ms | Duration of detected speech turns | | `turn.trailing_silence.ms` | Histogram | ms | Silence duration before turn end | ### Realtime LLM Metrics For speech-to-speech models like OpenAI Realtime: | Metric | Type | Unit | Description | | ----------------------------------- | --------- | ----- | ----------------------------- | | `realtime.sessions` | Counter | — | Sessions started | | `realtime.session_duration.ms` | Histogram | ms | Session duration | | `realtime.audio.input.bytes` | Counter | bytes | Audio bytes sent to LLM | | `realtime.audio.output.bytes` | Counter | bytes | Audio bytes received from LLM | | `realtime.audio.input.duration.ms` | Counter | ms | Audio duration sent | | `realtime.audio.output.duration.ms` | Counter | ms | Audio duration received | | `realtime.responses` | Counter | — | Complete responses received | | `realtime.transcriptions.user` | Counter | — | User speech transcriptions | | `realtime.transcriptions.agent` | Counter | — | Agent speech transcriptions | | `realtime.errors` | Counter | — | Realtime errors | ### VLM / Vision Metrics | Metric | Type | Unit | Description | | -------------------------- | --------- | ---- | --------------------------- | | `vlm.inference.latency.ms` | Histogram | ms | VLM inference latency | | `vlm.inferences` | Counter | — | Inference requests | | `vlm.tokens.input` | Counter | — | Input tokens (text + image) | | `vlm.tokens.output` | Counter | — | Output tokens | | `vlm.errors` | Counter | — | VLM errors | ### Video Processor Metrics | Metric | Type | Unit | Description | | ----------------------------- | --------- | ---- | ------------------------ | | `video.frames.processed` | Counter | — | Frames processed | | `video.processing.latency.ms` | Histogram | ms | Frame processing latency | | `video.detections` | Counter | — | Objects/items detected | ## AgentMetrics For in-process metrics without external infrastructure, access aggregated metrics directly from the agent: ```python theme={null} # After running your agent metrics = agent.metrics # STT print(f"Average STT latency: {metrics.stt_latency_ms__avg.value()} ms") print(f"Total audio processed: {metrics.stt_audio_duration_ms__total.value()} ms") # TTS print(f"Average TTS latency: {metrics.tts_latency_ms__avg.value()} ms") print(f"Characters synthesized: {metrics.tts_characters__total.value()}") # LLM print(f"Average LLM latency: {metrics.llm_latency_ms__avg.value()} ms") print(f"Input tokens: {metrics.llm_input_tokens__total.value()}") print(f"Output tokens: {metrics.llm_output_tokens__total.value()}") print(f"Tool calls: {metrics.llm_tool_calls__total.value()}") ``` ### Available AgentMetrics | Metric | Type | Description | | ------------------------------------------ | ------- | -------------------------------- | | `stt_latency_ms__avg` | Average | Average STT processing latency | | `stt_audio_duration_ms__total` | Counter | Total audio duration processed | | `tts_latency_ms__avg` | Average | Average TTS synthesis latency | | `tts_audio_duration_ms__total` | Counter | Total synthesized audio duration | | `tts_characters__total` | Counter | Total characters synthesized | | `llm_latency_ms__avg` | Average | Average LLM response latency | | `llm_time_to_first_token_ms__avg` | Average | Average time to first token | | `llm_input_tokens__total` | Counter | Total input tokens | | `llm_output_tokens__total` | Counter | Total output tokens | | `llm_tool_calls__total` | Counter | Total tool calls | | `llm_tool_latency_ms__avg` | Average | Average tool execution latency | | `turn_duration_ms__avg` | Average | Average turn duration | | `turn_trailing_silence_ms__avg` | Average | Average trailing silence | | `realtime_audio_input_bytes__total` | Counter | Total audio bytes sent | | `realtime_audio_output_bytes__total` | Counter | Total audio bytes received | | `realtime_audio_input_duration_ms__total` | Counter | Total input audio duration | | `realtime_audio_output_duration_ms__total` | Counter | Total output audio duration | | `realtime_user_transcriptions__total` | Counter | Total user transcriptions | | `realtime_agent_transcriptions__total` | Counter | Total agent transcriptions | | `vlm_inference_latency_ms__avg` | Average | Average VLM inference latency | | `vlm_inferences__total` | Counter | Total VLM inferences | | `vlm_input_tokens__total` | Counter | Total VLM input tokens | | `vlm_output_tokens__total` | Counter | Total VLM output tokens | | `video_frames_processed__total` | Counter | Total frames processed | | `video_processing_latency_ms__avg` | Average | Average frame processing latency | ## Prometheus Setup Export metrics to Prometheus for monitoring dashboards and alerting. **Step 1 — Install the exporter** ```bash theme={null} uv add opentelemetry-exporter-prometheus prometheus-client ``` **Step 2 — Configure OpenTelemetry** ```python theme={null} from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.exporter.prometheus import PrometheusMetricReader from prometheus_client import start_http_server # Start HTTP server for Prometheus scraping start_http_server(port=9464) # Configure OpenTelemetry reader = PrometheusMetricReader() metrics.set_meter_provider(MeterProvider(metric_readers=[reader])) ``` **Step 3 — Create and run your agent** ```python theme={null} from vision_agents.core import Agent, AgentLauncher, Runner agent = Agent(...) # MetricsCollector is automatically attached # Run with CLI Runner(AgentLauncher(create_agent=..., join_call=...)).cli() ``` View metrics at `http://localhost:9464/metrics`. ## Tracing with Jaeger Trace requests across components for debugging latency issues. **Step 1 — Install the exporter** ```bash theme={null} uv add opentelemetry-sdk opentelemetry-exporter-otlp ``` **Step 2 — Configure tracing** ```python theme={null} from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter resource = Resource.create({"service.name": "my-agent"}) provider = TracerProvider(resource=resource) exporter = OTLPSpanExporter(endpoint="localhost:4317", insecure=True) provider.add_span_processor(BatchSpanProcessor(exporter)) trace.set_tracer_provider(provider) ``` **Step 3 — Run Jaeger** ```bash theme={null} docker run --rm -it \ -e COLLECTOR_OTLP_ENABLED=true \ -p 16686:16686 -p 4317:4317 -p 4318:4318 \ jaegertracing/all-in-one:1.51 ``` View traces at `http://localhost:16686`. ## Complete Example ```python theme={null} """Prometheus metrics example with Vision Agents.""" # Configure OpenTelemetry from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.exporter.prometheus import PrometheusMetricReader from prometheus_client import start_http_server start_http_server(9464) reader = PrometheusMetricReader() metrics.set_meter_provider(MeterProvider(metric_readers=[reader])) # Now import agents from vision_agents.core import Agent, User, AgentLauncher, Runner from vision_agents.plugins import deepgram, getstream, gemini, elevenlabs async def create_agent(**kwargs) -> Agent: return Agent( edge=getstream.Edge(), agent_user=User(name="Metrics Agent", id="agent"), instructions="You're a helpful voice assistant.", llm=gemini.LLM("gemini-flash-lite-latest"), tts=elevenlabs.TTS(), stt=deepgram.STT(), ) async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None: # MetricsCollector is automatically attached to the agent call = await agent.create_call(call_type, call_id) async with agent.join(call): await agent.simple_response("Hello! Metrics are being collected.") await agent.finish() # Print summary after call m = agent.metrics print(f"LLM latency: {m.llm_latency_ms__avg.value():.1f} ms") print(f"Tokens: {m.llm_input_tokens__total.value()} in / {m.llm_output_tokens__total.value()} out") if __name__ == "__main__": Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli() ``` Run with: ```bash theme={null} uv run agent.py run --call-type default --call-id test ``` Metrics available at `http://localhost:9464/metrics`. ## Example Prometheus Queries OpenTelemetry metric names use dots (e.g., `llm.latency.ms`). Prometheus converts these to underscores when scraping (e.g., `llm_latency_ms`). **Average LLM latency over time:** ```promql theme={null} rate(llm_latency_ms_sum[5m]) / rate(llm_latency_ms_count[5m]) ``` **Total tokens used:** ```promql theme={null} sum(llm_tokens_input) + sum(llm_tokens_output) ``` **Error rate:** ```promql theme={null} rate(llm_errors_total[5m]) ``` ## Best Practices **Configure OpenTelemetry** - Set up providers to enable metric collection. If no providers are configured, metrics are no-ops. **MetricsCollector is automatic** - Each Agent automatically creates a MetricsCollector internally. If no OpenTelemetry provider is configured, metrics are no-ops with no performance impact. **Use AgentMetrics for simple logging** - Access `agent.metrics` directly for in-process metrics without external infrastructure. **Add resource attributes** - Include service name and environment in your metrics: ```python theme={null} from opentelemetry.sdk.resources import Resource resource = Resource.create({ "service.name": "my-agent", "service.version": "1.0.0", "deployment.environment": "production", }) provider = MeterProvider(resource=resource, metric_readers=[reader]) ``` **Set up alerting on:** * LLM latency > 2000ms (p95) * Error rate > 1% * Token usage anomalies ## Next Steps * [Kubernetes Deployment](/guides/kubernetes-deployment) - Helm chart with Prometheus and Grafana out of the box * [Built-in HTTP Server](/guides/http-server) - Console mode and HTTP server for session management # Expressive Voice Narrator Source: https://visionagents.ai/examples/cartesia-narrator Build a storytelling agent with expressive speech using Cartesia's Sonic 3 TTS