Vision Agents Skill
Product Summary
Vision Agents is an open-source Python framework for building real-time voice and video AI agents. You write an Agent that joins a session, connects to AI providers through swappable plugins (LLM, STT, TTS, vision models), and responds in real time. The framework handles call lifecycle, audio/video routing, turn-taking, and deployment. Key files: agent.py (agent definition), .env (API keys), pyproject.toml (dependencies). CLI: uv run agent.py run (console mode), uv run agent.py serve (HTTP server). Integrations: 35+ providers across LLMs, speech, vision, avatars, and telephony. Primary docs: https://visionagents.ai
When to Use
Reach for this skill when:
- Building voice agents with realtime models (Gemini, OpenAI, Qwen) or custom STT/LLM/TTS pipelines
- Deploying agents to production (Docker, Kubernetes, HTTP server with session management)
- Adding function calling, MCP servers, or RAG (Gemini FileSearch, TurboPuffer) to agents
- Integrating phone calls (Twilio, Telnyx) or video processing (YOLO, VLMs, avatars)
- Configuring turn detection, interruption handling, or multi-speaker audio routing
- Testing agents with pytest without spinning up audio/video infrastructure
- Monitoring agents with OpenTelemetry metrics and event subscriptions
- Swapping AI providers (LLM, STT, TTS, vision) without rewriting agent logic
Quick Reference
Agent Modes
| Mode |
Best For |
Setup |
| Realtime |
Lowest latency, native video |
llm=gemini.Realtime() — one provider handles speech in/out |
| Custom Pipeline |
Full control, mixed providers |
llm=gemini.LLM(), stt=deepgram.STT(), tts=elevenlabs.TTS() |
| Video (VLM) |
Frame analysis, video understanding |
llm=nvidia.VLM(fps=1, framebufferseconds=10) |
| Video (Processor) |
Object detection, pose estimation |
processors=[ultralytics.YOLOPoseProcessor(...)] |
Core Agent Constructor
Agent(
edge=getstream.Edge(), # Transport layer
agent_user=User(name="...", id="agent"), # Agent identity
instructions="...", # System prompt
llm=gemini.Realtime(), # LLM (realtime or standard)
stt=deepgram.STT(), # Speech-to-text (optional in realtime)
tts=elevenlabs.TTS(), # Text-to-speech (optional in realtime)
turn_detection=smart_turn.TurnDetector(), # Turn detection (optional)
processors=[...], # Video processors (optional)
avatar=anam.Avatar(...), # Avatar (optional)
mcp_servers=[...], # MCP servers for tools (optional)
)
Essential Methods
| Method |
Purpose |
await agent.createcall(calltype, call_id) |
Create a call on the edge provider |
async with agent.join(call): |
Join call as context manager (required) |
await agent.simple_response(text, interrupt=True) |
Send text to LLM, speak response |
await agent.say(text, interrupt=False) |
Speak text directly (bypass LLM) |
await agent.finish() |
Wait for call to end gracefully |
await agent.close() |
Clean up resources (called automatically) |
@agent.events.subscribe |
Subscribe to events (participant joins, transcripts, errors) |
@llm.register_function(description="...") |
Register tool for function calling |
Deployment Commands
# Console mode (development)
uv run agent.py run
# HTTP server (production)
uv run agent.py serve --host 0.0.0.0 --port 8000
# Docker
docker build -t my-agent .
docker run -e GOOGLE_API_KEY=... my-agent
# Kubernetes (with Helm)
helm install my-agent ./helm-chart
HTTP Server Endpoints
| Method |
Endpoint |
Purpose |
| POST |
/calls/{call_id}/sessions |
Start agent session |
| DELETE |
/calls/{callid}/sessions/{sessionid} |
Close session |
| GET |
/calls/{callid}/sessions/{sessionid}/metrics |
Get performance metrics |
| GET |
/health |
Liveness check |
| GET |
/ready |
Readiness check |
Plugin Installation
# Add plugins as extras
uv add "vision-agents[deepgram,elevenlabs,gemini]"
# Or explicit packages
uv add vision-agents-plugins-deepgram vision-agents-plugins-elevenlabs
Decision Guidance
When to Use Realtime vs Custom Pipeline
| Scenario |
Use Realtime |
Use Custom Pipeline |
| Fastest time to market, lowest latency |
✓ |
|
| Need specific STT provider (e.g., Deepgram with eager turn detection) |
|
✓ |
| Want to mix LLM, STT, TTS from different providers |
|
✓ |
| Need function calling with full control |
|
✓ |
| Building video agent with native vision support |
✓ |
|
| Prototyping quickly |
✓ |
|
When to Use Gemini FileSearch vs TurboPuffer for RAG
| Factor |
Gemini FileSearch |
TurboPuffer |
| Setup complexity |
Simple |
More setup |
| Chunking |
Automatic |
Configurable |
| Search type |
Managed |
Hybrid (vector + BM25) |
| Control level |
Less |
Full |
| Cost |
Included with Gemini |
Separate service |
| Best for |
Prototypes |
Production with custom needs |
When to Use HTTP Server vs Console Mode
| Use Case |
Console Mode |
HTTP Server |
| Local development |
✓ |
|
| Testing with browser demo |
✓ |
|
| Production deployment |
|
✓ |
| Multiple concurrent agents |
|
✓ |
| Session management, scaling |
|
✓ |
Workflow
1. Build a New Agent
- Scaffold the project:
``bash uvx vision-agents init my-agent && cd my-agent ``
- Add API keys to
.env:
``bash cp .env.example .env # Fill in: STREAMAPIKEY, STREAMAPISECRET, GOOGLEAPIKEY, etc. ``
- Review
agent.py — understand the three parts:
- createagent() — builds the Agent with plugins - joincall() — defines what happens when agent joins - runner — CLI entry point
- Customize instructions and plugins in
create_agent():
``python async def createagent(**kwargs) -> Agent: return Agent( edge=getstream.Edge(), agentuser=User(name="My Assistant", id="agent"), instructions="You're a helpful voice assistant.", llm=gemini.Realtime(), # or custom pipeline ) ``
- Run locally:
``bash uv run agent.py run `` Open the browser link to test.
2. Add Function Calling
- Register functions on the LLM:
```python llm = gemini.LLM() # Not Realtime
@llm.registerfunction(description="Get weather for a location") async def getweather(location: str) -> dict: return {"temp": "22C", "condition": "Sunny"} ```
- Pass the LLM to Agent:
``python agent = Agent( edge=getstream.Edge(), llm=llm, stt=deepgram.STT(), tts=elevenlabs.TTS(), ) ``
- LLM calls functions automatically during conversation.
3. Add RAG (Knowledge Base)
- Choose a provider:
- Gemini FileSearch (simple): await store.adddirectory("./docs") - TurboPuffer (full control): await rag.adddirectory("./docs")
- Register as function:
``python @llm.registerfunction(description="Search knowledge base") async def searchdocs(query: str) -> str: return await rag.search(query, top_k=5) ``
- Agent calls it when relevant.
4. Deploy to Production
- Run HTTP server locally first:
``bash uv run agent.py serve --host 0.0.0.0 --port 8000 ``
- Containerize with Docker:
- Use the scaffolded Dockerfile - Build: docker build -t my-agent . - Run: docker run -e GOOGLEAPIKEY=... my-agent
- Scale horizontally (multiple replicas):
- Add Redis for session registry (see Horizontal Scaling guide) - Deploy multiple containers behind a load balancer
- Orchestrate with Kubernetes:
- Use the Helm chart from examples - Configure health probes, resource limits, metrics
- Monitor with OpenTelemetry:
- Export metrics to Prometheus - View dashboards in Grafana
5. Test Agents
- Use
TestSession for text-only testing:
```python from vision_agents.testing import TestSession, LLMJudge
async def testgreeting(): llm = gemini.LLM() async with TestSession(llm=llm, instructions="Be friendly") as session: response = await session.simpleresponse("Hello") assert response.function_calls == [] ```
- Mock functions for call tracking:
```python async def fakeweather(**) -> dict: return {"temp": 55}
with session.mockfunctions({"getweather": fakeweather}) as mocked: response = await session.simpleresponse("Weather?") mocked["getweather"].assertcalled_once() ```
- Use LLMJudge to evaluate intent:
``python judge = LLMJudge(gemini.LLM()) verdict = await judge.evaluate(response.chat_messages[0], intent="Friendly greeting") assert verdict.success ``
Common Gotchas
- Don't reuse Agent instances. Create a new agent for each call. Calling
join() twice raises RuntimeError.
- Realtime LLMs disable STT/TTS. When using
llm=gemini.Realtime(), don't pass stt or tts — they're ignored with a warning.
- Turn detection conflicts. Don't use
turn_detection with Realtime LLMs (they handle it internally). If STT has built-in turn detection (e.g., Deepgram), the separate plugin is ignored.
- Async functions only.
@llm.register_function() requires async functions; sync functions raise ValueError.
- Event handlers are fire-and-forget. Don't rely on handlers completing before the next line of agent code. Use
agent.simple_response(..., interrupt=True) for synchronous control.
- Agent requires at least one audio path. In non-realtime mode, provide STT, TTS, turn detection, or video processors; video-only agents without LLM are allowed only with processors.
- API keys in
.env. Vision Agents auto-loads from .env for all plugins. Missing keys cause silent failures at runtime.
- Session limits in production. Set
maxconcurrentsessions, maxsessionspercall, and agentidle_timeout to prevent resource exhaustion.
- Interrupt parameter behavior.
interrupt=True preempts in-flight responses; interrupt=False queues after current output. Use interrupt=True for urgent messages.
- Video override path. Set
agent.setvideotrackoverridepath() before calling join(), not after.
Verification Checklist
Before submitting agent code:
Resources
Comprehensive navigation: https://visionagents.ai/llms.txt
Critical documentation:
- Quickstart — Build your first agent in 5 minutes
- Voice Agents — Realtime vs custom pipeline, function calling, phone integration
- Deploying Overview — Path from local dev to Kubernetes
- Agent Class Reference — Full API, lifecycle, event system
- HTTP Server — Session management, scaling, authentication
- Integrations — 35+ providers, installation, swapping
- Testing — TestSession, mocking, LLMJudge
- Event System — Subscribing, patterns, error handling
For additional documentation and navigation, see: https://visionagents.ai/llms.txt