o i@ @sdZddlZddlZddlZddlmZmZddlmZddl m Z m Z m Z m Z mZddlmZddlmZddlmZdd lmZmZmZmZmZmZmZmZmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*dd l+m,Z,dd l-m.Z.dd l/m0Z0m1Z1dd l2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9m:Z:ddl;mZ>m?Z?ddl@mAZAddlBmCZCzddlDmEZFWn eGyZHzeIdeHeIdeJdeHdZH[HwweGdddZKeGddde=ZLGddde:ZMdS)zGrok 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 N) dataclassfield)fields)AnyDictMappingOptionalType)logger) ToolsSchema)GrokRealtimeLLMAdapter)AggregationTypeBotStoppedSpeakingFrame CancelFrameEndFrameFrameInputAudioRawFrameInterruptionFrameLLMContextFrameLLMFullResponseEndFrameLLMFullResponseStartFrameLLMMessagesAppendFrameLLMSetToolsFrame LLMTextFrame StartFrameTranscriptionFrameTTSAudioRawFrameTTSStartedFrameTTSStoppedFrame TTSTextFrameUserStartedSpeakingFrameUserStoppedSpeakingFrame) LLMTokenUsage) LLMContext)LLMAssistantAggregatorParamsLLMUserAggregatorParams)LLMContextAggregatorPair)OpenAILLMContext)FrameDirection)FunctionCallFromLLM LLMService) NOT_GIVEN LLMSettings _NotGivenis_given)time_now_iso8601)events)connectz Exception: zJIn order to use Grok Realtime, you need to `pip install pipecat-ai[grok]`.zMissing module: c@s6eZdZUdZeed<eed<eed<dZeed<dS)CurrentAudioResponseaoTracks 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 content_index start_time_msr total_sizeN)__name__ __module__ __qualname____doc__str__annotations__intr7r?r?V/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/services/grok/realtime/llm.pyr3Ms  r3cseZdZUdZedddZejeBe d<e ddd Z d dd e e efffd d Zededdee efd dfddZZS)GrokRealtimeLLMSettingsaSettings 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. cCstS)N)r+r?r?r?r@isz GrokRealtimeLLMSettings.)default_factorysession_propertiessettingsGrokRealtimeLLMService.SettingscCs.t|jsdS|j}t|jr|j|_dSdS)zBPush top-level ``system_instruction`` into ``session_properties``.N)r.rDsystem_instruction instructions)rEspr?r?r@_sync_top_level_to_spns   z-GrokRealtimeLLMSettings._sync_top_level_to_spdeltareturncsdt|}d|vr+t|jr+|j}d|vr+|jdur+|j}|j|_||jkr+||d<|||S)a 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. rDrGN)super apply_updater.rDrHrGrJ)selfrKchangedrIold_si __class__r?r@rNys   z$GrokRealtimeLLMSettings.apply_updateclsc Csddt|Ddh}i}i}i}ttjj}|D]!\}}|j||} | |vr2||| <q| |vr;||| <q|||<q|rLtjdi||d<|di|} || _ | S)a1Build 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``. cSsh|]}|jqSr?)name).0fr?r?r@ sz7GrokRealtimeLLMSettings.from_mapping..extrarDNr?) dataclass_fieldssetr1SessionProperties model_fieldskeysitems_aliasesgetrY) rTrEown_field_namestopsp_dictrYsp_keyskeyvalue canonicalinstancer?r?r@ from_mappings"    z$GrokRealtimeLLMSettings.from_mapping)rErF)r8r9r:r;rrDr1r\r-r= staticmethodrJrr<rrN classmethodr rrj __classcell__r?r?rRr@rA^s    rAc s~eZdZUdZeZeed<eZdddddde de d e e j d e ed e f fd d Zde fddZde fddZde de efddZdefddZde fddZddZddZdd Zd!d"Z #d{d$ed%ed&edefd'd(Zd)d*Zd+ed,efd-d.Zd/effd0d1 Zd/effd2d3 Zd/e ffd4d5 Z!d/e"de#ffd6d7 Z$d8e%fd9d:Z&d;d<Z'd=e j(fd>d?Z)d@dAZ*dBdCZ+dDdEZ,fdFdGZ-dHdIZ.dJdKZ/dLdMZ0dNdOZ1dPdQZ2dRdSZ3dTdUZ4dVdWZ5dXdYZ6dZd[Z7d\d]Z8d^e fd_d`Z9dadbZ:dcddZ;dedfZdkdlZ?dme fdndoZ@dpdqZAdre dse fdtduZBeCeDdvd8eEdweCdxeDdeFfdydzZGZHS)|GrokRealtimeLLMServiceaGrok 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) _settingszwss://api.x.ai/v1/realtimeNF)base_urlrDrEstart_audio_pausedapi_keyrprDrErqc s|jdddddddddddtd }|dur*td|jd||_|jdur*|j|_|j||dur9||t j d ||d|||_ ||_ ||_ d|_d|_d|_d|_d|_d|_d|_d|_d|_i|_i|_t|_|d|ddS) aInitialize 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. NF) modelrG temperature max_tokenstop_ptop_kfrequency_penaltypresence_penaltyseedfilter_incomplete_user_turnsuser_turn_completion_configrDrD)rprETon_conversation_item_createdon_conversation_item_updatedr?)Settingsr1r\_warn_deprecated_paramrDrHrGrJrNrM__init__rrrp_audio_input_paused _websocket _receive_task_context_llm_needs_conversation_setup_disconnecting_api_session_ready_run_llm_when_api_session_ready_current_assistant_response_current_audio_response_messages_added_manually_pending_function_callsr[_completed_tool_calls_register_event_handler)rOrrrprDrErqkwargsdefault_settingsrRr?r@rsb!    zGrokRealtimeLLMService.__init__rLcCsdS)zCheck if the service can generate usage metrics. Returns: True if metrics generation is supported. Tr?rOr?r?r@can_generate_metrics3sz+GrokRealtimeLLMService.can_generate_metricspausedcCs ||_dS)zzSet whether audio input is paused. Args: paused: True to pause audio input, False to resume. N)r)rOrr?r?r@set_audio_input_paused;s z-GrokRealtimeLLMService.set_audio_input_paused directioncCs`|jjjsdS|dkr|jjjjn|jjjj}|r.|jr.t|jdr&|jjS|jjdvr.dSdS)a!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). Ninputrate)z audio/pcmuz audio/pcmai@) rorDaudioroutputformathasattrrtype)rOr audio_configr?r?r@_get_configured_sample_rateCs     z2GrokRealtimeLLMService._get_configured_sample_ratecCs|d}|dur td|S)zGet 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. rNz"Output sample rate not configured.)r RuntimeError)rOrr?r?r@_get_output_sample_rate`s z.GrokRealtimeLLMService._get_output_sample_ratecCs|jjjr |jjjjdkSdS)z$Check if server-side VAD is enabled. server_vadF)rorDturn_detectionrrr?r?r@_is_turn_detection_enabledos z1GrokRealtimeLLMService._is_turn_detection_enabledcs|s|tIdH|tIdH|IdH|IdH|jr>|t IdH|t IdHdSdS)z-Handle user interruption of assistant speech.N) rsend_client_eventr1InputAudioBufferClearEventResponseCancelEvent _truncate_current_audio_responsestop_all_metricsr push_framerrrr?r?r@_handle_interruptionusz+GrokRealtimeLLMService._handle_interruptioncdS)z#Handle user started speaking event.Nr?rOframer?r?r@_handle_user_started_speakingz4GrokRealtimeLLMService._handle_user_started_speakingcs:|s|tIdH|tIdHdSdS)z#Handle user stopped speaking event.N)rrr1InputAudioBufferCommitEventResponseCreateEventrr?r?r@_handle_user_stopped_speakings z4GrokRealtimeLLMService._handle_user_stopped_speakingcs d|_dS)z"Handle bot stopped speaking event.N)rrr?r?r@_handle_bot_stopped_speakings z3GrokRealtimeLLMService._handle_bot_stopped_speaking total_bytes sample_ratebytes_per_samplecCs,|dur|}||}||}t|dS)zGCalculate audio duration in milliseconds based on PCM audio parameters.N)rr>)rOrrrsamplesduration_secondsr?r?r@_calculate_audio_duration_mss  z3GrokRealtimeLLMService._calculate_audio_duration_msc sP|jsdSzd|_WdSty'}ztd|WYd}~dSd}~ww)zTruncates the current audio response. Note: Grok may not support truncation events like OpenAI. This is a best-effort cleanup. Nz-Audio truncation cleanup failed (non-fatal): )r Exceptionr warningrOer?r?r@rs z7GrokRealtimeLLMService._truncate_current_audio_responseinput_sample_rateoutput_sample_ratecCs`|jj}|js t|_|jjstjtj|dd|j_|jjs.tj tj|dd|j_dSdS)a)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). )r)rN) rorDrr1AudioConfigurationr AudioInputPCMAudioFormatr AudioOutput)rOrrpropsr?r?r@_ensure_audio_configs     z+GrokRealtimeLLMService._ensure_audio_configrcs6t|IdH||j|j|IdHdS)zStart the service and establish WebSocket connection. Args: frame: The start frame triggering service initialization. N)rMstartraudio_in_sample_rateaudio_out_sample_rate_connectrrRr?r@rszGrokRealtimeLLMService.startc&t|IdH|IdHdS)zStop the service and close WebSocket connection. Args: frame: The end frame triggering service shutdown. N)rMstop _disconnectrrRr?r@rzGrokRealtimeLLMService.stopcr)zCancel the service and close WebSocket connection. Args: frame: The cancel frame triggering service cancellation. N)rMcancelrrrRr?r@rrzGrokRealtimeLLMService.cancelcst||IdHt|trnpt|tr ||jIdHnat|tr1|js0| |IdHnPt|t r>| IdHnCt|t rL| |IdHn5t|trZ||IdHn't|trg|IdHnt|tru||IdHn t|tr|IdH|||IdHdS)zProcess incoming frames from the pipeline. Args: frame: The frame to process. direction: The direction of frame flow in the pipeline. N)rM process_frame isinstancerr_handle_contextcontextrr_send_user_audiorrr rr!rrrr_handle_messages_appendr_send_session_updater)rOrrrRr?r@rs.         z$GrokRealtimeLLMService.process_framercsN|js||_|jddIdH|IdHdS||_|jddIdHdS)zHandle LLM context updates.Fsend_new_resultsNT)r!_process_completed_function_calls_create_response)rOrr?r?r@rsz&GrokRealtimeLLMService._handle_contextcstddS)z)Handle appending messages to the context.ztimerrbase64 b64decoderKr7lenrr)rOrrrr?r?r@rs" z.GrokRealtimeLLMService._handle_evt_audio_deltacs"|jr|tIdHdSdS)zHandle audio done event.N)rrrrr?r?r@r z-GrokRealtimeLLMService._handle_evt_audio_donecs|jjdkr"|jj|jvr|j|j|jj<n td|jjd|d|jj|jIdH|j |jjr?|j|jj=dS|jj dkrT|j|_ | t IdHdSdS)z%Handle conversation.item.added event. function_callzFunction call z already tracked, skippingr}N assistant)itemrcall_idrr r_call_event_handleridrrarolerrrrr?r?r@r s   z:GrokRealtimeLLMService._handle_evt_conversation_item_addedcsZ|d|jdIdH|jr|jnd}|r+|t|dt|dtjIdHdSdS)z1Handle input audio transcription completed event.r~N)result) r"r4 transcriptstriprrr/r(UPSTREAM)rOrr'r?r?r@r szFGrokRealtimeLLMService._handle_evt_input_audio_transcription_completedcs|jp|jj}|r%|jr%t|jpd|jpd|jpdd}||IdH|IdH|t IdHd|_ |jj dkrUd}|jj rJt |jj }|j|dIdHdS|jjD] }|d|j|IdHqYdS)zHandle response.done event.r) prompt_tokenscompletion_tokens total_tokensNfailedzResponse failedrr~)usageresponser,r" input_tokens output_tokensstart_llm_usage_metricsstop_processing_metricsrrrstatusstatus_detailsr<rrr"r#)rOrr/tokensrr r?r?r@r s*    z0GrokRealtimeLLMService._handle_evt_response_donecs"|jr||jIdHdSdS)z$Handle audio transcript delta event.N)rK#_push_output_transcript_text_framesrr?r?r@rrz9GrokRealtimeLLMService._handle_evt_audio_transcript_deltatextcsHt|}d|_||IdHt|tjd}d|_||IdHdS)NF) aggregated_byT)rappend_to_contextrrr SENTENCEincludes_inter_frame_spaces)rOr9llm_text_frametts_text_framer?r?r@r8s z:GrokRealtimeLLMService._push_output_transcript_text_framesc sz@t|j}|j|j}|r6|j|j=t|j|j|j|dg}| |IdHt d|jWdSt d|jWdSt y\}zt d|WYd}~dSd}~ww)z*Handle function call arguments done event.)r tool_call_id function_name argumentsNzProcessed function call: z,No tracked function call found for call_id: z+Failed to process function call arguments: )rloadsrBrrar!r)rrUrun_function_callsr rrrr)rOrargsfunction_call_itemfunction_callsrr?r?r@r/s(   z?GrokRealtimeLLMService._handle_evt_function_call_arguments_donecs2|IdH|tIdH|IdHdS)z%Handle speech started event from VAD.N)rbroadcast_framer broadcast_interruptionrr?r?r@r Isz1GrokRealtimeLLMService._handle_evt_speech_startedcs2|IdH|IdH|tIdHdS)z%Handle speech stopped event from VAD.N)start_ttfb_metricsstart_processing_metricsrHr!rr?r?r@rOsz1GrokRealtimeLLMService._handle_evt_speech_stoppedcs"|jd|jjdIdHdS)zHandle error event.zGrok Realtime Error: r.N)rrrrr?r?r@rUs z(GrokRealtimeLLMService._handle_evt_errorcsDtd|IdHd|_|jddIdH|IdHdS)z9Reset the conversation by disconnecting and reconnecting.zResetting Grok conversationNTFr)r rrrrrrr?r?r@reset_conversation]s  z)GrokRealtimeLLMService.reset_conversationcs|js d|_dS|}|jrJtd||j||j}|d}|D]}t j |d}d|j |j j <||IdHq(|IdHd|_td|tIdH|IdH|IdH|t jt jdd gd d IdHdS) zCreate an assistant response.TNz4Setting up Grok conversation with initial messages: messagesr FzCreating Grok responser9r) modalities)r0)rrrrr rget_messages_for_loggingrrr1ConversationItemCreateEventrr r#rrrrrKrJrResponseProperties)rOrrrMr rr?r?r@rgs8    z'GrokRealtimeLLMService._create_responsercsd}|jD]0}|dr8|ddkr8|d}|r8||jvr8|r2d}|||dIdH|j|q|rD|IdHdSdS)zAProcess completed function calls and send results to the service.Fr$content IN_PROGRESSr@TN)r get_messagesrar_send_tool_resultaddr)rOrsent_new_resultrr@r?r?r@rs  z8GrokRealtimeLLMService._process_completed_function_callscs:|jrdSt|jd}|tj|dIdHdS)zSend user audio to Grok.Nzutf-8)r)rr b64encoderdecoderr1InputAudioBufferAppendEvent)rOrpayloadr?r?r@rs z'GrokRealtimeLLMService._send_user_audior@r&cs8tjd|tj|ddd}|tj|dIdHdS)z Send a tool call result to Grok.function_call_outputF) ensure_ascii)rr!rrNN)r1ConversationItemrrrrQ)rOr@r&r r?r?r@rVs z(GrokRealtimeLLMService._send_tool_result user_paramsassistant_paramsrarbcCst|}d|_t|||dS)aNCreate 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. Fr`)r#from_openai_contextexpect_stripped_wordsr&)rOrrarbr?r?r@create_context_aggregators z0GrokRealtimeLLMService.create_context_aggregator)Nr)Ir8r9r:r;rArr=r adapter_classr<rr1r\boolrrrr>rrrrrrrrrrrrrrrrrr(rr#rr ClientEventrrrrrrrrrrrr r r r rr8rr rrrLrrrrVr%r$r'r&rermr?r?rRr@rns a        <  %  rn)Nr;rrr dataclassesrrrrZtypingrrrrr logurur %pipecat.adapters.schemas.tools_schemar /pipecat.adapters.services.grok_realtime_adapterr pipecat.frames.framesr rrrrrrrrrrrrrrrrrrr r!pipecat.metrics.metricsr"*pipecat.processors.aggregators.llm_contextr#+pipecat.processors.aggregators.llm_responser$r%5pipecat.processors.aggregators.llm_response_universalr&1pipecat.processors.aggregators.openai_llm_contextr'"pipecat.processors.frame_processorr(pipecat.services.llm_servicer)r*pipecat.services.settingsr+r,r-r.pipecat.utils.timer/r%r1websockets.asyncio.clientr2rModuleNotFoundErrorrrrr3rArnr?r?r?r@sD    \        ]