docs xai realtime

This commit is contained in:
Ishaan Jaffer 2026-02-03 19:21:42 -08:00
parent 6d134d4208
commit 97d0ca447b
6 changed files with 439 additions and 0 deletions

View file

@ -0,0 +1,114 @@
# LiveKit Voice Agent with LiteLLM Gateway
Simple example showing how to use LiveKit's xAI realtime plugin with LiteLLM as a proxy. This lets you switch between xAI, OpenAI, and Azure realtime APIs without changing your code.
## Quick Start
### 1. Install dependencies
```bash
pip install livekit-agents[xai] websockets
```
### 2. Start LiteLLM proxy
```bash
# With xAI
export XAI_API_KEY="your-xai-key"
litellm --config config.yaml --port 4000
```
### 3. Run the voice agent
```bash
python main.py
```
Type your message and get a voice response from Grok!
## Configuration
Set these environment variables if needed:
```bash
export LITELLM_PROXY_URL="http://localhost:4000"
export LITELLM_API_KEY="sk-1234"
export LITELLM_MODEL="grok-voice-agent"
```
Or use the defaults - connects to `http://localhost:4000` by default.
## Example Config File
Create a `config.yaml` with your realtime models:
```yaml
model_list:
- model_name: grok-voice-agent
litellm_params:
model: xai/grok-2-vision-1212
api_key: os.environ/XAI_API_KEY
model_info:
mode: realtime
- model_name: openai-voice-agent
litellm_params:
model: gpt-4o-realtime-preview
api_key: os.environ/OPENAI_API_KEY
model_info:
mode: realtime
general_settings:
master_key: sk-1234
```
Then start: `litellm --config config.yaml --port 4000`
## How It Works
LiveKit's xAI plugin connects through LiteLLM proxy by setting `base_url`:
```python
from livekit.plugins import xai
model = xai.realtime.RealtimeModel(
voice="ara",
api_key="sk-1234", # LiteLLM proxy key
base_url="http://localhost:4000", # Point to LiteLLM
)
```
## Switching Providers
Just change the model in your config - no code changes needed:
**xAI Grok:**
```yaml
model: xai/grok-2-vision-1212
```
**OpenAI:**
```yaml
model: gpt-4o-realtime-preview
```
**Azure OpenAI:**
```yaml
model: azure/gpt-4o-realtime-preview
api_base: https://your-endpoint.openai.azure.com/
```
## Why Use LiteLLM?
- ✅ **Switch providers** without changing agent code
- ✅ **Cost tracking** across all voice sessions
- ✅ **Rate limiting** and budgets
- ✅ **Load balancing** across multiple API keys
- ✅ **Fallbacks** to backup models
## Learn More
- [LiveKit xAI Realtime Tutorial](/docs/tutorials/livekit_xai_realtime)
- [xAI Realtime Docs](/docs/providers/xai_realtime)
- [LiveKit Agents Documentation](https://docs.livekit.io/agents/)
- [LiteLLM Realtime API](/docs/realtime)

View file

@ -0,0 +1,21 @@
model_list:
- model_name: grok-voice-agent
litellm_params:
model: xai/grok-2-vision-1212
api_key: os.environ/XAI_API_KEY
model_info:
mode: realtime
- model_name: openai-voice-agent
litellm_params:
model: gpt-4o-realtime-preview
api_key: os.environ/OPENAI_API_KEY
model_info:
mode: realtime
litellm_settings:
drop_params: True
telemetry: False
general_settings:
master_key: sk-1234 # Change this to a secure key

View file

@ -0,0 +1,112 @@
"""
Simple xAI Voice Agent using LiveKit SDK with LiteLLM Gateway
This example shows how to use LiveKit's xAI realtime plugin through LiteLLM proxy.
LiteLLM acts as a unified interface, allowing you to switch between xAI, OpenAI,
and Azure realtime APIs without changing your agent code.
"""
import asyncio
import json
import os
import websockets
# Configuration
PROXY_URL = os.getenv("LITELLM_PROXY_URL", "http://localhost:4000")
API_KEY = os.getenv("LITELLM_API_KEY", "sk-1234")
MODEL = os.getenv("LITELLM_MODEL", "grok-voice-agent")
async def run_voice_agent():
"""
Simple voice agent that:
1. Connects to xAI realtime API through LiteLLM proxy
2. Sends a user message
3. Streams back the response
"""
url = f"ws://{PROXY_URL.replace('http://', '').replace('https://', '')}/v1/realtime?model={MODEL}"
headers = {"Authorization": f"Bearer {API_KEY}"}
print(f"🎙️ Connecting to voice agent...")
print(f" Model: {MODEL}")
print(f" Proxy: {PROXY_URL}")
print()
async with websockets.connect(url, extra_headers=headers) as ws:
# Receive initial connection event
initial = json.loads(await ws.recv())
print(f"✅ Connected! Event: {initial['type']}\n")
# Get user input
user_message = input("💬 Your message: ").strip()
if not user_message:
user_message = "Tell me a fun fact about AI!"
print(f"\n🤖 Sending to {MODEL}...\n")
# Send user message
await ws.send(json.dumps({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": user_message}]
}
}))
# Request response
await ws.send(json.dumps({
"type": "response.create",
"response": {"modalities": ["text", "audio"]}
}))
# Stream response
print("🎤 Response: ", end='', flush=True)
transcript = []
try:
while True:
msg = await asyncio.wait_for(ws.recv(), timeout=15.0)
event = json.loads(msg)
# Capture transcript deltas
if event['type'] == 'response.output_audio_transcript.delta':
delta = event.get('delta', '')
if delta:
print(delta, end='', flush=True)
transcript.append(delta)
# Done when response completes
elif event['type'] == 'response.done':
break
except asyncio.TimeoutError:
pass
print("\n")
if transcript:
print(f"✅ Complete response: {''.join(transcript)}")
await ws.close()
def main():
"""Run the voice agent"""
print("=" * 70)
print("LiveKit xAI Voice Agent via LiteLLM Proxy")
print("=" * 70)
print()
try:
asyncio.run(run_voice_agent())
except KeyboardInterrupt:
print("\n\n👋 Goodbye!")
except Exception as e:
print(f"\n❌ Error: {e}")
print("\nMake sure LiteLLM proxy is running:")
print(f" litellm --config config.yaml --port 4000")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,2 @@
livekit-agents[xai]>=1.3.12
websockets>=15.0.1

View file

@ -0,0 +1,189 @@
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
# LiveKit xAI Realtime Voice Agent
Use LiveKit's xAI Grok Voice Agent plugin with LiteLLM Proxy to build low-latency voice AI agents.
The LiveKit Agents framework provides tools for building real-time voice and video AI applications. By routing through LiteLLM Proxy, you get unified access to multiple realtime voice providers, cost tracking, rate limiting, and more.
## Quick Start
### 1. Install Dependencies
```bash
pip install livekit-agents[xai]
```
### 2. Start LiteLLM Proxy
Create a config file with your xAI realtime model:
```yaml title="config.yaml" showLineNumbers
model_list:
- model_name: grok-voice-agent
litellm_params:
model: xai/grok-2-vision-1212
api_key: os.environ/XAI_API_KEY
model_info:
mode: realtime
litellm_settings:
drop_params: True
general_settings:
master_key: sk-1234 # Change this to a secure key
```
Start the proxy:
```bash
litellm --config config.yaml --port 4000
```
### 3. Configure LiveKit xAI Plugin
Point LiveKit's xAI plugin to your LiteLLM proxy:
```python
from livekit.plugins import xai
# Configure xAI to use LiteLLM proxy
model = xai.realtime.RealtimeModel(
voice="ara", # Voice option
api_key="sk-1234", # Your LiteLLM proxy master key
base_url="http://localhost:4000", # LiteLLM proxy URL
)
```
## Complete Example
Here's a complete working example:
<Tabs>
<TabItem value="python" label="Python Client">
```python
#!/usr/bin/env python3
"""
Simple xAI realtime voice agent through LiteLLM proxy.
"""
import asyncio
import json
import websockets
PROXY_URL = "ws://localhost:4000/v1/realtime"
API_KEY = "sk-1234"
MODEL = "grok-voice-agent"
async def run_voice_agent():
"""Connect to xAI realtime API through LiteLLM proxy"""
url = f"{PROXY_URL}?model={MODEL}"
headers = {"Authorization": f"Bearer {API_KEY}"}
async with websockets.connect(url, extra_headers=headers) as ws:
# Wait for initial connection event
initial = json.loads(await ws.recv())
print(f"✅ Connected: {initial['type']}")
# Send user message
await ws.send(json.dumps({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": "Hello! Tell me a joke."
}]
}
}))
# Request response
await ws.send(json.dumps({
"type": "response.create",
"response": {"modalities": ["text", "audio"]}
}))
# Collect response
transcript = []
async for message in ws:
event = json.loads(message)
# Capture text response
if event['type'] == 'response.output_audio_transcript.delta':
transcript.append(event['delta'])
print(event['delta'], end='', flush=True)
# Done when response completes
elif event['type'] == 'response.done':
break
print(f"\n\n✅ Full response: {''.join(transcript)}")
if __name__ == "__main__":
asyncio.run(run_voice_agent())
```
</TabItem>
<TabItem value="livekit" label="LiveKit Agent">
```python
from livekit.agents import Agent, AgentSession, WorkerOptions, cli
from livekit.plugins import xai
class VoiceAgent(Agent):
def __init__(self):
super().__init__(
instructions="You are a helpful voice assistant.",
llm=xai.realtime.RealtimeModel(
voice="ara",
api_key="sk-1234",
base_url="http://localhost:4000",
),
)
if __name__ == "__main__":
cli.run_app(
WorkerOptions(
agent_factory=VoiceAgent,
)
)
```
</TabItem>
</Tabs>
## Running the Example
1. **Start LiteLLM Proxy** (if not already running):
```bash
litellm --config config.yaml --port 4000
```
2. **Run the example**:
```bash
python your_script.py
```
## Expected Output
```
✅ Connected: conversation.created
Hello! Here's a joke for you: Why don't scientists trust atoms?
Because they make up everything!
✅ Full response: Hello! Here's a joke for you: Why don't scientists trust atoms? Because they make up everything!
```
## Complete Working Example (Cookbook)
See the full working example in our [LiveKit Agent SDK Cookbook](https://github.com/BerriAI/litellm/tree/main/cookbook/livekit_agent_sdk).
## Learn More
- [xAI Realtime API Documentation](/docs/providers/xai_realtime)
- [LiveKit xAI Plugin Docs](https://docs.livekit.io/agents/models/realtime/plugins/xai/)
- [LiteLLM Realtime API](/docs/realtime)

View file

@ -151,6 +151,7 @@ const sidebars = {
items: [
"tutorials/claude_agent_sdk",
"tutorials/google_adk",
"tutorials/livekit_xai_realtime",
]
},