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. 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 b12c469ef..6d0d54e79 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 @@ -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 148e8f127..05e1efaf9 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 @@ -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 58389d6d9..9942550e6 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 @@ -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 8784359f0..3f1543871 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 @@ -181,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 0d44470e5..304ccce8e 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 @@ -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 ef58b7805..dedb7b950 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 @@ -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.", ), 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