From 1c99a537b24206211a66b37f0e5b0bbd9c1232ff Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 25 Mar 2026 12:07:40 -0400 Subject: [PATCH 1/3] Consolidate Grok services into xai module Both GrokLLMService and XAIHttpTTSService use the same xAI API (api.x.ai), so move Grok source files into the xai module. Leave deprecation shims in the old grok/ paths for backward compatibility. --- .../foundational/07e-interruptible-xai.py | 2 +- .../foundational/14g-function-calling-grok.py | 2 +- .../20f-persistent-context-grok-realtime.py | 4 +- examples/foundational/51-grok-realtime.py | 6 +- .../55zo-update-settings-grok-realtime.py | 4 +- .../55zza-update-settings-grok-llm.py | 2 +- .../services/grok_realtime_adapter.py | 2 +- src/pipecat/services/grok/__init__.py | 4 +- src/pipecat/services/grok/llm.py | 252 +---- src/pipecat/services/grok/realtime/events.py | 872 +--------------- src/pipecat/services/grok/realtime/llm.py | 972 +----------------- src/pipecat/services/xai/llm.py | 250 +++++ src/pipecat/services/xai/realtime/__init__.py | 0 src/pipecat/services/xai/realtime/events.py | 872 ++++++++++++++++ src/pipecat/services/xai/realtime/llm.py | 971 +++++++++++++++++ tests/test_settings.py | 4 +- 16 files changed, 2145 insertions(+), 2074 deletions(-) create mode 100644 src/pipecat/services/xai/llm.py create mode 100644 src/pipecat/services/xai/realtime/__init__.py create mode 100644 src/pipecat/services/xai/realtime/events.py create mode 100644 src/pipecat/services/xai/realtime/llm.py diff --git a/examples/foundational/07e-interruptible-xai.py b/examples/foundational/07e-interruptible-xai.py index b12c469ef..8a19a1570 100644 --- a/examples/foundational/07e-interruptible-xai.py +++ b/examples/foundational/07e-interruptible-xai.py @@ -23,7 +23,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.grok.llm import GrokLLMService +from pipecat.services.xai.llm import GrokLLMService from pipecat.services.xai.tts import XAIHttpTTSService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams diff --git a/examples/foundational/14g-function-calling-grok.py b/examples/foundational/14g-function-calling-grok.py index 148e8f127..df94fa2f8 100644 --- a/examples/foundational/14g-function-calling-grok.py +++ b/examples/foundational/14g-function-calling-grok.py @@ -26,8 +26,8 @@ from pipecat.processors.aggregators.llm_response_universal import ( from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.grok.llm import GrokLLMService from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.xai.llm import GrokLLMService from pipecat.services.xai.tts import XAIHttpTTSService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams diff --git a/examples/foundational/20f-persistent-context-grok-realtime.py b/examples/foundational/20f-persistent-context-grok-realtime.py index 58389d6d9..ed4a4d273 100644 --- a/examples/foundational/20f-persistent-context-grok-realtime.py +++ b/examples/foundational/20f-persistent-context-grok-realtime.py @@ -36,9 +36,9 @@ from pipecat.processors.aggregators.llm_response_universal import ( ) 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.services.xai.realtime.events import SessionProperties, TurnDetection +from pipecat.services.xai.realtime.llm import GrokRealtimeLLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams diff --git a/examples/foundational/51-grok-realtime.py b/examples/foundational/51-grok-realtime.py index 8784359f0..5c4ac7d96 100644 --- a/examples/foundational/51-grok-realtime.py +++ b/examples/foundational/51-grok-realtime.py @@ -51,11 +51,9 @@ from pipecat.processors.aggregators.llm_response_universal import ( ) from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport -from pipecat.services.grok.realtime.events import ( - SessionProperties, -) -from pipecat.services.grok.realtime.llm import GrokRealtimeLLMService from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.xai.realtime.events import SessionProperties +from pipecat.services.xai.realtime.llm import GrokRealtimeLLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams diff --git a/examples/foundational/55zo-update-settings-grok-realtime.py b/examples/foundational/55zo-update-settings-grok-realtime.py index 0d44470e5..9e853420e 100644 --- a/examples/foundational/55zo-update-settings-grok-realtime.py +++ b/examples/foundational/55zo-update-settings-grok-realtime.py @@ -23,8 +23,8 @@ from pipecat.processors.aggregators.llm_response_universal import ( ) from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport -from pipecat.services.grok.realtime import events -from pipecat.services.grok.realtime.llm import GrokRealtimeLLMService +from pipecat.services.xai.realtime import events +from pipecat.services.xai.realtime.llm import GrokRealtimeLLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams diff --git a/examples/foundational/55zza-update-settings-grok-llm.py b/examples/foundational/55zza-update-settings-grok-llm.py index ef58b7805..04cd39a95 100644 --- a/examples/foundational/55zza-update-settings-grok-llm.py +++ b/examples/foundational/55zza-update-settings-grok-llm.py @@ -24,7 +24,7 @@ from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.grok.llm import GrokLLMService +from pipecat.services.xai.llm import GrokLLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams diff --git a/src/pipecat/adapters/services/grok_realtime_adapter.py b/src/pipecat/adapters/services/grok_realtime_adapter.py index b95efb62c..4e5e2a8c5 100644 --- a/src/pipecat/adapters/services/grok_realtime_adapter.py +++ b/src/pipecat/adapters/services/grok_realtime_adapter.py @@ -21,7 +21,7 @@ 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 +from pipecat.services.xai.realtime import events class GrokRealtimeLLMInvocationParams(TypedDict): diff --git a/src/pipecat/services/grok/__init__.py b/src/pipecat/services/grok/__init__.py index f76d4b305..4e6879d25 100644 --- a/src/pipecat/services/grok/__init__.py +++ b/src/pipecat/services/grok/__init__.py @@ -8,6 +8,6 @@ import sys from pipecat.services import DeprecatedModuleProxy -from .llm import * +from .llm import * # noqa: F401,F403 -sys.modules[__name__] = DeprecatedModuleProxy(globals(), "grok", "grok.llm") +sys.modules[__name__] = DeprecatedModuleProxy(globals(), "grok", "xai.llm") diff --git a/src/pipecat/services/grok/llm.py b/src/pipecat/services/grok/llm.py index 160ad3331..e81db5458 100644 --- a/src/pipecat/services/grok/llm.py +++ b/src/pipecat/services/grok/llm.py @@ -4,247 +4,21 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Grok LLM service implementation using OpenAI-compatible interface. +"""Grok LLM service implementation. -This module provides a service for interacting with Grok's API through an -OpenAI-compatible interface, including specialized token usage tracking -and context aggregation functionality. +.. deprecated:: + This module is deprecated. Please use GrokLLMService from + pipecat.services.xai.llm instead. """ -from dataclasses import dataclass -from typing import Optional +import warnings -from loguru import logger +from pipecat.services.xai.llm import * # noqa: F401,F403 -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.openai_llm_context import OpenAILLMContext -from pipecat.services.openai.base_llm import BaseOpenAILLMService -from pipecat.services.openai.llm import ( - OpenAIAssistantContextAggregator, - OpenAILLMService, - OpenAIUserContextAggregator, -) - - -@dataclass -class GrokContextAggregatorPair: - """Pair of context aggregators for user and assistant interactions. - - Provides a convenient container for managing both user and assistant - context aggregators together for Grok LLM interactions. - - .. deprecated:: 0.0.99 - `GrokContextAggregatorPair` is deprecated and will be removed in a future version. - Use the universal `LLMContext` and `LLMContextAggregatorPair` instead. - See `OpenAILLMContext` docstring for migration guide. - - Parameters: - _user: The user context aggregator instance. - _assistant: The assistant context aggregator instance. - """ - - # Aggregators handle deprecation warnings - _user: OpenAIUserContextAggregator - _assistant: OpenAIAssistantContextAggregator - - def user(self) -> OpenAIUserContextAggregator: - """Get the user context aggregator. - - Returns: - The user context aggregator instance. - """ - return self._user - - def assistant(self) -> OpenAIAssistantContextAggregator: - """Get the assistant context aggregator. - - Returns: - The assistant context aggregator instance. - """ - return self._assistant - - -@dataclass -class GrokLLMSettings(BaseOpenAILLMService.Settings): - """Settings for GrokLLMService.""" - - pass - - -class GrokLLMService(OpenAILLMService): - """A service for interacting with Grok's API using the OpenAI-compatible interface. - - This service extends OpenAILLMService to connect to Grok's API endpoint while - maintaining full compatibility with OpenAI's interface and functionality. - Includes specialized token usage tracking that accumulates metrics during - processing and reports final totals. - """ - - Settings = GrokLLMSettings - _settings: Settings - - def __init__( - self, - *, - api_key: str, - base_url: str = "https://api.x.ai/v1", - model: Optional[str] = None, - settings: Optional[Settings] = None, - **kwargs, - ): - """Initialize the GrokLLMService with API key and model. - - Args: - api_key: The API key for accessing Grok's API. - base_url: The base URL for Grok API. Defaults to "https://api.x.ai/v1". - model: The model identifier to use. Defaults to "grok-3-beta". - - .. deprecated:: 0.0.105 - Use ``settings=GrokLLMService.Settings(model=...)`` instead. - - settings: Runtime-updatable settings. When provided alongside deprecated - parameters, ``settings`` values take precedence. - **kwargs: Additional keyword arguments passed to OpenAILLMService. - """ - # 1. Initialize default_settings with hardcoded defaults - default_settings = self.Settings(model="grok-3-beta") - - # 2. Apply direct init arg overrides (deprecated) - if model is not None: - self._warn_init_param_moved_to_settings("model", "model") - default_settings.model = model - - # 3. (No step 3, as there's no params object to apply) - - # 4. Apply settings delta (canonical API, always wins) - if settings is not None: - default_settings.apply_update(settings) - - super().__init__(api_key=api_key, base_url=base_url, settings=default_settings, **kwargs) - # Initialize counters for token usage metrics - self._prompt_tokens = 0 - self._completion_tokens = 0 - self._total_tokens = 0 - self._has_reported_prompt_tokens = False - self._is_processing = False - - def create_client(self, api_key=None, base_url=None, **kwargs): - """Create OpenAI-compatible client for Grok API endpoint. - - Args: - api_key: The API key to use. If None, uses instance default. - base_url: The base URL to use. If None, uses instance default. - **kwargs: Additional arguments passed to client creation. - - Returns: - The configured client instance for Grok API. - """ - logger.debug(f"Creating Grok client with api {base_url}") - return super().create_client(api_key, base_url, **kwargs) - - async def _process_context(self, context: OpenAILLMContext | LLMContext): - """Process a context through the LLM and accumulate token usage metrics. - - This method overrides the parent class implementation to handle Grok's - incremental token reporting style, accumulating the counts and reporting - them once at the end of processing. - - Args: - context: The context to process, containing messages and other - information needed for the LLM interaction. - """ - # Reset all counters and flags at the start of processing - self._prompt_tokens = 0 - self._completion_tokens = 0 - self._total_tokens = 0 - self._cache_read_input_tokens = None - self._reasoning_tokens = None - self._has_reported_prompt_tokens = False - self._is_processing = True - - try: - await super()._process_context(context) - finally: - self._is_processing = False - # Report final accumulated token usage at the end of processing - if self._prompt_tokens > 0 or self._completion_tokens > 0: - self._total_tokens = self._prompt_tokens + self._completion_tokens - tokens = LLMTokenUsage( - prompt_tokens=self._prompt_tokens, - completion_tokens=self._completion_tokens, - total_tokens=self._total_tokens, - cache_read_input_tokens=self._cache_read_input_tokens, - reasoning_tokens=self._reasoning_tokens, - ) - await super().start_llm_usage_metrics(tokens) - - async def start_llm_usage_metrics(self, tokens: LLMTokenUsage): - """Accumulate token usage metrics during processing. - - This method intercepts the incremental token updates from Grok's API - and accumulates them instead of passing each update to the metrics system. - The final accumulated totals are reported at the end of processing. - - Args: - tokens: The token usage metrics for the current chunk of processing, - containing prompt_tokens, completion_tokens, and optional cached/reasoning tokens. - """ - # Only accumulate metrics during active processing - if not self._is_processing: - return - - # Record prompt tokens the first time we see them - if not self._has_reported_prompt_tokens and tokens.prompt_tokens > 0: - self._prompt_tokens = tokens.prompt_tokens - self._has_reported_prompt_tokens = True - - # Update completion tokens count if it has increased - if tokens.completion_tokens > self._completion_tokens: - self._completion_tokens = tokens.completion_tokens - - # Capture cached & reasoning tokens (these typically only appear once per request) - if tokens.cache_read_input_tokens is not None: - self._cache_read_input_tokens = tokens.cache_read_input_tokens - - if tokens.reasoning_tokens is not None: - self._reasoning_tokens = tokens.reasoning_tokens - - def create_context_aggregator( - self, - context: OpenAILLMContext, - *, - user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), - assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), - ) -> GrokContextAggregatorPair: - """Create an instance of GrokContextAggregatorPair from an OpenAILLMContext. - - Constructor keyword arguments for both the user and assistant aggregators - can be provided. - - Args: - context: The LLM context to create aggregators for. - user_params: Parameters for configuring the user aggregator. - assistant_params: Parameters for configuring the assistant aggregator. - - Returns: - GrokContextAggregatorPair: A pair of context aggregators, one for - the user and one for the assistant, encapsulated in an - GrokContextAggregatorPair. - - .. deprecated:: 0.0.99 - `create_context_aggregator()` is deprecated and will be removed in a future version. - Use the universal `LLMContext` and `LLMContextAggregatorPair` instead. - See `OpenAILLMContext` docstring for migration guide. - """ - context.set_llm_adapter(self.get_llm_adapter()) - - # Aggregators handle deprecation warnings - user = OpenAIUserContextAggregator(context, params=user_params) - assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) - - return GrokContextAggregatorPair(_user=user, _assistant=assistant) +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "pipecat.services.grok.llm is deprecated. Please use pipecat.services.xai.llm instead.", + DeprecationWarning, + stacklevel=2, + ) diff --git a/src/pipecat/services/grok/realtime/events.py b/src/pipecat/services/grok/realtime/events.py index 1f89a92f7..546308e26 100644 --- a/src/pipecat/services/grok/realtime/events.py +++ b/src/pipecat/services/grok/realtime/events.py @@ -4,869 +4,21 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Event models and data structures for Grok Voice Agent API communication. +"""Grok Realtime event models. -Based on xAI's Grok Voice Agent API documentation: -https://docs.x.ai/docs/guides/voice/agent +.. deprecated:: + This module is deprecated. Please use pipecat.services.xai.realtime.events instead. """ -import json -import uuid -from typing import Any, Dict, List, Literal, Optional, Union +import warnings -from pydantic import BaseModel, ConfigDict, Field +from pipecat.services.xai.realtime.events import * # noqa: F401,F403 -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. - Defaults to "Ara". - turn_detection: Configuration for turn detection. Defaults to server-side VAD. - Set to None for manual turn detection. - 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 | str] = "Ara" - turn_detection: Optional[TurnDetection] = Field( - default_factory=lambda: TurnDetection(type="server_vad") +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "pipecat.services.grok.realtime.events is deprecated. " + "Please use pipecat.services.xai.realtime.events instead.", + DeprecationWarning, + stacklevel=2, ) - 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 ResponseFunctionCallArgumentsDelta(ServerEvent): - """Event containing incremental function call arguments. - - Parameters: - type: Event type, always "response.function_call_arguments.delta". - response_id: ID of the response. - item_id: ID of the conversation item. - call_id: ID of the function call. - delta: Incremental function arguments as JSON. - previous_item_id: ID of the previous item, if any. - """ - - type: Literal["response.function_call_arguments.delta"] - response_id: Optional[str] = None - item_id: Optional[str] = None - call_id: str - delta: str - previous_item_id: Optional[str] = None - - -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.delta": ResponseFunctionCallArgumentsDelta, - "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 index 1317e7269..5d35e158f 100644 --- a/src/pipecat/services/grok/realtime/llm.py +++ b/src/pipecat/services/grok/realtime/llm.py @@ -4,968 +4,22 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Grok Realtime Voice Agent LLM service implementation with WebSocket support. +"""Grok Realtime LLM service. -Based on xAI's Grok Voice Agent API documentation: -https://docs.x.ai/docs/guides/voice/agent +.. deprecated:: + This module is deprecated. Please use GrokRealtimeLLMService from + pipecat.services.xai.realtime.llm instead. """ -import base64 -import json -import time -from dataclasses import dataclass, field -from dataclasses import fields as dataclass_fields -from typing import Any, Dict, Mapping, Optional, Type +import warnings -from loguru import logger +from pipecat.services.xai.realtime.llm import * # noqa: F401,F403 -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, - InterruptionFrame, - LLMContextFrame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMMessagesAppendFrame, - LLMSetToolsFrame, - LLMTextFrame, - 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.services.settings import ( - NOT_GIVEN, - LLMSettings, - _NotGiven, - is_given, -) -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 - - -@dataclass -class GrokRealtimeLLMSettings(LLMSettings): - """Settings for GrokRealtimeLLMService. - - Parameters: - session_properties: Grok Realtime session properties (voice, audio config, - tools, etc.). ``instructions`` is synced bidirectionally with the - top-level ``system_instruction`` field. - """ - - session_properties: events.SessionProperties | _NotGiven = field( - default_factory=lambda: NOT_GIVEN +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "pipecat.services.grok.realtime.llm is deprecated. " + "Please use pipecat.services.xai.realtime.llm instead.", + DeprecationWarning, + stacklevel=2, ) - - # -- Bidirectional sync helpers ------------------------------------------ - - @staticmethod - def _sync_top_level_to_sp(settings: "GrokRealtimeLLMService.Settings"): - """Push top-level ``system_instruction`` into ``session_properties``.""" - if not is_given(settings.session_properties): - return - sp = settings.session_properties - if is_given(settings.system_instruction): - sp.instructions = settings.system_instruction - - # -- apply_update override ----------------------------------------------- - - def apply_update(self, delta: "GrokRealtimeLLMService.Settings") -> Dict[str, Any]: - """Merge a delta, keeping ``system_instruction`` in sync with SP. - - When the delta contains ``session_properties``, it **replaces** the - stored SP wholesale (matching legacy behaviour). Top-level field - values always take precedence over conflicting SP values. - """ - # 1. Let the base class handle all fields including session_properties - # (wholesale replacement when given). - changed = super().apply_update(delta) - - # 2. SP → top-level: if the SP was just replaced and carries - # instructions that the delta didn't set at top level, pull it up. - if "session_properties" in changed and is_given(self.session_properties): - sp = self.session_properties - if "system_instruction" not in changed and sp.instructions is not None: - old_si = self.system_instruction - self.system_instruction = sp.instructions - if old_si != self.system_instruction: - changed["system_instruction"] = old_si - - # 3. Top-level → SP: ensure SP mirrors the authoritative top-level - # values. Covers all cases: top-level-only delta, SP-only delta, - # and mixed deltas where top-level takes precedence. - self._sync_top_level_to_sp(self) - - return changed - - # -- from_mapping override ----------------------------------------------- - - @classmethod - def from_mapping( - cls: Type["GrokRealtimeLLMService.Settings"], settings: Mapping[str, Any] - ) -> "GrokRealtimeLLMService.Settings": - """Build a delta from a plain dict, routing SP keys into ``session_properties``. - - Keys that correspond to ``SessionProperties`` fields are collected into - a nested ``session_properties`` value. ``model`` is always routed to - the top-level field. Unknown keys go to ``extra``. - """ - # Determine which keys belong to our own dataclass fields. - own_field_names = {f.name for f in dataclass_fields(cls)} - {"extra"} - - top: Dict[str, Any] = {} - sp_dict: Dict[str, Any] = {} - extra: Dict[str, Any] = {} - - sp_keys = set(events.SessionProperties.model_fields.keys()) - - for key, value in settings.items(): - # Resolve aliases first - canonical = cls._aliases.get(key, key) - if canonical in own_field_names: - top[canonical] = value - elif canonical in sp_keys: - sp_dict[canonical] = value - else: - extra[key] = value - - if sp_dict: - top["session_properties"] = events.SessionProperties(**sp_dict) - - instance = cls(**top) - instance.extra = extra - return instance - - -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) - """ - - Settings = GrokRealtimeLLMSettings - _settings: Settings - - # Use the Grok-specific adapter - adapter_class = GrokRealtimeLLMAdapter - - def __init__( - self, - *, - api_key: str, - base_url: str = "wss://api.x.ai/v1/realtime", - session_properties: Optional[events.SessionProperties] = None, - settings: Optional[Settings] = None, - start_audio_paused: bool = False, - **kwargs, - ): - """Initialize the Grok Realtime Voice Agent LLM service. - - Args: - api_key: xAI API key for authentication. - 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 voice "Ara". - - .. deprecated:: 0.0.105 - Use ``settings=GrokRealtimeLLMService.Settings(session_properties=...)`` - instead. - - To set a different voice, configure it in session_properties: - - session_properties = events.SessionProperties(voice="Rex") - - Available voices: Ara, Rex, Sal, Eve, Leo. - settings: Runtime-updatable settings for this service. - start_audio_paused: Whether to start with audio input paused. Defaults to False. - **kwargs: Additional arguments passed to parent LLMService. - """ - # 1. Initialize default_settings with hardcoded defaults - default_settings = self.Settings( - model=None, - system_instruction=None, - temperature=None, - max_tokens=None, - top_p=None, - top_k=None, - frequency_penalty=None, - presence_penalty=None, - seed=None, - filter_incomplete_user_turns=False, - user_turn_completion_config=None, - session_properties=events.SessionProperties(), - ) - - # 2. Apply direct init arg overrides (deprecated) - if session_properties is not None: - _warn_deprecated_param( - "session_properties", - self.Settings, - "session_properties", - ) - default_settings.session_properties = session_properties - # Sync instructions from the deprecated SP arg to top-level - if session_properties.instructions is not None: - default_settings.system_instruction = session_properties.instructions - - # Sync top-level system_instruction back into session_properties - self.Settings._sync_top_level_to_sp(default_settings) - - # 3. Apply settings delta (canonical API, always wins) - if settings is not None: - default_settings.apply_update(settings) - - super().__init__( - base_url=base_url, - settings=default_settings, - **kwargs, - ) - - self.api_key = api_key - self.base_url = base_url - - 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 _get_configured_sample_rate(self, direction: str) -> Optional[int]: - """Get manually configured sample rate for input or output. - - Args: - direction: Either "input" or "output". - - Returns: - Configured sample rate or None if not manually configured. - For PCMU/PCMA formats, returns 8000 Hz (G.711 standard). - """ - if not self._settings.session_properties.audio: - return None - - audio_config = ( - self._settings.session_properties.audio.input - if direction == "input" - else self._settings.session_properties.audio.output - ) - - if audio_config and audio_config.format: - # PCM format has configurable rate - if hasattr(audio_config.format, "rate"): - return audio_config.format.rate - # PCMU/PCMA formats are fixed at 8000 Hz (G.711 standard) - elif audio_config.format.type in ("audio/pcmu", "audio/pcma"): - return 8000 - - return None - - def _get_output_sample_rate(self) -> int: - """Get the output sample rate from session properties. - - Returns: - Output sample rate in Hz. - - Note: - This assumes start() has been called, which guarantees - session_properties.audio.output exists. - """ - rate = self._get_configured_sample_rate("output") - if rate is None: - raise RuntimeError("Output sample rate not configured.") - return rate - - def _is_turn_detection_enabled(self) -> bool: - """Check if server-side VAD is enabled.""" - if self._settings.session_properties.turn_detection: - return self._settings.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._get_output_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 - # - - def _ensure_audio_config(self, input_sample_rate: int, output_sample_rate: int): - """Ensure session_properties.audio has input and output configs. - - Fills in any missing audio configuration using the given sample rates. - - Args: - input_sample_rate: Sample rate for audio input (Hz). - output_sample_rate: Sample rate for audio output (Hz). - """ - props = self._settings.session_properties - if not props.audio: - props.audio = events.AudioConfiguration() - if not props.audio.input: - props.audio.input = events.AudioInput( - format=events.PCMAudioFormat(rate=input_sample_rate) - ) - if not props.audio.output: - props.audio.output = events.AudioOutput( - format=events.PCMAudioFormat(rate=output_sample_rate) - ) - - 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) - self._ensure_audio_config(frame.audio_in_sample_rate, frame.audio_out_sample_rate) - 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, LLMSetToolsFrame): - await self._send_session_update() - - 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, delta): - """Apply a settings delta, sending a session update when needed.""" - # Capture audio config before the update — a wholesale SP replacement - # would lose it since the new SP likely has audio=None. - input_rate = self._get_configured_sample_rate("input") - output_rate = self._get_configured_sample_rate("output") - - changed = await super()._update_settings(delta) - - # Re-establish audio config if it was lost during SP replacement. - if "session_properties" in changed and input_rate and output_rate: - self._ensure_audio_config(input_rate, output_rate) - - handled = {"session_properties", "system_instruction"} - if changed.keys() & handled: - await self._send_session_update() - self._warn_unhandled_updated_settings(changed.keys() - handled) - return changed - - async def _send_session_update(self): - """Update session settings on the server.""" - settings = self._settings.session_properties - adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter() - - if self._context: - llm_invocation_params = adapter.get_llm_invocation_params( - self._context, system_instruction=self._settings.system_instruction - ) - - if llm_invocation_params["tools"]: - settings.tools = llm_invocation_params["tools"] - - # The adapter resolves conflicts between init-provided and - # context-provided system instructions (preferring init-provided). - 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": - if evt.error.code in ( - "response_cancel_not_active", - "conversation_already_has_active_response", - ): - logger.debug(f"{self} {evt.error.message}") - else: - 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._send_session_update() - - 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._get_output_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: - await self._push_output_transcript_text_frames(evt.delta) - - async def _push_output_transcript_text_frames(self, text: str): - # In a typical "cascade" LLM + TTS setup, LLMTextFrames would not - # proceed beyond the TTS service. Therefore, since a speech-to-speech - # service like Grok Realtime combines both LLM and TTS functionality, - # you might think we wouldn't need to push LLMTextFrames at all. - # However, RTVI relies on LLMTextFrames being pushed to trigger its - # "bot-llm-text" event. So here we push an LLMTextFrame, too, but avoid - # appending it to context to avoid context message duplication. - - # Push LLMTextFrame - llm_text_frame = LLMTextFrame(text) - llm_text_frame.append_to_context = False - await self.push_frame(llm_text_frame) - - # Push TTSTextFrame - tts_text_frame = TTSTextFrame(text, aggregated_by=AggregationType.SENTENCE) - tts_text_frame.includes_inter_frame_spaces = True - await self.push_frame(tts_text_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.broadcast_frame(UserStartedSpeakingFrame) - await self.broadcast_interruption() - - 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.broadcast_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._send_session_update() - 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.""" - # Don't send audio if conversation setup is still pending, as it can - # lead to errors. For example: audio sent before conversation setup - # will be interpreted as having Grok's default sample rate (24000), - # and if that differs from the sample rate we eventually set through - # the conversation setup, Grok will error out. - if self._llm_needs_conversation_setup: - return - - 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, ensure_ascii=False), - ) - 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 - ) diff --git a/src/pipecat/services/xai/llm.py b/src/pipecat/services/xai/llm.py new file mode 100644 index 000000000..160ad3331 --- /dev/null +++ b/src/pipecat/services/xai/llm.py @@ -0,0 +1,250 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Grok LLM service implementation using OpenAI-compatible interface. + +This module provides a service for interacting with Grok's API through an +OpenAI-compatible interface, including specialized token usage tracking +and context aggregation functionality. +""" + +from dataclasses import dataclass +from typing import Optional + +from loguru import logger + +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.openai_llm_context import OpenAILLMContext +from pipecat.services.openai.base_llm import BaseOpenAILLMService +from pipecat.services.openai.llm import ( + OpenAIAssistantContextAggregator, + OpenAILLMService, + OpenAIUserContextAggregator, +) + + +@dataclass +class GrokContextAggregatorPair: + """Pair of context aggregators for user and assistant interactions. + + Provides a convenient container for managing both user and assistant + context aggregators together for Grok LLM interactions. + + .. deprecated:: 0.0.99 + `GrokContextAggregatorPair` is deprecated and will be removed in a future version. + Use the universal `LLMContext` and `LLMContextAggregatorPair` instead. + See `OpenAILLMContext` docstring for migration guide. + + Parameters: + _user: The user context aggregator instance. + _assistant: The assistant context aggregator instance. + """ + + # Aggregators handle deprecation warnings + _user: OpenAIUserContextAggregator + _assistant: OpenAIAssistantContextAggregator + + def user(self) -> OpenAIUserContextAggregator: + """Get the user context aggregator. + + Returns: + The user context aggregator instance. + """ + return self._user + + def assistant(self) -> OpenAIAssistantContextAggregator: + """Get the assistant context aggregator. + + Returns: + The assistant context aggregator instance. + """ + return self._assistant + + +@dataclass +class GrokLLMSettings(BaseOpenAILLMService.Settings): + """Settings for GrokLLMService.""" + + pass + + +class GrokLLMService(OpenAILLMService): + """A service for interacting with Grok's API using the OpenAI-compatible interface. + + This service extends OpenAILLMService to connect to Grok's API endpoint while + maintaining full compatibility with OpenAI's interface and functionality. + Includes specialized token usage tracking that accumulates metrics during + processing and reports final totals. + """ + + Settings = GrokLLMSettings + _settings: Settings + + def __init__( + self, + *, + api_key: str, + base_url: str = "https://api.x.ai/v1", + model: Optional[str] = None, + settings: Optional[Settings] = None, + **kwargs, + ): + """Initialize the GrokLLMService with API key and model. + + Args: + api_key: The API key for accessing Grok's API. + base_url: The base URL for Grok API. Defaults to "https://api.x.ai/v1". + model: The model identifier to use. Defaults to "grok-3-beta". + + .. deprecated:: 0.0.105 + Use ``settings=GrokLLMService.Settings(model=...)`` instead. + + settings: Runtime-updatable settings. When provided alongside deprecated + parameters, ``settings`` values take precedence. + **kwargs: Additional keyword arguments passed to OpenAILLMService. + """ + # 1. Initialize default_settings with hardcoded defaults + default_settings = self.Settings(model="grok-3-beta") + + # 2. Apply direct init arg overrides (deprecated) + if model is not None: + self._warn_init_param_moved_to_settings("model", "model") + default_settings.model = model + + # 3. (No step 3, as there's no params object to apply) + + # 4. Apply settings delta (canonical API, always wins) + if settings is not None: + default_settings.apply_update(settings) + + super().__init__(api_key=api_key, base_url=base_url, settings=default_settings, **kwargs) + # Initialize counters for token usage metrics + self._prompt_tokens = 0 + self._completion_tokens = 0 + self._total_tokens = 0 + self._has_reported_prompt_tokens = False + self._is_processing = False + + def create_client(self, api_key=None, base_url=None, **kwargs): + """Create OpenAI-compatible client for Grok API endpoint. + + Args: + api_key: The API key to use. If None, uses instance default. + base_url: The base URL to use. If None, uses instance default. + **kwargs: Additional arguments passed to client creation. + + Returns: + The configured client instance for Grok API. + """ + logger.debug(f"Creating Grok client with api {base_url}") + return super().create_client(api_key, base_url, **kwargs) + + async def _process_context(self, context: OpenAILLMContext | LLMContext): + """Process a context through the LLM and accumulate token usage metrics. + + This method overrides the parent class implementation to handle Grok's + incremental token reporting style, accumulating the counts and reporting + them once at the end of processing. + + Args: + context: The context to process, containing messages and other + information needed for the LLM interaction. + """ + # Reset all counters and flags at the start of processing + self._prompt_tokens = 0 + self._completion_tokens = 0 + self._total_tokens = 0 + self._cache_read_input_tokens = None + self._reasoning_tokens = None + self._has_reported_prompt_tokens = False + self._is_processing = True + + try: + await super()._process_context(context) + finally: + self._is_processing = False + # Report final accumulated token usage at the end of processing + if self._prompt_tokens > 0 or self._completion_tokens > 0: + self._total_tokens = self._prompt_tokens + self._completion_tokens + tokens = LLMTokenUsage( + prompt_tokens=self._prompt_tokens, + completion_tokens=self._completion_tokens, + total_tokens=self._total_tokens, + cache_read_input_tokens=self._cache_read_input_tokens, + reasoning_tokens=self._reasoning_tokens, + ) + await super().start_llm_usage_metrics(tokens) + + async def start_llm_usage_metrics(self, tokens: LLMTokenUsage): + """Accumulate token usage metrics during processing. + + This method intercepts the incremental token updates from Grok's API + and accumulates them instead of passing each update to the metrics system. + The final accumulated totals are reported at the end of processing. + + Args: + tokens: The token usage metrics for the current chunk of processing, + containing prompt_tokens, completion_tokens, and optional cached/reasoning tokens. + """ + # Only accumulate metrics during active processing + if not self._is_processing: + return + + # Record prompt tokens the first time we see them + if not self._has_reported_prompt_tokens and tokens.prompt_tokens > 0: + self._prompt_tokens = tokens.prompt_tokens + self._has_reported_prompt_tokens = True + + # Update completion tokens count if it has increased + if tokens.completion_tokens > self._completion_tokens: + self._completion_tokens = tokens.completion_tokens + + # Capture cached & reasoning tokens (these typically only appear once per request) + if tokens.cache_read_input_tokens is not None: + self._cache_read_input_tokens = tokens.cache_read_input_tokens + + if tokens.reasoning_tokens is not None: + self._reasoning_tokens = tokens.reasoning_tokens + + def create_context_aggregator( + self, + context: OpenAILLMContext, + *, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), + ) -> GrokContextAggregatorPair: + """Create an instance of GrokContextAggregatorPair from an OpenAILLMContext. + + Constructor keyword arguments for both the user and assistant aggregators + can be provided. + + Args: + context: The LLM context to create aggregators for. + user_params: Parameters for configuring the user aggregator. + assistant_params: Parameters for configuring the assistant aggregator. + + Returns: + GrokContextAggregatorPair: A pair of context aggregators, one for + the user and one for the assistant, encapsulated in an + GrokContextAggregatorPair. + + .. deprecated:: 0.0.99 + `create_context_aggregator()` is deprecated and will be removed in a future version. + Use the universal `LLMContext` and `LLMContextAggregatorPair` instead. + See `OpenAILLMContext` docstring for migration guide. + """ + context.set_llm_adapter(self.get_llm_adapter()) + + # Aggregators handle deprecation warnings + user = OpenAIUserContextAggregator(context, params=user_params) + assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) + + return GrokContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/xai/realtime/__init__.py b/src/pipecat/services/xai/realtime/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/services/xai/realtime/events.py b/src/pipecat/services/xai/realtime/events.py new file mode 100644 index 000000000..1f89a92f7 --- /dev/null +++ b/src/pipecat/services/xai/realtime/events.py @@ -0,0 +1,872 @@ +# +# Copyright (c) 2024-2026, 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. + Defaults to "Ara". + turn_detection: Configuration for turn detection. Defaults to server-side VAD. + Set to None for manual turn detection. + 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 | str] = "Ara" + turn_detection: Optional[TurnDetection] = Field( + default_factory=lambda: TurnDetection(type="server_vad") + ) + 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 ResponseFunctionCallArgumentsDelta(ServerEvent): + """Event containing incremental function call arguments. + + Parameters: + type: Event type, always "response.function_call_arguments.delta". + response_id: ID of the response. + item_id: ID of the conversation item. + call_id: ID of the function call. + delta: Incremental function arguments as JSON. + previous_item_id: ID of the previous item, if any. + """ + + type: Literal["response.function_call_arguments.delta"] + response_id: Optional[str] = None + item_id: Optional[str] = None + call_id: str + delta: str + previous_item_id: Optional[str] = None + + +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.delta": ResponseFunctionCallArgumentsDelta, + "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/xai/realtime/llm.py b/src/pipecat/services/xai/realtime/llm.py new file mode 100644 index 000000000..1317e7269 --- /dev/null +++ b/src/pipecat/services/xai/realtime/llm.py @@ -0,0 +1,971 @@ +# +# Copyright (c) 2024-2026, 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, field +from dataclasses import fields as dataclass_fields +from typing import Any, Dict, Mapping, Optional, Type + +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, + InterruptionFrame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesAppendFrame, + LLMSetToolsFrame, + LLMTextFrame, + 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.services.settings import ( + NOT_GIVEN, + LLMSettings, + _NotGiven, + is_given, +) +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 + + +@dataclass +class GrokRealtimeLLMSettings(LLMSettings): + """Settings for GrokRealtimeLLMService. + + Parameters: + session_properties: Grok Realtime session properties (voice, audio config, + tools, etc.). ``instructions`` is synced bidirectionally with the + top-level ``system_instruction`` field. + """ + + session_properties: events.SessionProperties | _NotGiven = field( + default_factory=lambda: NOT_GIVEN + ) + + # -- Bidirectional sync helpers ------------------------------------------ + + @staticmethod + def _sync_top_level_to_sp(settings: "GrokRealtimeLLMService.Settings"): + """Push top-level ``system_instruction`` into ``session_properties``.""" + if not is_given(settings.session_properties): + return + sp = settings.session_properties + if is_given(settings.system_instruction): + sp.instructions = settings.system_instruction + + # -- apply_update override ----------------------------------------------- + + def apply_update(self, delta: "GrokRealtimeLLMService.Settings") -> Dict[str, Any]: + """Merge a delta, keeping ``system_instruction`` in sync with SP. + + When the delta contains ``session_properties``, it **replaces** the + stored SP wholesale (matching legacy behaviour). Top-level field + values always take precedence over conflicting SP values. + """ + # 1. Let the base class handle all fields including session_properties + # (wholesale replacement when given). + changed = super().apply_update(delta) + + # 2. SP → top-level: if the SP was just replaced and carries + # instructions that the delta didn't set at top level, pull it up. + if "session_properties" in changed and is_given(self.session_properties): + sp = self.session_properties + if "system_instruction" not in changed and sp.instructions is not None: + old_si = self.system_instruction + self.system_instruction = sp.instructions + if old_si != self.system_instruction: + changed["system_instruction"] = old_si + + # 3. Top-level → SP: ensure SP mirrors the authoritative top-level + # values. Covers all cases: top-level-only delta, SP-only delta, + # and mixed deltas where top-level takes precedence. + self._sync_top_level_to_sp(self) + + return changed + + # -- from_mapping override ----------------------------------------------- + + @classmethod + def from_mapping( + cls: Type["GrokRealtimeLLMService.Settings"], settings: Mapping[str, Any] + ) -> "GrokRealtimeLLMService.Settings": + """Build a delta from a plain dict, routing SP keys into ``session_properties``. + + Keys that correspond to ``SessionProperties`` fields are collected into + a nested ``session_properties`` value. ``model`` is always routed to + the top-level field. Unknown keys go to ``extra``. + """ + # Determine which keys belong to our own dataclass fields. + own_field_names = {f.name for f in dataclass_fields(cls)} - {"extra"} + + top: Dict[str, Any] = {} + sp_dict: Dict[str, Any] = {} + extra: Dict[str, Any] = {} + + sp_keys = set(events.SessionProperties.model_fields.keys()) + + for key, value in settings.items(): + # Resolve aliases first + canonical = cls._aliases.get(key, key) + if canonical in own_field_names: + top[canonical] = value + elif canonical in sp_keys: + sp_dict[canonical] = value + else: + extra[key] = value + + if sp_dict: + top["session_properties"] = events.SessionProperties(**sp_dict) + + instance = cls(**top) + instance.extra = extra + return instance + + +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) + """ + + Settings = GrokRealtimeLLMSettings + _settings: Settings + + # Use the Grok-specific adapter + adapter_class = GrokRealtimeLLMAdapter + + def __init__( + self, + *, + api_key: str, + base_url: str = "wss://api.x.ai/v1/realtime", + session_properties: Optional[events.SessionProperties] = None, + settings: Optional[Settings] = None, + start_audio_paused: bool = False, + **kwargs, + ): + """Initialize the Grok Realtime Voice Agent LLM service. + + Args: + api_key: xAI API key for authentication. + 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 voice "Ara". + + .. deprecated:: 0.0.105 + Use ``settings=GrokRealtimeLLMService.Settings(session_properties=...)`` + instead. + + To set a different voice, configure it in session_properties: + + session_properties = events.SessionProperties(voice="Rex") + + Available voices: Ara, Rex, Sal, Eve, Leo. + settings: Runtime-updatable settings for this service. + start_audio_paused: Whether to start with audio input paused. Defaults to False. + **kwargs: Additional arguments passed to parent LLMService. + """ + # 1. Initialize default_settings with hardcoded defaults + default_settings = self.Settings( + model=None, + system_instruction=None, + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=False, + user_turn_completion_config=None, + session_properties=events.SessionProperties(), + ) + + # 2. Apply direct init arg overrides (deprecated) + if session_properties is not None: + _warn_deprecated_param( + "session_properties", + self.Settings, + "session_properties", + ) + default_settings.session_properties = session_properties + # Sync instructions from the deprecated SP arg to top-level + if session_properties.instructions is not None: + default_settings.system_instruction = session_properties.instructions + + # Sync top-level system_instruction back into session_properties + self.Settings._sync_top_level_to_sp(default_settings) + + # 3. Apply settings delta (canonical API, always wins) + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + base_url=base_url, + settings=default_settings, + **kwargs, + ) + + self.api_key = api_key + self.base_url = base_url + + 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 _get_configured_sample_rate(self, direction: str) -> Optional[int]: + """Get manually configured sample rate for input or output. + + Args: + direction: Either "input" or "output". + + Returns: + Configured sample rate or None if not manually configured. + For PCMU/PCMA formats, returns 8000 Hz (G.711 standard). + """ + if not self._settings.session_properties.audio: + return None + + audio_config = ( + self._settings.session_properties.audio.input + if direction == "input" + else self._settings.session_properties.audio.output + ) + + if audio_config and audio_config.format: + # PCM format has configurable rate + if hasattr(audio_config.format, "rate"): + return audio_config.format.rate + # PCMU/PCMA formats are fixed at 8000 Hz (G.711 standard) + elif audio_config.format.type in ("audio/pcmu", "audio/pcma"): + return 8000 + + return None + + def _get_output_sample_rate(self) -> int: + """Get the output sample rate from session properties. + + Returns: + Output sample rate in Hz. + + Note: + This assumes start() has been called, which guarantees + session_properties.audio.output exists. + """ + rate = self._get_configured_sample_rate("output") + if rate is None: + raise RuntimeError("Output sample rate not configured.") + return rate + + def _is_turn_detection_enabled(self) -> bool: + """Check if server-side VAD is enabled.""" + if self._settings.session_properties.turn_detection: + return self._settings.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._get_output_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 + # + + def _ensure_audio_config(self, input_sample_rate: int, output_sample_rate: int): + """Ensure session_properties.audio has input and output configs. + + Fills in any missing audio configuration using the given sample rates. + + Args: + input_sample_rate: Sample rate for audio input (Hz). + output_sample_rate: Sample rate for audio output (Hz). + """ + props = self._settings.session_properties + if not props.audio: + props.audio = events.AudioConfiguration() + if not props.audio.input: + props.audio.input = events.AudioInput( + format=events.PCMAudioFormat(rate=input_sample_rate) + ) + if not props.audio.output: + props.audio.output = events.AudioOutput( + format=events.PCMAudioFormat(rate=output_sample_rate) + ) + + 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) + self._ensure_audio_config(frame.audio_in_sample_rate, frame.audio_out_sample_rate) + 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, LLMSetToolsFrame): + await self._send_session_update() + + 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, delta): + """Apply a settings delta, sending a session update when needed.""" + # Capture audio config before the update — a wholesale SP replacement + # would lose it since the new SP likely has audio=None. + input_rate = self._get_configured_sample_rate("input") + output_rate = self._get_configured_sample_rate("output") + + changed = await super()._update_settings(delta) + + # Re-establish audio config if it was lost during SP replacement. + if "session_properties" in changed and input_rate and output_rate: + self._ensure_audio_config(input_rate, output_rate) + + handled = {"session_properties", "system_instruction"} + if changed.keys() & handled: + await self._send_session_update() + self._warn_unhandled_updated_settings(changed.keys() - handled) + return changed + + async def _send_session_update(self): + """Update session settings on the server.""" + settings = self._settings.session_properties + adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter() + + if self._context: + llm_invocation_params = adapter.get_llm_invocation_params( + self._context, system_instruction=self._settings.system_instruction + ) + + if llm_invocation_params["tools"]: + settings.tools = llm_invocation_params["tools"] + + # The adapter resolves conflicts between init-provided and + # context-provided system instructions (preferring init-provided). + 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": + if evt.error.code in ( + "response_cancel_not_active", + "conversation_already_has_active_response", + ): + logger.debug(f"{self} {evt.error.message}") + else: + 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._send_session_update() + + 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._get_output_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: + await self._push_output_transcript_text_frames(evt.delta) + + async def _push_output_transcript_text_frames(self, text: str): + # In a typical "cascade" LLM + TTS setup, LLMTextFrames would not + # proceed beyond the TTS service. Therefore, since a speech-to-speech + # service like Grok Realtime combines both LLM and TTS functionality, + # you might think we wouldn't need to push LLMTextFrames at all. + # However, RTVI relies on LLMTextFrames being pushed to trigger its + # "bot-llm-text" event. So here we push an LLMTextFrame, too, but avoid + # appending it to context to avoid context message duplication. + + # Push LLMTextFrame + llm_text_frame = LLMTextFrame(text) + llm_text_frame.append_to_context = False + await self.push_frame(llm_text_frame) + + # Push TTSTextFrame + tts_text_frame = TTSTextFrame(text, aggregated_by=AggregationType.SENTENCE) + tts_text_frame.includes_inter_frame_spaces = True + await self.push_frame(tts_text_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.broadcast_frame(UserStartedSpeakingFrame) + await self.broadcast_interruption() + + 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.broadcast_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._send_session_update() + 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.""" + # Don't send audio if conversation setup is still pending, as it can + # lead to errors. For example: audio sent before conversation setup + # will be interpreted as having Grok's default sample rate (24000), + # and if that differs from the sample rate we eventually set through + # the conversation setup, Grok will error out. + if self._llm_needs_conversation_setup: + return + + 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, ensure_ascii=False), + ) + 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 + ) diff --git a/tests/test_settings.py b/tests/test_settings.py index 8ffe355ad..3419f940e 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -10,8 +10,6 @@ from unittest.mock import patch from pipecat.services.deepgram.stt import DeepgramSTTService, DeepgramSTTSettings from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTSettings -from pipecat.services.grok.realtime import events as grok_events -from pipecat.services.grok.realtime.llm import GrokRealtimeLLMSettings from pipecat.services.openai.realtime import events from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMSettings from pipecat.services.settings import ( @@ -23,6 +21,8 @@ from pipecat.services.settings import ( _NotGiven, is_given, ) +from pipecat.services.xai.realtime import events as grok_events +from pipecat.services.xai.realtime.llm import GrokRealtimeLLMSettings # --------------------------------------------------------------------------- # NOT_GIVEN sentinel From 54a17ab1f330c226f40833b3c20cfaa148bfe618 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 25 Mar 2026 12:22:37 -0400 Subject: [PATCH 2/3] Add changelog for #4142 --- changelog/4142.changed.md | 1 + changelog/4142.deprecated.md | 1 + 2 files changed, 2 insertions(+) create mode 100644 changelog/4142.changed.md create mode 100644 changelog/4142.deprecated.md diff --git a/changelog/4142.changed.md b/changelog/4142.changed.md new file mode 100644 index 000000000..06fa50f74 --- /dev/null +++ b/changelog/4142.changed.md @@ -0,0 +1 @@ +- `GrokLLMService` and `GrokRealtimeLLMService` now live in the `pipecat.services.xai` module alongside `XAIHttpTTSService`, since all three use the same xAI API. Update imports from `pipecat.services.grok.*` to `pipecat.services.xai.*` (e.g. `from pipecat.services.xai.llm import GrokLLMService`). diff --git a/changelog/4142.deprecated.md b/changelog/4142.deprecated.md new file mode 100644 index 000000000..a17444edd --- /dev/null +++ b/changelog/4142.deprecated.md @@ -0,0 +1 @@ +- `pipecat.services.grok.llm`, `pipecat.services.grok.realtime.llm`, and `pipecat.services.grok.realtime.events` are deprecated. The old import paths still work but emit a `DeprecationWarning`; use `pipecat.services.xai.llm`, `pipecat.services.xai.realtime.llm`, and `pipecat.services.xai.realtime.events` instead. From 6d1918f12af7c0ad408e1dfbb793618fd4475311 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 25 Mar 2026 23:23:58 -0400 Subject: [PATCH 3/3] Update GROK_API_KEY to XAI_API_KEY --- env.example | 6 +++--- examples/foundational/07e-interruptible-xai.py | 4 ++-- examples/foundational/14g-function-calling-grok.py | 4 ++-- .../foundational/20f-persistent-context-grok-realtime.py | 2 +- examples/foundational/51-grok-realtime.py | 2 +- examples/foundational/55zo-update-settings-grok-realtime.py | 2 +- examples/foundational/55zza-update-settings-grok-llm.py | 2 +- 7 files changed, 11 insertions(+), 11 deletions(-) diff --git a/env.example b/env.example index 00b5fa775..3723fbf26 100644 --- a/env.example +++ b/env.example @@ -80,9 +80,6 @@ GOOGLE_TEST_CREDENTIALS=... # Gradium GRAPDIUM_API_KEY=... -# Grok -GROK_API_KEY=... - # Groq GROQ_API_KEY=... @@ -215,3 +212,6 @@ WHATSAPP_TOKEN=... WHATSAPP_WEBHOOK_VERIFICATION_TOKEN=... WHATSAPP_PHONE_NUMBER_ID=... WHATSAPP_APP_SECRET=... + +# xAI / Grok +XAI_API_KEY=... \ No newline at end of file diff --git a/examples/foundational/07e-interruptible-xai.py b/examples/foundational/07e-interruptible-xai.py index 8a19a1570..6d0d54e79 100644 --- a/examples/foundational/07e-interruptible-xai.py +++ b/examples/foundational/07e-interruptible-xai.py @@ -56,7 +56,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) tts = XAIHttpTTSService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), aiohttp_session=session, settings=XAIHttpTTSService.Settings( voice="eve", @@ -64,7 +64,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) llm = GrokLLMService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), settings=GrokLLMService.Settings( system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", ), diff --git a/examples/foundational/14g-function-calling-grok.py b/examples/foundational/14g-function-calling-grok.py index df94fa2f8..05e1efaf9 100644 --- a/examples/foundational/14g-function-calling-grok.py +++ b/examples/foundational/14g-function-calling-grok.py @@ -65,7 +65,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) tts = XAIHttpTTSService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), aiohttp_session=session, settings=XAIHttpTTSService.Settings( voice="eve", @@ -73,7 +73,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) llm = GrokLLMService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), settings=GrokLLMService.Settings( system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", ), diff --git a/examples/foundational/20f-persistent-context-grok-realtime.py b/examples/foundational/20f-persistent-context-grok-realtime.py index ed4a4d273..9942550e6 100644 --- a/examples/foundational/20f-persistent-context-grok-realtime.py +++ b/examples/foundational/20f-persistent-context-grok-realtime.py @@ -192,7 +192,7 @@ Remember, your responses should be short - just one or two sentences usually.""" ) llm = GrokRealtimeLLMService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), session_properties=session_properties, ) diff --git a/examples/foundational/51-grok-realtime.py b/examples/foundational/51-grok-realtime.py index 5c4ac7d96..3f1543871 100644 --- a/examples/foundational/51-grok-realtime.py +++ b/examples/foundational/51-grok-realtime.py @@ -179,7 +179,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # Create the Grok Realtime LLM service llm = GrokRealtimeLLMService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), settings=GrokRealtimeLLMService.Settings( system_instruction="""You are a helpful and friendly AI assistant powered by Grok. diff --git a/examples/foundational/55zo-update-settings-grok-realtime.py b/examples/foundational/55zo-update-settings-grok-realtime.py index 9e853420e..304ccce8e 100644 --- a/examples/foundational/55zo-update-settings-grok-realtime.py +++ b/examples/foundational/55zo-update-settings-grok-realtime.py @@ -50,7 +50,7 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - llm = GrokRealtimeLLMService(api_key=os.getenv("GROK_API_KEY")) + llm = GrokRealtimeLLMService(api_key=os.getenv("XAI_API_KEY")) messages = [ { diff --git a/examples/foundational/55zza-update-settings-grok-llm.py b/examples/foundational/55zza-update-settings-grok-llm.py index 04cd39a95..dedb7b950 100644 --- a/examples/foundational/55zza-update-settings-grok-llm.py +++ b/examples/foundational/55zza-update-settings-grok-llm.py @@ -60,7 +60,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) llm = GrokLLMService( - api_key=os.getenv("GROK_API_KEY"), + api_key=os.getenv("XAI_API_KEY"), settings=GrokLLMService.Settings( system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", ),