qiqdZddlZddlmZmZddlmZddlmZm Z m Z m Z m Z m Z mZmZddlmZddlmZddlmcmcmcmZddlmZdd lmZmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5dd l6m7Z7m8Z8m9Z9m:Z:dd l;mm?Z?dd l@mAZAmBZBddlCmDZDmEZEddlFmGZGddlHmIZIerddlJmKZKGddeLeZMeGddZNGdde<ZOy)zGRTVI observer for converting pipeline frames to outgoing RTVI messages.N) dataclassfield)Enum) TYPE_CHECKING AwaitableCallableDictListOptionalSetTuple)logger) BaseModel)calculate_audio_volume)AggregatedTextFrameAggregationTypeBotStartedSpeakingFrameBotStoppedSpeakingFrameFrameFunctionCallCancelFrameFunctionCallInProgressFrameFunctionCallResultFrameFunctionCallsStartedFrameInputAudioRawFrameInterimTranscriptionFrameLLMContextFrameLLMFullResponseEndFrameLLMFullResponseStartFrame LLMTextFrame MetricsFrameTranscriptionFrameTTSAudioRawFrameTTSStartedFrameTTSStoppedFrame TTSTextFrameUserMuteStartedFrameUserMuteStoppedFrameUserStartedSpeakingFrameUserStoppedSpeakingFrame)LLMUsageMetricsDataProcessingMetricsDataTTFBMetricsDataTTSUsageMetricsData) BaseObserver FramePushed)OpenAILLMContextFrame)FrameDirectionFrameProcessor)RTVIServerMessageFrameRTVIServerResponseFrame)BaseOutputTransport)match_endofsentence) RTVIProcessorc eZdZdZdZdZdZdZy)RTVIFunctionCallReportLevelaLevel of detail to include in function call RTVI events. Controls what information is exposed in function call events for security. Values: DISABLED: No events emitted for this function call. NONE: Events only with tool_call_id, no function name or metadata (most secure). NAME: Events with function name, no arguments or results. FULL: Events with function name, arguments, and results. disablednonenamefullN)__name__ __module__ __qualname____doc__DISABLEDNONENAMEFULL]/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/processors/frameworks/rtvi/observer.pyr9r9Ks H D D DrGr9ceZdZUdZdZeed<dZeed<dZeed<dZ eed<dZ eed<dZ eed <dZ eed <dZ eed <dZeed <dZeed <dZeed<dZeed<dZeeed<eeZeeed<dZeeeezed<dZeeeeezeeeezgeeffed<dZ e!ed<edZ"e#ee$fed<y)RTVIObserverParamsa6 Parameters for configuring RTVI Observer behavior. .. deprecated:: 0.0.87 Parameter `errors_enabled` is deprecated. Error messages are always enabled. Parameters: bot_output_enabled: Indicates if bot output messages should be sent. bot_llm_enabled: Indicates if the bot's LLM messages should be sent. bot_tts_enabled: Indicates if the bot's TTS messages should be sent. bot_speaking_enabled: Indicates if the bot's started/stopped speaking messages should be sent. bot_audio_level_enabled: Indicates if bot's audio level messages should be sent. user_llm_enabled: Indicates if the user's LLM input messages should be sent. user_speaking_enabled: Indicates if the user's started/stopped speaking messages should be sent. user_transcription_enabled: Indicates if user's transcription messages should be sent. user_audio_level_enabled: Indicates if user's audio level messages should be sent. metrics_enabled: Indicates if metrics messages should be sent. system_logs_enabled: Indicates if system logs should be sent. errors_enabled: [Deprecated] Indicates if errors messages should be sent. ignored_sources: List of frame processors whose frames should be silently ignored by this observer. Useful for suppressing RTVI messages from secondary pipeline branches (e.g. a silent evaluation LLM) that should not be visible to clients. Sources can also be added and removed dynamically via ``add_ignored_source()`` and ``remove_ignored_source()``. skip_aggregator_types: List of aggregation types to skip sending as tts/output messages. Note: if using this to avoid sending secure information, be sure to also disable bot_llm_enabled to avoid leaking through LLM messages. bot_output_transforms: A list of callables to transform text before just before sending it to TTS. Each callable takes the aggregated text and its type, and returns the transformed text. To register, provide a list of tuples of (aggregation_type | '*', transform_function). audio_level_period_secs: How often audio levels should be sent if enabled. function_call_report_level: Controls what information is exposed in function call events for security. A dict mapping function names to levels, where ``"*"`` sets the default level for unlisted functions:: function_call_report_level={ "*": RTVIFunctionCallReportLevel.NONE, # Default: events with no metadata "get_weather": RTVIFunctionCallReportLevel.FULL, # Expose everything } Levels: - DISABLED: No events emitted for this function. - NONE: Events with tool_call_id only (most secure when events needed). - NAME: Adds function name to events. - FULL: Adds function name, arguments, and results. Defaults to ``{"*": RTVIFunctionCallReportLevel.NONE}``. Tbot_output_enabledbot_llm_enabledbot_tts_enabledbot_speaking_enabledFbot_audio_level_enableduser_llm_enableduser_speaking_enableduser_mute_enableduser_transcription_enableduser_audio_level_enabledmetrics_enabledsystem_logs_enabledNerrors_enabled)default_factoryignored_sourcesskip_aggregator_typesbot_output_transformsg333333?audio_level_period_secsc&dtjiS)N*)r9rCrFrGrHzRTVIObserverParams.s&A&F&F GrGfunction_call_report_level)%r>r?r@rArKbool__annotations__rLrMrNrOrPrQrRrSrTrUrVrWr rlistrYr r2rZrstrr[r rrr\floatr`r r9rFrGrHrJrJ]sD/b $# OT  OT !%$%$)T)!d!"&4&"t"'++%*d* OT %%%)NHTN),1$,GOT.)GCG8D3)>$?@G 8 #%#45y~EFH   &*U)INGJS*E%E FrGrJceZdZdZ d%dddeddeeffdZ d&deee ezge efd e ezfd Z d&deee ezge efd e ezfd Z d e fd Zd e fdZdedefdZdZfdZd'dedefdZdefdZdefdZdefdZdefdZdefdZdefdZde fdZ!defd Z"de#e$zfd!Z%de&fd"Z'de(fd#Z)de(fd$Z*xZ+S)( RTVIObserveraPipeline frame observer for RTVI server message handling. This observer monitors pipeline frames and converts them into appropriate RTVI messages for client communication. It handles various frame types including speech events, transcriptions, LLM responses, and TTS events. Note: This observer only handles outgoing messages. Incoming RTVI client messages are handled by the RTVIProcessor. N)paramsrtvir7rhc t|di|||_|xs t|_t |jj |_t |_d|_ d|_ d|_ d|_ g|_ |jjr$tj |j"|_|jj&Dddl}|j+5|j-d|j/dt0ddd|jj2xsg|_y#1swY)xYw)zInitialize the RTVI observer. Args: rtvi: The RTVI processor to push frames to. params: Settings to enable/disable specific messages. **kwargs: Additional arguments passed to parent class. rFNalwayszLParameter `errors_enabled` is deprecated. Error messages are always enabled.rF)super__init___rtvirJ_paramssetrY_ignored_sources _frames_seen_bot_transcription_last_user_audio_level_last_bot_audio_level_bot_is_speaking_queued_aggregated_text_framesrVradd _logger_sink_system_logger_idrWwarningscatch_warnings simplefilterwarnDeprecationWarningr[_aggregation_transforms)selfrirhkwargsr| __class__s rHrnzRTVIObserver.__init__s  "6" 5!3!5 589U9U5VE"$&'#%&"!&IK+ << + +%+ZZ0A0A%BD " << & & 2 ((* %%h/ b&  LL . . 4" $  s %(D55D>transform_functionaggregation_typec>|jj||fy)aTransform text for a specific aggregation type before sending as Bot Output or TTS. Args: transform_function: The function to apply for transformation. This function should take the text and aggregation type as input and return the transformed text. Ex.: async def my_transform(text: str, aggregation_type: str) -> str: aggregation_type: The type of aggregation to transform. This value defaults to "*" to handle all text before sending to the client. N)rappend)rrrs rHadd_bot_output_transformerz'RTVIObserver.add_bot_output_transformers $$++-=?Q,RSrGcp|jDcgc]\}}||k(r||k(s||fc}}|_ycc}}w)zRemove a text transformer for a specific aggregation type. Args: transform_function: The function to remove. aggregation_type: The type of aggregation to remove the transformer for. N)r)rrragg_typefuncs rHremove_bot_output_transformerz*RTVIObserver.remove_bot_output_transformersD#'">">( $ 00T=O5Ot ( $( s2sourcec:|jj|y)aIgnore all frames pushed by the given processor. Any frame whose source matches ``source`` will be silently skipped, preventing RTVI messages from being emitted for activity in that processor. Useful for suppressing events from secondary pipeline branches (e.g. a silent evaluation LLM) that should not be visible to clients. Args: source: The frame processor to ignore. N)rrryrrs rHadd_ignored_sourcezRTVIObserver.add_ignored_sources !!&)rGc:|jj|y)aStop ignoring frames pushed by the given processor. Reverses a previous call to ``add_ignored_source()``. If ``source`` was not previously ignored this is a no-op. Args: source: The frame processor to stop ignoring. N)rrdiscardrs rHremove_ignored_sourcez"RTVIObserver.remove_ignored_sources %%f-rG function_namereturnc|jj}||vr||S|jdtjS)a1Get the report level for a specific function call. Args: function_name: The name of the function to get the report level for. Returns: The report level for the function. Looks up the function name first, then falls back to "*" key, then NONE. r^)rpr`getr9rC)rrlevelss rH_get_function_call_report_levelz,RTVIObserver._get_function_call_report_level!s>88 F "-( (zz#:??@@rGcKtjtj|}|j|d{y7w)z7Logger sink so we can send system logs to RTVI clients.textdataN)RTVISystemLogMessageTextMessageDatasend_rtvi_message)rmessages rHrzzRTVIObserver._logger_sink0s5''T-A-Aw-OP$$W---s?A AA cKt|d{|jjr t j |j yy7;w)z Cleanup RTVI observer resources.N)rmcleanuprprVrremover{)rrs rHrzRTVIObserver.cleanup5s@go << + + MM$00 1 , sAArC|jj@r-|jCtEjFd{nt|tHrC|jj@r-|jCtEjJd{nCt|tLr1|jj@r|jO|d{nt|tPrC|jjRr-|jCtEjTd{nt|tVrC|jjRr-|jCtEjXd{n\t|tZrk|jj\s|jjRr?t|t^rt|t`sd}n|jc|d{nt|tdr1|jjfr|ji|d{nt|tjr|jlD]}|jo|jp}|trjtk(r2tEjv}|trjxtrjzfvr|jp|_8tEj||}|jC|d{nt|t~r|jo|jp}|trjtk7rtEj|j}|trjxtrjzfvr|jp|_8|trjzk(r|j|_BtEj|}|jC|d{nt|tr|jo|jp}|trjtk7rtEj|jd}|trjxtrjzfvr|jp|_8tEj|}|jC|d{n-t|tr|jo|jp}|trjtk7rtEj|jd}|trjxtrjzfvr|jp|_8|trjzk(r|jr |jnd|_HtEj|}|jC|d{n7t|tr;tEj|j}|jC|d{nt|trB|j|j|d{n|j|d{nt|tr|jjrtj} | |jz } | |jjkDr7t|j|j} tEjtEj| }|jC|d{| |_Snt|tr|jjrtj} | |jz } | |jjkDrjt|j|j} tEjtEj| }|jC|d{| |_\|r&|jj|jyy7 5777i7%777G777/77.7H777L777;7pw)zProcess a frame being pushed through the pipeline. Args: data: Frame push event data containing source, frame, direction, and timestamp. NTFr) tool_call_id)r cancelled)value)_rframe directionrrbroadcast_sibling_idr1 DOWNSTREAMidrs isinstancer(r)rprQ_handle_interruptionsr&r'rR_handle_user_muterrrN_handle_bot_speakingr!rrS_handle_user_transcriptionsr0rrP_handle_contextrrLrrBotLLMStartedMessagerBotLLMStoppedMessager_handle_llm_text_framer#rMBotTTSStartedMessager$BotTTSStoppedMessagerrKr%r5_handle_aggregated_llm_textr rU_handle_metricsrfunction_callsrrr9rBLLMFunctionCallStartMessageDatarDrELLMFunctionCallStartMessager$LLMFunctionCallInProgressMessageDatar arguments LLMFunctionCallInProgressMessager!LLMFunctionCallStoppedMessageDataLLMFunctionCallStoppedMessagerresultr3 ServerMessagerr4error_send_error_response_send_server_responserrTtimerur\raudio sample_rateUserAudioLevelMessageAudioLevelMessageDatar"rOrvBotAudioLevelMessagery) rrsrcrr mark_as_seen function_call report_levelr curr_time diff_timelevels rH on_push_framezRTVIObserver.on_push_frameIs kk NN   SD,A,A%A   % % 1i>C\C\6\  88t(( (   u79QR S 22,,U3 3 3 u35IJ K ..((/ / / u68OP Q 11++E2 2 2 u13LM N 772259 9 9 u4oF G --&&u- - - 8 9dll>Z>Z(()B)B)DE E E 6 7DLL#G#GG;;=/44/44$*7)D)DD&::E,,W555 6: ;??@S@STL:CCC@@eN`N`a/44/44$*/)<)#C#CC%*__DN??TJ,,W555 6 7??@S@STL:CCC==!&!3!3" /44/44$*/)<)#C#CC27,,%,,DDK<<$G,,W555 5 6((ejj9G((1 1 1 6 7{{&//66600777 1 2t||7\7\ I!D$?$??I4<<???.u{{E$>>I4<<???.u{{E -666 6 27766sB;h0=g2>Ah0g5Ah0 g8 Ah0g;Ah0g>Ah0,h-Ah0?hAh0hAh0h Ah0&h 'A:h0!h"Ah0"h#Ch0(h)C(h0hCh0hC5h0 h A h0h"6h0 h%h0(h()Ch00h+1C h0=h.>5h05h08h0;h0>h0h0h0h0 h0 h0h0h0h0h0h0h0"h0%h0(h0+h0.h0rcKd}t|trtj}n$t|trtj }|r|j |d{yy7w)z)Handle user speaking interruption frames.N)rr(rUserStartedSpeakingMessager)UserStoppedSpeakingMessagerrrrs rHrz"RTVIObserver._handle_interruptionss\ e5 6557G 7 8557G ((1 1 1  1A"A-$A+%A-cKd}t|trtj}n$t|trtj }|r|j |d{yy7w)zHandle user mute/unmute frames.N)rr&rUserMuteStartedMessager'UserMuteStoppedMessagerrs rHrzRTVIObserver._handle_user_mutes\ e1 2113G 3 4113G ((1 1 1  1rcKt|trytj}|j |d{|j D]}|j |d{|j jd|_yt|tr5tj}|j |d{d|_yy77p7w)z!Handle bot speaking event frames.NTF) rrrBotStartedSpeakingMessagerrx_send_aggregated_llm_textclearrwrBotStoppedSpeakingMessage)rrr queued_frames rHrz!RTVIObserver._handle_bot_speakings e4 5446G((1 1 1 $ C C C 44\BBB C  / / 5 5 7$(D ! 6 7446G((1 1 1$)D !8 2C 2s49CC'C#C$A CCCCCcK|jr|j|d{y|jj|y7!w)z)Handle aggregated LLM text output frames.N)rwrrxr)rrs rHrz(RTVIObserver._handle_aggregated_llm_text s>  007 7 7  / / 6 6u = 8s!AA"AcK|jjr#|j|jjvry|j}|j}|jD]!\}}||k(s|dk(s|||d{}#t |t }|jjrEtjtj|||}|j|d{|r[|jjrDtjtj|}|j|d{yyy77d7 w)z"Send aggregated LLM text messages.Nr^)rspoken aggregated_byrr)rprZrrrrr%rKrBotOutputMessageBotOutputMessageDatarrMBotTTSTextMessager) rrrrr transformisTTSr tts_messages rHrz&RTVIObserver._send_aggregated_llm_texts) LL . .##t||'I'II zz&&+/+G+G 7 ' i8+/?3/F&tX66 75,/ << * *++..DV^_G((1 1 1 T\\1100d6J6JPT6UVK((5 5 5257 2 6s>A.E1 E=E >A,E*E+AEEEEEcKtjtj|j}|j |d{|xj |jz c_t |j rlt|j dkDrS|j tjtj|j d{d|_yyy77w)zHandle LLM text output frames.rrNrrk) rBotLLMTextMessagerrrrtr6lenBotTranscriptionMessagers rHrz#RTVIObserver._handle_llm_text_frame.s((d.B.B .ST$$W--- 5::- t66 7C@W@Ws e/ 03366U]]eoo]aG 8 93366U]]eoo]bG ((1 1 1  1sCCCCc>K t|tr|jj}n|jj }|sy|d}t |dr|j dk(rtt |drhdjd|jD}|rtjtj| }|j|d{yt|tr}|ddk(rt|d }t|trd jd |D}n|}tjtj| }|j|d{yyyy77 #t$r"}t!j"d |Yd}~yd}~wwxYww)zHProcess LLM context frames to extract user messages for the RTVI client.Nroleuserpartsrkc3NK|]}t|ds|jywrN)hasattrr).0parts rH z/RTVIObserver._handle_context..`s\TgdTZF[tyy\s%%rrcontent c30K|]}d|vs|dywrrF)ritems rHr z/RTVIObserver._handle_context..js'[FVZNV '[s  z0Caught an error while trying to handle context: )rr0contextmessages get_messagesrrjoinrrUserLLMTextMessagerrdictrc Exceptionrwarning)rrrrr rtvi_messager es rHrzRTVIObserver._handle_contextQsh S%!67 ==11 ==557rlGw'GLLF,BwwX_G`ww\W]]\\#'#:#:@T@TZ^@_#`L00>>>GT*6?f,%i0G!'40"xx'['[[&#'#:#:@T@TZ^@_#`L00>>>-+ >? S NNMaSQ R R SsmFAE/FBE/E+E/FBE/"E-#E/'F+E/-E// F8FFFFcKi}|jD]}t|tr.d|vrg|d<|dj|j dBt|t r.d|vrg|d<|dj|j dt|t r8d|vrg|d<|dj|jj dt|tsd|vrg|d<|dj|j dtj|}|j|d{y7w) z;Handle metrics frames and convert to RTVI metrics messages.ttfbT)r processingtokens charactersrN) rrr,r model_dumpr+r*rr-rMetricsMessager)rrmetricsdrs rHrzRTVIObserver._handle_metricsss6 NA!_-(&(GFO&&q|||'FGA45w.,.GL) %,,Q\\t\-LMA237*(*GH%!(();););)NOA23w.,.GL) %,,Q\\t\-LM! N$%%73$$W---sC'E *AE E E cKtjt|jjtj |jj |j}|j|d{y7w)5Send a response to the client for a specific request.)tr rrN) rServerResponserd client_msgmsg_idRawServerResponseDatatyperrrs rHrz"RTVIObserver._send_server_responses`%%5##**+++e.>.>.C.CuzzR $$W---sAB?BcKtjt|jjtj |j }|j|d{y7w)r")rr$N)r ErrorResponserdr&r'ErrorResponseDatarrrs rHrz!RTVIObserver._send_error_responsesQ$$5##**+$2H2Hu{{2[ $$W---sA'A1)A/*A1)N)r^)T),r>r?r@rAr rJrnrrdrrrrr2rrr9rrzrrrarr/rrrrrrrrrrrr0rrr rr4rr __classcell__)rs@rHrgrgs +/,504 ,5',5+, ,5b36T$c?S+@%A9S>%QRT*C/T&36 $c?S+@%A9S>%QR *C/ " * * .N . AS AE` A. 2 IY Id I[, [,z 2 2 2U 2 * *>7J>65H64),) 2u2& S+@?+R SD.<.0.1H..0G.rGrg)PrAr dataclassesrrenumrtypingrrrr r r r r logururpydanticr)pipecat.processors.frameworks.rtvi.models processors frameworksrimodelsrpipecat.audio.utilsrpipecat.frames.framesrrrrrrrrrrrrrrrr r!r"r#r$r%r&r'r(r)pipecat.metrics.metricsr*r+r,r-pipecat.observers.base_observerr.r/1pipecat.processors.aggregators.openai_llm_contextr0"pipecat.processors.frame_processorr1r2)pipecat.processors.frameworks.rtvi.framesr3r4pipecat.transports.base_outputr5pipecat.utils.stringr6,pipecat.processors.frameworks.rtvi.processorr7rdr9rJrgrFrGrHrAsN (   8866 FSM?4J#t$ LL L^k.<k.rG