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

Merge pull request #4142 from pipecat-ai/mb/grok-move-to-xai-module

Browse Source 
Consolidate Grok services into xai module
This commit is contained in:
Mark Backman
2026-03-25 23:32:18 -04:00
committed by GitHub
parent 2441c4f801 6d1918f12a
commit 7fef3b01eb
19 changed files with 2158 additions and 2085 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4142.changed.md Normal file
View File

@@ -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`).

1
changelog/4142.deprecated.md Normal file
View File

@@ -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.

6
env.example
View File

@@ -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=...

6
examples/foundational/07e-interruptible-xai.py
View File

@@ -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.",
),

6
examples/foundational/14g-function-calling-grok.py
View File

@@ -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.",
),

6
examples/foundational/20f-persistent-context-grok-realtime.py
View File

@@ -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,
)

8
examples/foundational/51-grok-realtime.py
View File

@@ -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.

6
examples/foundational/55zo-update-settings-grok-realtime.py
View File

@@ -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 = [
{

4
examples/foundational/55zza-update-settings-grok-llm.py
View File

@@ -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.",
),

2
src/pipecat/adapters/services/grok_realtime_adapter.py
View File

@@ -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):

4
src/pipecat/services/grok/__init__.py
View File

@@ -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")

252
src/pipecat/services/grok/llm.py
View File

@@ -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,
)

872
src/pipecat/services/grok/realtime/events.py
View File

@@ -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}")

972
src/pipecat/services/grok/realtime/llm.py
View File

@@ -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
)

250
src/pipecat/services/xai/llm.py Normal file
View File

@@ -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)

0
src/pipecat/services/xai/realtime/__init__.py Normal file
View File

872
src/pipecat/services/xai/realtime/events.py Normal file
View File

@@ -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}")

971
src/pipecat/services/xai/realtime/llm.py Normal file
View File

@@ -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
)

4
tests/test_settings.py
View File

@@ -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