diff --git a/changelog/3267.added.md b/changelog/3267.added.md new file mode 100644 index 000000000..bdeccd6ed --- /dev/null +++ b/changelog/3267.added.md @@ -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) diff --git a/examples/foundational/20f-persistent-context-grok-realtime.py b/examples/foundational/20f-persistent-context-grok-realtime.py new file mode 100644 index 000000000..e598aa3d9 --- /dev/null +++ b/examples/foundational/20f-persistent-context-grok-realtime.py @@ -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() diff --git a/examples/foundational/51-grok-realtime.py b/examples/foundational/51-grok-realtime.py new file mode 100644 index 000000000..61eaeb062 --- /dev/null +++ b/examples/foundational/51-grok-realtime.py @@ -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() diff --git a/src/pipecat/adapters/services/grok_realtime_adapter.py b/src/pipecat/adapters/services/grok_realtime_adapter.py new file mode 100644 index 000000000..a6270989a --- /dev/null +++ b/src/pipecat/adapters/services/grok_realtime_adapter.py @@ -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 diff --git a/src/pipecat/services/grok/realtime/__init__.py b/src/pipecat/services/grok/realtime/__init__.py new file mode 100644 index 000000000..d23112945 --- /dev/null +++ b/src/pipecat/services/grok/realtime/__init__.py @@ -0,0 +1,5 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# diff --git a/src/pipecat/services/grok/realtime/events.py b/src/pipecat/services/grok/realtime/events.py new file mode 100644 index 000000000..d93aad233 --- /dev/null +++ b/src/pipecat/services/grok/realtime/events.py @@ -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}") diff --git a/src/pipecat/services/grok/realtime/llm.py b/src/pipecat/services/grok/realtime/llm.py new file mode 100644 index 000000000..ec4f263bd --- /dev/null +++ b/src/pipecat/services/grok/realtime/llm.py @@ -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 + )