wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add GrokRealtimeLLMService for xAI Grok Voice Agent API (#3267)

Browse Source
This commit is contained in:
Mrunmay Chichkhede
2025-12-20 17:34:12 +05:30
committed by GitHub
parent 76bae6e699
commit d7d979dde1
7 changed files with 2375 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
changelog/3267.added.md Normal file
View File

@@ -0,0 +1,8 @@
- Added `GrokRealtimeLLMService` for xAI's Grok Voice Agent API with real-time voice conversations:
- Support for real-time audio streaming with WebSocket connection
- Built-in server-side VAD (Voice Activity Detection)
- Multiple voice options: Ara, Rex, Sal, Eve, Leo
- Built-in tools support: web_search, x_search, file_search
- Custom function calling with standard Pipecat tools schema
- Configurable audio formats (PCM at 8kHz-48kHz)

249
examples/foundational/20f-persistent-context-grok-realtime.py Normal file
View File

@@ -0,0 +1,249 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Grok Realtime persistent context example.
This example demonstrates how to save and load conversation history with
Grok's Realtime Voice Agent API. It allows:
- Saving the current conversation to a JSON file
- Loading a previous conversation from disk
- Listing all saved conversation files
This is useful for building voice agents that remember past conversations.
"""
import asyncio
import glob
import json
import os
from datetime import datetime
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.grok.realtime.events import SessionProperties, TurnDetection
from pipecat.services.grok.realtime.llm import GrokRealtimeLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
BASE_FILENAME = "/tmp/pipecat_grok_conversation_"
async def fetch_weather_from_api(params: FunctionCallParams):
"""Mock weather function for demonstration."""
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_saved_conversation_filenames(params: FunctionCallParams):
"""Get a list of saved conversation history files."""
full_pattern = f"{BASE_FILENAME}*.json"
matching_files = glob.glob(full_pattern)
logger.debug(f"matching files: {matching_files}")
await params.result_callback({"filenames": matching_files})
async def save_conversation(params: FunctionCallParams):
"""Save the current conversation to a JSON file."""
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(
f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
)
try:
with open(filename, "w") as file:
messages = params.context.get_messages()
# Remove the last message (the save instruction)
messages.pop()
json.dump(messages, file, indent=2)
await params.result_callback({"success": True})
except Exception as e:
await params.result_callback({"success": False, "error": str(e)})
async def load_conversation(params: FunctionCallParams):
"""Load a conversation history from a JSON file."""
async def _reset():
filename = params.arguments["filename"]
logger.debug(f"loading conversation from {filename}")
try:
with open(filename, "r") as file:
params.context.set_messages(json.load(file))
await params.llm.reset_conversation()
# Manually create a response since we've reset the conversation
await params.llm._create_response()
except Exception as e:
await params.result_callback({"success": False, "error": str(e)})
asyncio.create_task(_reset())
# Define the tools schema
tools = ToolsSchema(
standard_tools=[
FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the users location.",
},
},
required=["location", "format"],
),
FunctionSchema(
name="save_conversation",
description="Save the current conversation. Use this function to persist the current conversation to external storage.",
properties={},
required=[],
),
FunctionSchema(
name="get_saved_conversation_filenames",
description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp.",
properties={},
required=[],
),
FunctionSchema(
name="load_conversation",
description="Load a conversation history. Use this function to load a conversation history into the current session.",
properties={
"filename": {
"type": "string",
"description": "The filename of the conversation history to load.",
}
},
required=["filename"],
),
]
)
# Transport configuration - no local VAD needed since Grok has server-side VAD
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info("Starting Grok Realtime persistent context bot")
session_properties = SessionProperties(
voice="Ara",
turn_detection=TurnDetection(type="server_vad"),
instructions="""You are a helpful and friendly AI assistant powered by Grok.
Your voice and personality should be warm and engaging, with a lively and playful tone.
You are participating in a voice conversation. Keep your responses concise, short, and to the point
unless specifically asked to elaborate on a topic.
You have access to tools for:
- Getting weather information
- Saving the current conversation to disk
- Loading previous conversations from disk
- Listing saved conversation files
When the user asks to save or load a conversation, use the appropriate tool.
Remember, your responses should be short - just one or two sentences usually.""",
)
llm = GrokRealtimeLLMService(
api_key=os.getenv("GROK_API_KEY"),
session_properties=session_properties,
start_audio_paused=False,
)
# Register function handlers
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("save_conversation", save_conversation)
llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
llm.register_function("load_conversation", load_conversation)
context = LLMContext([{"role": "user", "content": "Say hello!"}], tools)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
llm,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected")
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

287
examples/foundational/51-grok-realtime.py Normal file
View File

@@ -0,0 +1,287 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""
Grok Voice Agent Realtime Example
This example demonstrates using xAI's Grok Voice Agent API for real-time
voice conversations. The Grok Voice Agent provides:
- Real-time audio streaming with low latency
- Built-in voice activity detection (VAD)
- Multiple voice options (Ara, Rex, Sal, Eve, Leo)
- Built-in tools: web_search, x_search, file_search
- Custom function calling
Requirements:
- XAI_API_KEY environment variable set
- pip install pipecat-ai[grok]
Usage:
python 50-grok-realtime.py --transport webrtc
python 50-grok-realtime.py --transport daily
"""
import asyncio
import os
from datetime import datetime
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
# Note: Grok has built-in server-side VAD, so we don't need local VAD
# from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, TranscriptionMessage
from pipecat.observers.loggers.transcription_log_observer import (
TranscriptionLogObserver,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
)
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.grok.realtime.events import (
SessionProperties,
TurnDetection,
WebSearchTool,
XSearchTool,
)
from pipecat.services.grok.realtime.llm import GrokRealtimeLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
# --- Function Handlers ---
async def fetch_weather_from_api(params: FunctionCallParams):
"""Handle weather function calls."""
temperature = 75 if params.arguments.get("format") == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": params.arguments.get("format", "celsius"),
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_current_time(params: FunctionCallParams):
"""Handle time function calls."""
await params.result_callback(
{
"time": datetime.now().strftime("%H:%M:%S"),
"date": datetime.now().strftime("%Y-%m-%d"),
"timezone": "local",
}
)
async def get_restaurant_recommendation(params: FunctionCallParams):
"""Handle restaurant recommendation function calls."""
location = params.arguments.get("location", "unknown")
await params.result_callback(
{
"name": "The Golden Dragon",
"cuisine": "Chinese",
"location": location,
"rating": 4.5,
}
)
# --- Function Schemas ---
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather for a location",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use.",
},
},
required=["location", "format"],
)
time_function = FunctionSchema(
name="get_current_time",
description="Get the current time and date",
properties={},
required=[],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation for a location",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
required=["location"],
)
# Create tools schema with custom functions
tools = ToolsSchema(standard_tools=[weather_function, time_function, restaurant_function])
# --- Transport Configuration ---
# Note: We don't need local VAD since Grok has built-in server-side VAD.
# Audio sample rates are configured via PipelineParams, not transport params.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info("Starting Grok Voice Agent bot")
# Configure Grok session properties
session_properties = SessionProperties(
# Voice options: Ara (warm, friendly), Rex (confident), Sal (smooth),
# Eve (energetic), Leo (authoritative)
voice="Ara",
# Enable server-side VAD for automatic turn detection
turn_detection=TurnDetection(type="server_vad"),
# System instructions
instructions="""You are a helpful and friendly AI assistant powered by Grok.
You have access to several tools:
- Weather information
- Current time
- Restaurant recommendations
- Web search (built-in)
- X/Twitter search (built-in)
Your voice and personality should be warm and engaging. Keep your responses
concise and conversational since this is a voice interaction.
If the user asks about current events or news, use web search.
If they ask about what people are saying on social media, use X search.
Always be helpful and proactive in offering assistance.""",
# Grok-specific built-in tools can be added here:
# tools=[
# WebSearchTool(), # Enable web search
# XSearchTool(), # Enable X/Twitter search
# ],
)
# Create the Grok Realtime LLM service
llm = GrokRealtimeLLMService(
api_key=os.getenv("GROK_API_KEY"),
session_properties=session_properties,
start_audio_paused=False,
)
# Register function handlers
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_current_time", get_current_time)
llm.register_function("get_restaurant_recommendation", get_restaurant_recommendation)
# Create transcript processor for logging
transcript = TranscriptProcessor()
# Create context with initial message and tools
context = LLMContext(
[{"role": "user", "content": "Say hello and introduce yourself!"}],
tools,
)
context_aggregator = LLMContextAggregatorPair(context)
# Build the pipeline
# Note: In realtime mode, transcription comes from Grok (upstream),
# so transcript.user() goes BEFORE llm
pipeline = Pipeline(
[
transport.input(), # Transport user input (audio)
context_aggregator.user(),
transcript.user(), # Transcription from Grok goes upstream
llm, # Grok Realtime LLM (handles STT + LLM + TTS)
transport.output(), # Transport bot output (audio)
transcript.assistant(), # Log assistant speech
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
observers=[TranscriptionLogObserver()],
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected")
# Kick off the conversation
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected")
await task.cancel()
# Log transcript updates
@transcript.event_handler("on_transcript_update")
async def on_transcript_update(processor, frame):
for msg in frame.messages:
if isinstance(msg, TranscriptionMessage):
timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
line = f"{timestamp}{msg.role}: {msg.content}"
logger.info(f"Transcript: {line}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

253
src/pipecat/adapters/services/grok_realtime_adapter.py Normal file
View File

@@ -0,0 +1,253 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Grok Realtime LLM adapter for Pipecat.
Converts Pipecat's tool schemas and context into the format required by
Grok's Voice Agent API.
"""
import copy
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, TypedDict
from loguru import logger
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
from pipecat.services.grok.realtime import events
class GrokRealtimeLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking Grok Realtime API.
Attributes:
system_instruction: System prompt/instructions for the session.
messages: List of conversation items formatted for Grok Realtime.
tools: List of tool definitions (function, web_search, x_search, file_search).
"""
system_instruction: Optional[str]
messages: List[events.ConversationItem]
tools: List[Dict[str, Any]]
class GrokRealtimeLLMAdapter(BaseLLMAdapter):
"""LLM adapter for Grok Voice Agent API.
Converts Pipecat's universal context and tool schemas into the specific
format required by Grok's Voice Agent Realtime API.
"""
@property
def id_for_llm_specific_messages(self) -> str:
"""Get the identifier used in LLMSpecificMessage instances for Grok Realtime."""
return "grok-realtime"
def get_llm_invocation_params(self, context: LLMContext) -> GrokRealtimeLLMInvocationParams:
"""Get Grok Realtime-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for invoking Grok's Voice Agent API.
"""
messages = self._from_universal_context_messages(self.get_messages(context))
return {
"system_instruction": messages.system_instruction,
"messages": messages.messages,
"tools": self.from_standard_tools(context.tools) or [],
}
def get_messages_for_logging(self, context) -> List[Dict[str, Any]]:
"""Get messages from context in a format safe for logging.
Removes or truncates sensitive data like audio content.
Args:
context: The LLM context containing messages.
Returns:
List of messages with sensitive data redacted.
"""
msgs = []
for message in self.get_messages(context):
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item.get("type") == "input_audio":
item["audio"] = "..."
if item.get("type") == "audio":
item["audio"] = "..."
msgs.append(msg)
return msgs
@dataclass
class ConvertedMessages:
"""Container for Grok-formatted messages converted from universal context."""
messages: List[events.ConversationItem]
system_instruction: Optional[str] = None
def _from_universal_context_messages(
self, universal_context_messages: List[LLMContextMessage]
) -> ConvertedMessages:
"""Convert universal context messages to Grok Realtime format.
Similar to OpenAI Realtime, we pack conversation history into a single
user message since the realtime API doesn't support loading long histories.
Args:
universal_context_messages: List of messages in universal format.
Returns:
ConvertedMessages with Grok-formatted messages and system instruction.
"""
if not universal_context_messages:
return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages)
system_instruction = None
# Extract system message as session instructions
if messages[0].get("role") == "system":
system = messages.pop(0)
content = system.get("content")
if isinstance(content, str):
system_instruction = content
elif isinstance(content, list):
system_instruction = content[0].get("text")
if not messages:
return self.ConvertedMessages(messages=[], system_instruction=system_instruction)
# Single user message can be sent normally
if len(messages) == 1 and messages[0].get("role") == "user":
return self.ConvertedMessages(
messages=[self._from_universal_context_message(messages[0])],
system_instruction=system_instruction,
)
# Pack multiple messages into a single user message
intro_text = """
This is a previously saved conversation. Please treat this conversation history as a
starting point for the current conversation."""
trailing_text = """
This is the end of the previously saved conversation. Please continue the conversation
from here. If the last message is a user instruction or question, act on that instruction
or answer the question. If the last message is an assistant response, simply say that you
are ready to continue the conversation."""
return self.ConvertedMessages(
messages=[
events.ConversationItem(
role="user",
type="message",
content=[
events.ItemContent(
type="input_text",
text="\n\n".join(
[
intro_text,
json.dumps(messages, indent=2),
trailing_text,
]
),
)
],
)
],
system_instruction=system_instruction,
)
def _from_universal_context_message(
self, message: LLMContextMessage
) -> events.ConversationItem:
"""Convert a single universal context message to Grok format.
Args:
message: Message in universal format.
Returns:
ConversationItem formatted for Grok Realtime API.
"""
if message.get("role") == "user":
content = message.get("content")
if isinstance(content, list):
text_content = ""
for c in content:
if c.get("type") == "text":
text_content += " " + c.get("text")
else:
logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}"
)
content = text_content.strip()
return events.ConversationItem(
role="user",
type="message",
content=[events.ItemContent(type="input_text", text=content)],
)
if message.get("role") == "assistant" and message.get("tool_calls"):
tc = message.get("tool_calls")[0]
return events.ConversationItem(
type="function_call",
call_id=tc["id"],
name=tc["function"]["name"],
arguments=tc["function"]["arguments"],
)
logger.error(f"Unhandled message type in _from_universal_context_message: {message}")
@staticmethod
def _to_grok_function_format(function: FunctionSchema) -> Dict[str, Any]:
"""Convert a function schema to Grok Realtime function format.
Args:
function: The function schema to convert.
Returns:
Dictionary in Grok Realtime function format.
"""
return {
"type": "function",
"name": function.name,
"description": function.description,
"parameters": {
"type": "object",
"properties": function.properties,
"required": function.required,
},
}
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
"""Convert tool schemas to Grok Realtime format.
Supports both standard function tools and Grok-specific tools
(web_search, x_search, file_search).
Args:
tools_schema: The tools schema containing functions to convert.
Returns:
List of tool definitions in Grok Realtime format.
"""
# Convert standard function tools
functions_schema = tools_schema.standard_tools
standard_tools = [self._to_grok_function_format(func) for func in functions_schema]
# Support shimmed custom tools for backward compatibility
shimmed_tools = []
if tools_schema.custom_tools:
shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, [])
return standard_tools + shimmed_tools

5
src/pipecat/services/grok/realtime/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

847
src/pipecat/services/grok/realtime/events.py Normal file
View File

@@ -0,0 +1,847 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Event models and data structures for Grok Voice Agent API communication.
Based on xAI's Grok Voice Agent API documentation:
https://docs.x.ai/docs/guides/voice/agent
"""
import json
import uuid
from typing import Any, Dict, List, Literal, Optional, Union
from pydantic import BaseModel, ConfigDict, Field
from pipecat.adapters.schemas.tools_schema import ToolsSchema
#
# Audio format configuration
#
# Grok supports configurable sample rates for PCM audio
SUPPORTED_SAMPLE_RATES = Literal[8000, 16000, 21050, 24000, 32000, 44100, 48000]
class AudioFormat(BaseModel):
"""Base class for audio format configuration."""
type: str
class PCMAudioFormat(AudioFormat):
"""PCM audio format configuration with configurable sample rate.
Grok supports: 8000, 16000, 21050, 24000, 32000, 44100, 48000 Hz
Parameters:
type: Audio format type, always "audio/pcm".
rate: Sample rate in Hz. Defaults to 24000.
"""
type: Literal["audio/pcm"] = "audio/pcm"
rate: SUPPORTED_SAMPLE_RATES = 24000
class PCMUAudioFormat(AudioFormat):
"""PCMU (G.711 μ-law) audio format configuration.
Fixed at 8000 Hz sample rate.
Parameters:
type: Audio format type, always "audio/pcmu".
"""
type: Literal["audio/pcmu"] = "audio/pcmu"
class PCMAAudioFormat(AudioFormat):
"""PCMA (G.711 A-law) audio format configuration.
Fixed at 8000 Hz sample rate.
Parameters:
type: Audio format type, always "audio/pcma".
"""
type: Literal["audio/pcma"] = "audio/pcma"
#
# Turn detection configuration
#
class TurnDetection(BaseModel):
"""Server-side voice activity detection configuration.
Parameters:
type: Detection type, must be "server_vad" or None for manual.
"""
type: Optional[Literal["server_vad"]] = "server_vad"
#
# Audio configuration
#
class AudioInput(BaseModel):
"""Audio input configuration.
Parameters:
format: The format configuration for input audio.
"""
format: Optional[Union[PCMAudioFormat, PCMUAudioFormat, PCMAAudioFormat]] = None
class AudioOutput(BaseModel):
"""Audio output configuration.
Parameters:
format: The format configuration for output audio.
"""
format: Optional[Union[PCMAudioFormat, PCMUAudioFormat, PCMAAudioFormat]] = None
class AudioConfiguration(BaseModel):
"""Audio configuration for input and output.
Parameters:
input: Configuration for input audio.
output: Configuration for output audio.
"""
input: Optional[AudioInput] = None
output: Optional[AudioOutput] = None
#
# Tool definitions - Grok-specific tools
#
class WebSearchTool(BaseModel):
"""Web search tool configuration.
Enables the voice agent to search the web for current information.
"""
type: Literal["web_search"] = "web_search"
class XSearchTool(BaseModel):
"""X (Twitter) search tool configuration.
Enables the voice agent to search X for posts and information.
Parameters:
type: Tool type, always "x_search".
allowed_x_handles: Optional list of X handles to filter search results.
"""
type: Literal["x_search"] = "x_search"
allowed_x_handles: Optional[List[str]] = None
class FileSearchTool(BaseModel):
"""File/Collection search tool configuration.
Enables the voice agent to search through uploaded document collections.
Parameters:
type: Tool type, always "file_search".
vector_store_ids: List of collection IDs to search.
max_num_results: Maximum number of results to return.
"""
type: Literal["file_search"] = "file_search"
vector_store_ids: List[str]
max_num_results: Optional[int] = 10
class FunctionTool(BaseModel):
"""Custom function tool configuration.
Parameters:
type: Tool type, always "function".
name: Name of the function.
description: Description of what the function does.
parameters: JSON schema for function parameters.
"""
type: Literal["function"] = "function"
name: str
description: str
parameters: Dict[str, Any]
# Union type for all Grok tools
GrokTool = Union[WebSearchTool, XSearchTool, FileSearchTool, FunctionTool, Dict[str, Any]]
#
# Voice options
#
# Grok voice options: Ara (default), Rex, Sal, Eve, Leo
GrokVoice = Literal["Ara", "Rex", "Sal", "Eve", "Leo"]
#
# Session properties
#
class SessionProperties(BaseModel):
"""Configuration properties for a Grok Voice Agent session.
Parameters:
instructions: System instructions for the assistant.
voice: The voice the model uses to respond. Options: Ara, Rex, Sal, Eve, Leo.
turn_detection: Configuration for turn detection, or None for manual.
audio: Configuration for input and output audio.
tools: Available tools for the assistant (web_search, x_search, file_search, function).
"""
# Needed to support ToolSchema in tools field.
model_config = ConfigDict(arbitrary_types_allowed=True)
instructions: Optional[str] = None
voice: Optional[GrokVoice] = "Ara"
turn_detection: Optional[TurnDetection] = None
audio: Optional[AudioConfiguration] = None
# Tools can be ToolsSchema when provided by user, or list of dicts for API
tools: Optional[ToolsSchema | List[GrokTool]] = None
#
# Conversation items
#
class ItemContent(BaseModel):
"""Content within a conversation item.
Parameters:
type: Content type (input_text, input_audio, text, audio).
text: Text content for text-based items.
audio: Base64-encoded audio data for audio items.
transcript: Transcribed text for audio items.
"""
type: Literal["text", "audio", "input_text", "input_audio", "output_text", "output_audio"]
text: Optional[str] = None
audio: Optional[str] = None # base64-encoded audio
transcript: Optional[str] = None
class ConversationItem(BaseModel):
"""A conversation item in the realtime session.
Parameters:
id: Unique identifier for the item, auto-generated if not provided.
object: Object type identifier for the realtime API.
type: Item type (message, function_call, or function_call_output).
status: Current status of the item.
role: Speaker role for message items (user, assistant, or system).
content: Content list for message items.
call_id: Function call identifier for function_call items.
name: Function name for function_call items.
arguments: Function arguments as JSON string for function_call items.
output: Function output as JSON string for function_call_output items.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4().hex))
object: Optional[Literal["realtime.item"]] = None
type: Literal["message", "function_call", "function_call_output"]
status: Optional[Literal["completed", "in_progress", "incomplete"]] = None
role: Optional[Literal["user", "assistant", "system", "tool"]] = None
content: Optional[List[ItemContent]] = None
call_id: Optional[str] = None
name: Optional[str] = None
arguments: Optional[str] = None
output: Optional[str] = None
class RealtimeConversation(BaseModel):
"""A realtime conversation session.
Parameters:
id: Unique identifier for the conversation.
object: Object type identifier, always "realtime.conversation".
"""
id: str
object: Literal["realtime.conversation"]
class ResponseProperties(BaseModel):
"""Properties for configuring assistant responses.
Parameters:
modalities: Output modalities for the response (text, audio, or both).
"""
modalities: Optional[List[Literal["text", "audio"]]] = ["text", "audio"]
#
# Error class
#
class RealtimeError(BaseModel):
"""Error information from the realtime API.
Parameters:
type: Error type identifier.
code: Specific error code.
message: Human-readable error message.
param: Parameter name that caused the error, if applicable.
event_id: Event ID associated with the error, if applicable.
"""
type: Optional[str] = None
code: Optional[str] = ""
message: str
param: Optional[str] = None
event_id: Optional[str] = None
#
# Client Events (sent to Grok)
#
class ClientEvent(BaseModel):
"""Base class for client events sent to the realtime API.
Parameters:
event_id: Unique identifier for the event, auto-generated if not provided.
"""
event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class SessionUpdateEvent(ClientEvent):
"""Event to update session properties.
Parameters:
type: Event type, always "session.update".
session: Updated session properties.
"""
type: Literal["session.update"] = "session.update"
session: SessionProperties
class InputAudioBufferAppendEvent(ClientEvent):
"""Event to append audio data to the input buffer.
Parameters:
type: Event type, always "input_audio_buffer.append".
audio: Base64-encoded audio data to append.
"""
type: Literal["input_audio_buffer.append"] = "input_audio_buffer.append"
audio: str # base64-encoded audio
class InputAudioBufferCommitEvent(ClientEvent):
"""Event to commit the current input audio buffer.
Used when turn_detection is null (manual mode).
Parameters:
type: Event type, always "input_audio_buffer.commit".
"""
type: Literal["input_audio_buffer.commit"] = "input_audio_buffer.commit"
class InputAudioBufferClearEvent(ClientEvent):
"""Event to clear the input audio buffer.
Parameters:
type: Event type, always "input_audio_buffer.clear".
"""
type: Literal["input_audio_buffer.clear"] = "input_audio_buffer.clear"
class ConversationItemCreateEvent(ClientEvent):
"""Event to create a new conversation item.
Parameters:
type: Event type, always "conversation.item.create".
previous_item_id: ID of the item to insert after, if any.
item: The conversation item to create.
"""
type: Literal["conversation.item.create"] = "conversation.item.create"
previous_item_id: Optional[str] = None
item: ConversationItem
class ResponseCreateEvent(ClientEvent):
"""Event to create a new assistant response.
Parameters:
type: Event type, always "response.create".
response: Optional response configuration properties.
"""
type: Literal["response.create"] = "response.create"
response: Optional[ResponseProperties] = None
class ResponseCancelEvent(ClientEvent):
"""Event to cancel the current assistant response.
Parameters:
type: Event type, always "response.cancel".
"""
type: Literal["response.cancel"] = "response.cancel"
#
# Server Events (received from Grok)
#
class ServerEvent(BaseModel):
"""Base class for server events received from the realtime API.
Parameters:
event_id: Unique identifier for the event.
type: Type of the server event.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
event_id: str
type: str
class SessionUpdatedEvent(ServerEvent):
"""Event indicating a session has been updated.
Parameters:
type: Event type, always "session.updated".
session: The updated session properties.
"""
type: Literal["session.updated"]
session: SessionProperties
class ConversationCreated(ServerEvent):
"""Event indicating a conversation has been created.
This is the first message received after connecting.
Parameters:
type: Event type, always "conversation.created".
conversation: The created conversation.
"""
type: Literal["conversation.created"]
conversation: RealtimeConversation
class ConversationItemAdded(ServerEvent):
"""Event indicating a conversation item has been added.
Parameters:
type: Event type, always "conversation.item.added".
previous_item_id: ID of the previous item, if any.
item: The added conversation item.
"""
type: Literal["conversation.item.added"]
previous_item_id: Optional[str] = None
item: ConversationItem
class ConversationItemInputAudioTranscriptionCompleted(ServerEvent):
"""Event indicating input audio transcription is complete.
Parameters:
type: Event type, always "conversation.item.input_audio_transcription.completed".
item_id: ID of the conversation item that was transcribed.
transcript: Complete transcription text.
"""
type: Literal["conversation.item.input_audio_transcription.completed"]
item_id: str
transcript: str
class InputAudioBufferSpeechStarted(ServerEvent):
"""Event indicating speech has started in the input audio buffer.
Only sent when turn_detection is "server_vad".
Parameters:
type: Event type, always "input_audio_buffer.speech_started".
item_id: ID of the associated conversation item.
"""
type: Literal["input_audio_buffer.speech_started"]
item_id: str
class InputAudioBufferSpeechStopped(ServerEvent):
"""Event indicating speech has stopped in the input audio buffer.
Only sent when turn_detection is "server_vad".
Parameters:
type: Event type, always "input_audio_buffer.speech_stopped".
item_id: ID of the associated conversation item.
"""
type: Literal["input_audio_buffer.speech_stopped"]
item_id: str
class InputAudioBufferCommitted(ServerEvent):
"""Event indicating the input audio buffer has been committed.
Parameters:
type: Event type, always "input_audio_buffer.committed".
previous_item_id: ID of the previous item, if any.
item_id: ID of the committed conversation item.
"""
type: Literal["input_audio_buffer.committed"]
previous_item_id: Optional[str] = None
item_id: str
class InputAudioBufferCleared(ServerEvent):
"""Event indicating the input audio buffer has been cleared.
Parameters:
type: Event type, always "input_audio_buffer.cleared".
"""
type: Literal["input_audio_buffer.cleared"]
class ResponseCreated(ServerEvent):
"""Event indicating an assistant response has been created.
Parameters:
type: Event type, always "response.created".
response: The created response object.
"""
type: Literal["response.created"]
response: "Response"
class ResponseOutputItemAdded(ServerEvent):
"""Event indicating an output item has been added to a response.
Parameters:
type: Event type, always "response.output_item.added".
response_id: ID of the response.
output_index: Index of the output item.
item: The added conversation item.
"""
type: Literal["response.output_item.added"]
response_id: str
output_index: int
item: ConversationItem
class ResponseAudioTranscriptDelta(ServerEvent):
"""Event containing incremental audio transcript from a response.
Parameters:
type: Event type, always "response.output_audio_transcript.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
delta: Incremental transcript text.
"""
type: Literal["response.output_audio_transcript.delta"]
response_id: str
item_id: str
delta: str
class ResponseAudioTranscriptDone(ServerEvent):
"""Event indicating audio transcript is complete.
Parameters:
type: Event type, always "response.output_audio_transcript.done".
response_id: ID of the response.
item_id: ID of the conversation item.
"""
type: Literal["response.output_audio_transcript.done"]
response_id: str
item_id: str
class ResponseAudioDelta(ServerEvent):
"""Event containing incremental audio data from a response.
Parameters:
type: Event type, always "response.output_audio.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
delta: Base64-encoded incremental audio data.
"""
type: Literal["response.output_audio.delta"]
response_id: str
item_id: str
output_index: int
content_index: int
delta: str # base64-encoded audio
class ResponseAudioDone(ServerEvent):
"""Event indicating audio content is complete.
Parameters:
type: Event type, always "response.output_audio.done".
response_id: ID of the response.
item_id: ID of the conversation item.
"""
type: Literal["response.output_audio.done"]
response_id: str
item_id: str
class ResponseFunctionCallArgumentsDone(ServerEvent):
"""Event indicating function call arguments are complete.
Parameters:
type: Event type, always "response.function_call_arguments.done".
call_id: ID of the function call.
name: Name of the function being called.
arguments: Complete function arguments as JSON string.
"""
type: Literal["response.function_call_arguments.done"]
call_id: str
name: str
arguments: str
class Usage(BaseModel):
"""Token usage statistics for a response.
All fields are optional because Grok sends empty usage in some events.
Parameters:
total_tokens: Total number of tokens used.
input_tokens: Number of input tokens used.
output_tokens: Number of output tokens used.
"""
total_tokens: Optional[int] = None
input_tokens: Optional[int] = None
output_tokens: Optional[int] = None
class Response(BaseModel):
"""A complete assistant response.
Parameters:
id: Unique identifier for the response.
object: Object type, always "realtime.response".
status: Current status of the response.
output: List of conversation items in the response.
usage: Token usage statistics for the response.
"""
id: str
object: Literal["realtime.response"]
status: Literal["completed", "in_progress", "incomplete", "cancelled", "failed"]
status_details: Optional[Any] = None
output: List[ConversationItem]
usage: Optional[Usage] = None
class ResponseCreated(ServerEvent):
"""Event indicating an assistant response has been created.
Parameters:
type: Event type, always "response.created".
response: The created response object.
"""
type: Literal["response.created"]
response: Response
class ResponseDone(ServerEvent):
"""Event indicating an assistant response is complete.
Parameters:
type: Event type, always "response.done".
response: The completed response object.
usage: Token usage (also available at top level in Grok).
"""
type: Literal["response.done"]
response: Response
usage: Optional[Usage] = None
class ResponseOutputItemDone(ServerEvent):
"""Event indicating an output item is complete.
Parameters:
type: Event type, always "response.output_item.done".
response_id: ID of the response.
output_index: Index of the output item.
item: The completed conversation item.
"""
type: Literal["response.output_item.done"]
response_id: str
output_index: int
item: ConversationItem
class ContentPart(BaseModel):
"""A content part within a response.
Parameters:
type: Type of the content part (audio, text).
transcript: Transcript text if applicable.
"""
type: str
transcript: Optional[str] = None
class ResponseContentPartAdded(ServerEvent):
"""Event indicating a content part has been added to a response.
Parameters:
type: Event type, always "response.content_part.added".
response_id: ID of the response.
item_id: ID of the conversation item.
content_index: Index of the content part.
output_index: Index of the output item.
part: The added content part.
"""
type: Literal["response.content_part.added"]
response_id: str
item_id: str
content_index: int
output_index: int
part: ContentPart
class ResponseContentPartDone(ServerEvent):
"""Event indicating a content part is complete.
Parameters:
type: Event type, always "response.content_part.done".
response_id: ID of the response.
item_id: ID of the conversation item.
content_index: Index of the content part.
output_index: Index of the output item.
"""
type: Literal["response.content_part.done"]
response_id: str
item_id: str
content_index: int
output_index: int
class PingEvent(ServerEvent):
"""Keep-alive ping event from the server.
Parameters:
type: Event type, always "ping".
timestamp: Server timestamp in milliseconds.
"""
type: Literal["ping"]
timestamp: int
class ErrorEvent(ServerEvent):
"""Event indicating an error occurred.
Parameters:
type: Event type, always "error".
error: Error details.
"""
type: Literal["error"]
error: RealtimeError
#
# Event parsing
#
_server_event_types = {
"error": ErrorEvent,
"ping": PingEvent,
"session.updated": SessionUpdatedEvent,
"conversation.created": ConversationCreated,
"conversation.item.added": ConversationItemAdded,
"conversation.item.input_audio_transcription.completed": ConversationItemInputAudioTranscriptionCompleted,
"input_audio_buffer.speech_started": InputAudioBufferSpeechStarted,
"input_audio_buffer.speech_stopped": InputAudioBufferSpeechStopped,
"input_audio_buffer.committed": InputAudioBufferCommitted,
"input_audio_buffer.cleared": InputAudioBufferCleared,
"response.created": ResponseCreated,
"response.output_item.added": ResponseOutputItemAdded,
"response.output_item.done": ResponseOutputItemDone,
"response.content_part.added": ResponseContentPartAdded,
"response.content_part.done": ResponseContentPartDone,
"response.output_audio_transcript.delta": ResponseAudioTranscriptDelta,
"response.output_audio_transcript.done": ResponseAudioTranscriptDone,
"response.output_audio.delta": ResponseAudioDelta,
"response.output_audio.done": ResponseAudioDone,
"response.function_call_arguments.done": ResponseFunctionCallArgumentsDone,
"response.done": ResponseDone,
}
def parse_server_event(data: str):
"""Parse a server event from JSON string.
Args:
data: JSON string containing the server event.
Returns:
Parsed server event object of the appropriate type.
Raises:
Exception: If the event type is unimplemented or parsing fails.
"""
try:
event = json.loads(data)
event_type = event["type"]
if event_type not in _server_event_types:
raise Exception(f"Unimplemented server event type: {event_type}")
return _server_event_types[event_type].model_validate(event)
except Exception as e:
raise Exception(f"{e} \n\n{data}")

726
src/pipecat/services/grok/realtime/llm.py Normal file
View File

@@ -0,0 +1,726 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Grok Realtime Voice Agent LLM service implementation with WebSocket support.
Based on xAI's Grok Voice Agent API documentation:
https://docs.x.ai/docs/guides/voice/agent
"""
import base64
import json
import time
from dataclasses import dataclass
from typing import Optional
from loguru import logger
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.grok_realtime_adapter import GrokRealtimeLLMAdapter
from pipecat.frames.frames import (
AggregationType,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
Frame,
InputAudioRawFrame,
InterimTranscriptionFrame,
InterruptionFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
LLMSetToolsFrame,
LLMTextFrame,
LLMUpdateSettingsFrame,
StartFrame,
TranscriptionFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
TTSTextFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from . import events
try:
from websockets.asyncio.client import connect as websocket_connect
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Grok Realtime, you need to `pip install pipecat-ai[grok]`.")
raise Exception(f"Missing module: {e}")
@dataclass
class CurrentAudioResponse:
"""Tracks the current audio response from the assistant.
Parameters:
item_id: Unique identifier for the audio response item.
content_index: Index of the audio content within the item.
start_time_ms: Timestamp when the audio response started in milliseconds.
total_size: Total size of audio data received in bytes. Defaults to 0.
"""
item_id: str
content_index: int
start_time_ms: int
total_size: int = 0
class GrokRealtimeLLMService(LLMService):
"""Grok Realtime Voice Agent LLM service providing real-time audio and text communication.
Implements the Grok Voice Agent API with WebSocket communication for low-latency
bidirectional audio and text interactions. Supports function calling, conversation
management, and real-time transcription.
Features:
- Real-time audio streaming (PCM, PCMU, PCMA formats)
- Configurable sample rates (8kHz to 48kHz for PCM)
- Multiple voice options (Ara, Rex, Sal, Eve, Leo)
- Built-in tools (web_search, x_search, file_search)
- Custom function calling
- Server-side VAD (Voice Activity Detection)
"""
# Use the Grok-specific adapter
adapter_class = GrokRealtimeLLMAdapter
def __init__(
self,
*,
api_key: str,
voice: events.GrokVoice = "Ara",
base_url: str = "wss://api.x.ai/v1/realtime",
session_properties: Optional[events.SessionProperties] = None,
start_audio_paused: bool = False,
sample_rate: int = 24000,
**kwargs,
):
"""Initialize the Grok Realtime Voice Agent LLM service.
Args:
api_key: xAI API key for authentication.
voice: Voice to use for responses. Options: Ara, Rex, Sal, Eve, Leo.
Defaults to "Ara".
base_url: WebSocket base URL for the realtime API.
Defaults to "wss://api.x.ai/v1/realtime".
session_properties: Configuration properties for the realtime session.
If None, uses default SessionProperties with the specified voice.
start_audio_paused: Whether to start with audio input paused. Defaults to False.
sample_rate: Audio sample rate in Hz. Supported: 8000, 16000, 21050, 24000,
32000, 44100, 48000. Defaults to 24000.
**kwargs: Additional arguments passed to parent LLMService.
"""
super().__init__(base_url=base_url, **kwargs)
self.api_key = api_key
self.base_url = base_url
self._sample_rate = sample_rate
self._voice = voice
# Initialize session_properties with voice and audio config
if session_properties:
self._session_properties = session_properties
# Ensure voice is set
if not self._session_properties.voice:
self._session_properties.voice = voice
else:
self._session_properties = events.SessionProperties(
voice=voice,
turn_detection=events.TurnDetection(type="server_vad"),
audio=events.AudioConfiguration(
input=events.AudioInput(format=events.PCMAudioFormat(rate=sample_rate)),
output=events.AudioOutput(format=events.PCMAudioFormat(rate=sample_rate)),
),
)
self._audio_input_paused = start_audio_paused
self._websocket = None
self._receive_task = None
self._context: LLMContext = None
self._llm_needs_conversation_setup = True
self._disconnecting = False
self._api_session_ready = False
self._run_llm_when_api_session_ready = False
self._current_assistant_response = None
self._current_audio_response = None
self._messages_added_manually = {}
self._pending_function_calls = {}
self._completed_tool_calls = set()
self._register_event_handler("on_conversation_item_created")
self._register_event_handler("on_conversation_item_updated")
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True if metrics generation is supported.
"""
return True
def set_audio_input_paused(self, paused: bool):
"""Set whether audio input is paused.
Args:
paused: True to pause audio input, False to resume.
"""
self._audio_input_paused = paused
def _is_turn_detection_enabled(self) -> bool:
"""Check if server-side VAD is enabled."""
if self._session_properties.turn_detection:
return self._session_properties.turn_detection.type == "server_vad"
return False
async def _handle_interruption(self):
"""Handle user interruption of assistant speech."""
if not self._is_turn_detection_enabled():
await self.send_client_event(events.InputAudioBufferClearEvent())
await self.send_client_event(events.ResponseCancelEvent())
await self._truncate_current_audio_response()
await self.stop_all_metrics()
if self._current_assistant_response:
await self.push_frame(LLMFullResponseEndFrame())
await self.push_frame(TTSStoppedFrame())
async def _handle_user_started_speaking(self, frame):
"""Handle user started speaking event."""
pass
async def _handle_user_stopped_speaking(self, frame):
"""Handle user stopped speaking event."""
if not self._is_turn_detection_enabled():
await self.send_client_event(events.InputAudioBufferCommitEvent())
await self.send_client_event(events.ResponseCreateEvent())
async def _handle_bot_stopped_speaking(self):
"""Handle bot stopped speaking event."""
self._current_audio_response = None
def _calculate_audio_duration_ms(
self, total_bytes: int, sample_rate: int = None, bytes_per_sample: int = 2
) -> int:
"""Calculate audio duration in milliseconds based on PCM audio parameters."""
if sample_rate is None:
sample_rate = self._sample_rate
samples = total_bytes / bytes_per_sample
duration_seconds = samples / sample_rate
return int(duration_seconds * 1000)
async def _truncate_current_audio_response(self):
"""Truncates the current audio response.
Note: Grok may not support truncation events like OpenAI.
This is a best-effort cleanup.
"""
if not self._current_audio_response:
return
try:
self._current_audio_response = None
except Exception as e:
logger.warning(f"Audio truncation cleanup failed (non-fatal): {e}")
#
# Standard AIService frame handling
#
async def start(self, frame: StartFrame):
"""Start the service and establish WebSocket connection.
Args:
frame: The start frame triggering service initialization.
"""
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the service and close WebSocket connection.
Args:
frame: The end frame triggering service shutdown.
"""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel the service and close WebSocket connection.
Args:
frame: The cancel frame triggering service cancellation.
"""
await super().cancel(frame)
await self._disconnect()
#
# Frame processing
#
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames from the pipeline.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
if isinstance(frame, TranscriptionFrame):
pass
elif isinstance(frame, LLMContextFrame):
await self._handle_context(frame.context)
elif isinstance(frame, InputAudioRawFrame):
if not self._audio_input_paused:
await self._send_user_audio(frame)
elif isinstance(frame, InterruptionFrame):
await self._handle_interruption()
elif isinstance(frame, UserStartedSpeakingFrame):
await self._handle_user_started_speaking(frame)
elif isinstance(frame, UserStoppedSpeakingFrame):
await self._handle_user_stopped_speaking(frame)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._handle_bot_stopped_speaking()
elif isinstance(frame, LLMMessagesAppendFrame):
await self._handle_messages_append(frame)
elif isinstance(frame, LLMUpdateSettingsFrame):
self._session_properties = events.SessionProperties(**frame.settings)
await self._update_settings()
elif isinstance(frame, LLMSetToolsFrame):
await self._update_settings()
await self.push_frame(frame, direction)
async def _handle_context(self, context: LLMContext):
"""Handle LLM context updates."""
if not self._context:
self._context = context
await self._process_completed_function_calls(send_new_results=False)
await self._create_response()
else:
self._context = context
await self._process_completed_function_calls(send_new_results=True)
async def _handle_messages_append(self, frame):
"""Handle appending messages to the context."""
logger.warning("LLMMessagesAppendFrame not yet implemented for Grok Realtime")
#
# WebSocket communication
#
async def send_client_event(self, event: events.ClientEvent):
"""Send a client event to the Grok Voice Agent API.
Args:
event: The client event to send.
"""
await self._ws_send(event.model_dump(exclude_none=True))
async def _connect(self):
"""Establish WebSocket connection to Grok."""
try:
if self._websocket:
return
self._websocket = await websocket_connect(
uri=self.base_url,
additional_headers={
"Authorization": f"Bearer {self.api_key}",
},
)
self._receive_task = self.create_task(self._receive_task_handler())
except Exception as e:
await self.push_error(error_msg=f"Error connecting to Grok: {e}", exception=e)
self._websocket = None
async def _disconnect(self):
"""Close WebSocket connection."""
try:
self._disconnecting = True
self._api_session_ready = False
await self.stop_all_metrics()
if self._websocket:
await self._websocket.close()
self._websocket = None
if self._receive_task:
await self.cancel_task(self._receive_task, timeout=1.0)
self._receive_task = None
self._completed_tool_calls = set()
self._disconnecting = False
except Exception as e:
await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e)
async def _ws_send(self, realtime_message):
"""Send a message over the WebSocket connection."""
try:
if not self._disconnecting and self._websocket:
await self._websocket.send(json.dumps(realtime_message))
except Exception as e:
if self._disconnecting or not self._websocket:
return
await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
async def _update_settings(self):
"""Update session settings on the server."""
settings = self._session_properties
adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter()
if self._context:
llm_invocation_params = adapter.get_llm_invocation_params(self._context)
if llm_invocation_params["tools"]:
settings.tools = llm_invocation_params["tools"]
if llm_invocation_params["system_instruction"]:
settings.instructions = llm_invocation_params["system_instruction"]
# Convert ToolsSchema to list of dicts if needed
if settings.tools and isinstance(settings.tools, ToolsSchema):
settings.tools = adapter.from_standard_tools(settings.tools)
await self.send_client_event(events.SessionUpdateEvent(session=settings))
#
# Inbound server event handling
#
async def _receive_task_handler(self):
"""Handle incoming WebSocket messages."""
async for message in self._websocket:
try:
evt = events.parse_server_event(message)
except Exception as e:
logger.warning(f"Failed to parse server event: {e}")
continue
if evt.type == "ping":
# Ignore ping events (keep-alive)
pass
elif evt.type == "conversation.created":
await self._handle_evt_conversation_created(evt)
elif evt.type == "session.updated":
await self._handle_evt_session_updated(evt)
elif evt.type == "response.created":
await self._handle_evt_response_created(evt)
elif evt.type == "response.output_audio.delta":
await self._handle_evt_audio_delta(evt)
elif evt.type == "response.output_audio.done":
await self._handle_evt_audio_done(evt)
elif evt.type == "response.content_part.added":
# Content part added - we can ignore this for now
pass
elif evt.type == "response.content_part.done":
# Content part done - we can ignore this for now
pass
elif evt.type == "response.output_item.added":
await self._handle_evt_conversation_item_added(evt)
elif evt.type == "response.output_item.done":
# Output item done - we can ignore this for now
pass
elif evt.type == "conversation.item.added":
await self._handle_evt_conversation_item_added(evt)
elif evt.type == "conversation.item.input_audio_transcription.completed":
await self._handle_evt_input_audio_transcription_completed(evt)
elif evt.type == "response.done":
await self._handle_evt_response_done(evt)
elif evt.type == "input_audio_buffer.speech_started":
await self._handle_evt_speech_started(evt)
elif evt.type == "input_audio_buffer.speech_stopped":
await self._handle_evt_speech_stopped(evt)
elif evt.type == "response.output_audio_transcript.delta":
await self._handle_evt_audio_transcript_delta(evt)
elif evt.type == "response.function_call_arguments.delta":
# Function call arguments streaming - we wait for the .done event
pass
elif evt.type == "response.function_call_arguments.done":
await self._handle_evt_function_call_arguments_done(evt)
elif evt.type == "error":
await self._handle_evt_error(evt)
return
async def _handle_evt_conversation_created(self, evt):
"""Handle conversation.created event - first event after connecting."""
await self._update_settings()
async def _handle_evt_response_created(self, evt):
"""Handle response.created event - response generation started."""
pass
async def _handle_evt_session_updated(self, evt):
"""Handle session.updated event."""
self._api_session_ready = True
if self._run_llm_when_api_session_ready:
self._run_llm_when_api_session_ready = False
await self._create_response()
async def _handle_evt_audio_delta(self, evt):
"""Handle audio delta event - streaming audio from assistant."""
await self.stop_ttfb_metrics()
if not self._current_audio_response:
self._current_audio_response = CurrentAudioResponse(
item_id=evt.item_id,
content_index=evt.content_index,
start_time_ms=int(time.time() * 1000),
)
await self.push_frame(TTSStartedFrame())
audio = base64.b64decode(evt.delta)
self._current_audio_response.total_size += len(audio)
frame = TTSAudioRawFrame(
audio=audio,
sample_rate=self._sample_rate,
num_channels=1,
)
await self.push_frame(frame)
async def _handle_evt_audio_done(self, evt):
"""Handle audio done event."""
if self._current_audio_response:
await self.push_frame(TTSStoppedFrame())
async def _handle_evt_conversation_item_added(self, evt):
"""Handle conversation.item.added event."""
if evt.item.type == "function_call":
# Track this function call for when arguments are completed
# Only add if not already tracked (prevent duplicates)
if evt.item.call_id not in self._pending_function_calls:
self._pending_function_calls[evt.item.call_id] = evt.item
else:
# Grok may send multiple conversation.item.added events for the same function call
logger.debug(f"Function call {evt.item.call_id} already tracked, skipping")
await self._call_event_handler("on_conversation_item_created", evt.item.id, evt.item)
if self._messages_added_manually.get(evt.item.id):
del self._messages_added_manually[evt.item.id]
return
if evt.item.role == "assistant":
self._current_assistant_response = evt.item
await self.push_frame(LLMFullResponseStartFrame())
async def _handle_evt_input_audio_transcription_completed(self, evt):
"""Handle input audio transcription completed event."""
await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
# Only push transcription if we have actual text (not empty or just whitespace)
transcript = evt.transcript.strip() if evt.transcript else ""
if transcript:
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), result=evt),
FrameDirection.UPSTREAM,
)
async def _handle_evt_response_done(self, evt):
"""Handle response.done event."""
# Usage metrics - check both response.usage and top-level usage
usage = evt.usage or evt.response.usage
if usage and usage.total_tokens:
tokens = LLMTokenUsage(
prompt_tokens=usage.input_tokens or 0,
completion_tokens=usage.output_tokens or 0,
total_tokens=usage.total_tokens or 0,
)
await self.start_llm_usage_metrics(tokens)
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
self._current_assistant_response = None
# Error handling
if evt.response.status == "failed":
error_msg = "Response failed"
if evt.response.status_details:
error_msg = str(evt.response.status_details)
await self.push_error(error_msg=error_msg)
return
# Update conversation items
for item in evt.response.output:
await self._call_event_handler("on_conversation_item_updated", item.id, item)
async def _handle_evt_audio_transcript_delta(self, evt):
"""Handle audio transcript delta event."""
if evt.delta:
frame = TTSTextFrame(evt.delta, aggregated_by=AggregationType.SENTENCE)
frame.includes_inter_frame_spaces = True
await self.push_frame(frame)
async def _handle_evt_function_call_arguments_done(self, evt):
"""Handle function call arguments done event."""
try:
args = json.loads(evt.arguments)
function_call_item = self._pending_function_calls.get(evt.call_id)
if function_call_item:
del self._pending_function_calls[evt.call_id]
function_calls = [
FunctionCallFromLLM(
context=self._context,
tool_call_id=evt.call_id,
function_name=evt.name,
arguments=args,
)
]
await self.run_function_calls(function_calls)
logger.debug(f"Processed function call: {evt.name}")
else:
logger.warning(f"No tracked function call found for call_id: {evt.call_id}")
except Exception as e:
logger.error(f"Failed to process function call arguments: {e}")
async def _handle_evt_speech_started(self, evt):
"""Handle speech started event from VAD."""
await self._truncate_current_audio_response()
await self.push_interruption_task_frame_and_wait()
await self.push_frame(UserStartedSpeakingFrame())
async def _handle_evt_speech_stopped(self, evt):
"""Handle speech stopped event from VAD."""
await self.start_ttfb_metrics()
await self.start_processing_metrics()
await self.push_frame(UserStoppedSpeakingFrame())
async def _handle_evt_error(self, evt):
"""Handle error event."""
await self.push_error(error_msg=f"Grok Realtime Error: {evt.error.message}")
#
# Response creation
#
async def reset_conversation(self):
"""Reset the conversation by disconnecting and reconnecting."""
logger.debug("Resetting Grok conversation")
await self._disconnect()
self._llm_needs_conversation_setup = True
await self._process_completed_function_calls(send_new_results=False)
await self._connect()
async def _create_response(self):
"""Create an assistant response."""
if not self._api_session_ready:
self._run_llm_when_api_session_ready = True
return
adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter()
if self._llm_needs_conversation_setup:
logger.debug(
f"Setting up Grok conversation with initial messages: "
f"{adapter.get_messages_for_logging(self._context)}"
)
llm_invocation_params = adapter.get_llm_invocation_params(self._context)
messages = llm_invocation_params["messages"]
for item in messages:
evt = events.ConversationItemCreateEvent(item=item)
self._messages_added_manually[evt.item.id] = True
await self.send_client_event(evt)
await self._update_settings()
self._llm_needs_conversation_setup = False
logger.debug("Creating Grok response")
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
await self.start_ttfb_metrics()
await self.send_client_event(
events.ResponseCreateEvent(
response=events.ResponseProperties(modalities=["text", "audio"])
)
)
async def _process_completed_function_calls(self, send_new_results: bool):
"""Process completed function calls and send results to the service."""
sent_new_result = False
for message in self._context.get_messages():
if message.get("role") and message.get("content") != "IN_PROGRESS":
tool_call_id = message.get("tool_call_id")
if tool_call_id and tool_call_id not in self._completed_tool_calls:
if send_new_results:
sent_new_result = True
await self._send_tool_result(tool_call_id, message.get("content"))
self._completed_tool_calls.add(tool_call_id)
if sent_new_result:
await self._create_response()
async def _send_user_audio(self, frame):
"""Send user audio to Grok."""
payload = base64.b64encode(frame.audio).decode("utf-8")
await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload))
async def _send_tool_result(self, tool_call_id: str, result: str):
"""Send a tool call result to Grok."""
item = events.ConversationItem(
type="function_call_output",
call_id=tool_call_id,
output=json.dumps(result),
)
await self.send_client_event(events.ConversationItemCreateEvent(item=item))
def create_context_aggregator(
self,
context: OpenAILLMContext,
*,
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> LLMContextAggregatorPair:
"""Create context aggregators for the Grok Realtime service.
Args:
context: The LLM context.
user_params: User aggregator parameters.
assistant_params: Assistant aggregator parameters.
Returns:
LLMContextAggregatorPair for user and assistant context aggregation.
"""
context = LLMContext.from_openai_context(context)
assistant_params.expect_stripped_words = False
return LLMContextAggregatorPair(
context, user_params=user_params, assistant_params=assistant_params
)