j`dZddlmZddlmZddlmZddlmZm Z m Z ddl m Z ddl mZddlmZdd lmZmZmZdd lmZmZd d lmZmZmZmZmZmZmZm Z d Z!dZ"dZ#dZ$ddhZ%dZ&ddhZ'dZ(dZ)dZ*dZ+dZ,dddddZ-d Z.Gd!d"eZ/y#)$uOTel → LangSmith bridge for LiveKit Agents. Rewrites LiveKit's ``lk.*`` span data into the ``gen_ai.*`` / ``langsmith.*`` namespaces LangSmith ingests; non-LiveKit spans on the same provider pass through untouched. The call recording is attached to the root span, either from a local file (``audio_path_provider``, dev) or via :meth:`expect_recording` / :meth:`complete_recording` (LiveKit Egress, production). Shared export / ``thread_id`` / message plumbing lives in :class:`BaseLangSmithSpanProcessor`. ) annotations)MutableMapping)Path)AnyCallableOptional)TTLCache) SpanProcessor)get_package_version)build_assistant_messagebuild_user_messagetry_parse_json_object)BaseLangSmithSpanProcessorTranslatedSpan)build_message_from_eventextract_llm_usageextract_model_from_lk_metrics extract_provider_from_lk_metricsextract_realtime_usage$flatten_lk_attributes_to_ls_metadatais_livekit_spannormalize_providerg @i user_turn llm_requestllm_nodellm_request_run tts_requesttts_nodetts_request_run agent_turn agent_session function_toolrealtime_metrics user_speakingsystemuser assistanttool)zgen_ai.system.messagezgen_ai.user.messagezgen_ai.assistant.messagezgen_ai.tool.messagez gen_ai.choicecHeZdZdZ d"ddddded d#fdZd$fd Zd%dZd&dZd dd  d'd Z d(d Z d)d Z d)dZ d)dZ d)dZd*dZd(dZd+dZd,dZd-dZd)dZd.dZddd/dZd(dZd0dZd1fd Zd2d3fd Zd)dZd4d Zd)d!ZxZS)5LiveKitLangSmithSpanProcessorzIEnriches LiveKit Agents' OTel spans with LangSmith-compatible attributes.Nz audio/ogg)api_keyprojectendpointaudio_path_provideraudio_mime_typestate_ttl_secondsc t ||f|||d|||_||_dfd } | |_| |_| |_| |_| |_| |_ | |_ | |_ y)aCreate the processor. Args: audio_path_provider: returns a local recording path to embed on the root (dev only); for production use :meth:`expect_recording` / :meth:`complete_recording`. audio_mime_type: default MIME type for embedded recordings. state_ttl_seconds: lifetime for per-conversation state. )r,r-r.c&ttS)N)maxsizettl)r DEFAULT_STATE_MAXSIZE)r1sr/var/www/html/Fitness-lenito-AI-main/venv/lib/python3.12/site-packages/langsmith/integrations/livekit/processor.py_cachez6LiveKitLangSmithSpanProcessor.__init__.._cachegs$9?PQ QN)returnr) super__init__r/_audio_mime_type_transcript_by_trace_deferred_root_by_trace_ended_session_traces_threads_awaiting_recording_pending_audio_by_thread_trace_by_thread_deferred_user_speaking_pending_user_transcripts) selfdownstream_processorr,r-r.r/r0r1kwargsr8 __class__s ` r7r<z&LiveKitLangSmithSpanProcessor.__init__Hs*       $7 / R H !MSH$@F"FLh(CI8%:@( H $JP&r9cDt|||||j|<y)uIAlso index thread→trace (so ``complete_recording`` can find the trace).N)r;_remember_thread_idrC)rFtrace_id thread_idrIs r7rKz1LiveKitLangSmithSpanProcessor._remember_thread_id}s" #Hi8+3i(r9c>|jddfd }y)u(Subscribe this processor to a LiveKit ``AgentSession``'s events. A realtime (speech-to-speech) model's user transcript arrives via the ``user_input_transcribed`` session event — never on a span — so the processor can't see it from spans alone, and the trace ends up with only the agent's turns. Call this once after creating the session to wire the transcript in:: processor = configure_livekit(...) session = AgentSession(llm=...) set_thread_id(conversation_id) processor.instrument_session(session, conversation_id) Each final transcript is paired FIFO with the next ``user_speaking`` span that has no transcript yet (we have no id to match a transcript to its exact span). No-op for the cascade pipeline, where the transcript already rides the STT ``user_turn`` span. Args: session: the LiveKit ``AgentSession`` to subscribe to. thread_id: the conversation id, matching :func:`set_thread_id`. user_input_transcribedcvt|ddr+jtt|ddxsdyy)Nis_finalF transcript)getattr_record_user_transcriptstr)evrFrMs r7_on_user_input_transcribedzTLiveKitLangSmithSpanProcessor.instrument_session.._on_user_input_transcribeds9r:u-,, NGB b$A$GR.r9N)rWrr:None)on)rFsessionrMrXs` ` r7instrument_sessionz0LiveKitLangSmithSpanProcessor.instrument_sessions!0 , -  . r9c4d|jt|<y)zHold the root span open until :meth:`complete_recording` supplies the audio. Call at conversation start (with egress); always pair with a ``complete_recording`` call. ``thread_id`` must match ``set_thread_id``. TN)rArV)rFrMs r7expect_recordingz.LiveKitLangSmithSpanProcessor.expect_recordings <@((Y8r9z recording.ogg)name mime_typec(|r3|t||xs |jd|jt|<|jj t|d|j jt|}||j|yy)aAttach an egress recording and release the held root. ``data`` is the recording bytes (embedded as an attachment), or ``None`` to release without audio. ``thread_id`` must match :meth:`expect_recording`. Safe to call before or after the session ends. r_datar`N) bytesr=rBrVrApoprCget_maybe_release)rFrMrcr_r`rLs r7complete_recordingz0LiveKitLangSmithSpanProcessor.complete_recordings d &?$*?*?=D ) )#i. 9 ((,,S^TB((,,S^<      ) r9c|jjj}|jj}|tk(r|j |y|t k(r|j|y|tvr|jdy|tk(r|j|y|tvr|jdy|tk(r|j|y|tk(r|j!|S|t"k(r\|jdd|j$|<|j'|j(j+||j-|y|dk(r|jdy|t.k(r|j1|y|t2k(r|jdy|jj4(t7|jr|j9||yy)NchainT eou_detectionllmF)spancontextrLr_ _STT_SPAN _handle_stt_LLM_INFERENCE_SPAN_handle_llm_request_LLM_WRAPPER_SPANSset_kind_TTS_INFERENCE_SPAN _handle_tts_TTS_WRAPPER_SPANS _TURN_SPAN _handle_turn_USER_SPEAKING_SPAN_handle_user_speaking _SESSION_SPANr@_flush_user_speaking_thread_id_by_tracerfrg _TOOL_SPAN _handle_tool_REALTIME_METRICS_SPANparentr _handle_root)rFtspanrLr_s r7 _dispatchz'LiveKitLangSmithSpanProcessor._dispatchs::%%..zz 9    U #<;( (  $ $U +87' ' NN7 #43( (   U #0/' ' NN7 #,+Z    e $('( (--e4 4 ] " NN7 #37D & &x 0  % %d&>&>&B&B8&L M    )_ $ NN7 #Z    e $+ + NN5 !  ZZ   &?5::+F   eX .r9c|jd|j|jjd|j t |jjd|jjd}|rE|j td|dg|j tt|g|jy ) u6STT (``user_turn``): audio input → transcribed text.rlgen_ai.request.modelgen_ai.provider.namelk.user_transcriptz Audio for: ""prompt completionN) rt set_model attributesrf set_providerr set_messagesr r rVexclude_from_message_view)rFrrRs r7rpz)LiveKitLangSmithSpanProcessor._handle_stts u ((,,-CDE  u//334JK L %%))*>?    *\*Q+GHI     +B3z?+S*T  U '')r9c|jdg}g}|jD]n}|jtk(r|j t d|2t j|jx}T|j t ||p|j|xsd|xsdt|jjd}|jt|t|jjd}|r|jdi||jDcgc])}|jtk7r|jt vr|+c}|jddycc}w)z``llm_request``: rebuild prompt/completion from the gen_ai.* events. The translated events are then stripped so the ingester doesn't render them twice. rlr(Nrrzlk.llm_metrics)rteventsr__LLM_CHOICE_EVENTappendr_LLM_EVENT_ROLESrfrrrrrr set_usage) rFrrreventroleproviderusagees r7rrz1LiveKitLangSmithSpanProcessor._handle_llm_requestsG u!# \\Ezz..!!":;"NO*..uzz::$G 6tUCD " &.DZ=O4P3    !1 2  -h78"%"2"2"6"67G"HI  EOO $e $\\ !vv**qvv=M/M !  Q s2.E1c||jd|j|jjdxs<|jjdxs|jjdxsd}|j t t |gtd|dg|j|jjd xs$t|jjd t|jjd }|jt|y ) uB``tts_request``: synthesize text → audio (an ``llm`` inference).rlz lk.input_textzlk.request.textzlk.textrSzGenerated audio for: "rrrzlk.tts_metricsN) rtrrrfrr rVr rrrrr)rFrtextrs r7rvz)LiveKitLangSmithSpanProcessor._handle_tts s% u '')     1 ##$56 ##I.  &s4y12/2Ha0PQR      !7 8 U,U-=-=-A-ABR-ST 4    !1 2  -h78r9c|jd|jvrdnd|jjd}|jjd}|jjj }|jj }|r:tt|}|j|g|j||||r;tt|}|j|g|j|||yy) zRender an ``agent_turn`` and append it to the running transcript. ``llm`` for a realtime model (the turn is the model call, usage stamped here); ``chain`` for cascade (the STT/LLM/TTS children carry their usage). lk.realtime_model_metricsrlrjz lk.user_inputzlk.response.textrrN) rtrrfrmrnrL start_timer rVr_append_transcriptr )rFr user_inputresponserLstartmsgs r7ryz*LiveKitLangSmithSpanProcessor._handle_turn9s 0E4D4DDE' %%))/: ##''(:;::%%.. %% $S_5C   se  ,  # #Hc5 9 )#h-8C   3%  0  # #Hc5 9 r9c|jj|xsg}|j||f||j|<y)zIAppend a message to the transcript the root rolls up, keyed for ordering.N)r>rfr)rFrLmessagesort_key conversations r7rz0LiveKitLangSmithSpanProcessor._append_transcriptPs@0044X>D" Xw/0.:!!(+r9c|jd|jjd}|yt|}|jj|}|rc|j d}|r||j|<n|jj |d|j |||j|y|jj|xsg}|j|||j|<y)u6Handle a ``user_speaking`` span — the realtime user turn. Deferred (``False``): stamp+export now if the transcript was already buffered, else hold it until one is fed (or flushed untouched at session end). Exported as-is (``True``) when there's no thread id to pair against. rjzlangsmith.metadata.thread_idNTrF) rtrrfrVrEre_apply_user_transcript_exportrDr)rFrthreadpendingrRhelds r7r{z3LiveKitLangSmithSpanProcessor._handle_user_speakingYs w!!%%&DE >V0044V<  QJ9@..v6..2264@  ' 'z : LL ++//7=2 E/3$$V,r9ct|}|jj|}|rc|jd}|r||j|<n|jj|d|j |||j |y|j j|xsg}|j|||j |<y)zPair a realtime transcript (from ``instrument_session``) with its span. Applies it to the oldest held ``user_speaking`` span for the thread, or buffers it if that span hasn't ended yet. rN)rVrDrfrerrrEr)rFrMrRtidrrrs r7rUz5LiveKitLangSmithSpanProcessor._record_user_transcriptvs )n++//4 HHQKE48,,S1,,00d;  ' 'z : LL  0044S9?Rz".5&&s+r9c|jd|rs||jd<t|}|j|g|j |j j j||j jyy)u9Render a fed transcript onto a ``user_speaking`` span as the user's turn. Unlike the cascade STT ``user_turn`` (which is excluded), this is the only record of the realtime user's words, so it's shown — as a plain ``user`` message. An empty transcript renders no fabricated I/O. rlrrN) rtrr rrrmrnrLr)rFrrRrs r7rz4LiveKitLangSmithSpanProcessor._apply_user_transcriptsu u 5?E  1 2$Z0C   se  ,  # # ""++S%**2G2G  r9c|yt|}|jj|dxsgD]}|j||jj|dy)zJExport held ``user_speaking`` spans untranscribed (no transcript arrived).N)rVrDrerrE)rFrMrrs r7r}z2LiveKitLangSmithSpanProcessor._flush_user_speakings\   )n1155c4@FBFE LL G &&**35r9cP|jd|jjd}|r|jdt ||jjd}||j ||jjd}||j |yy)z@``function_tool``: render as a proper ``tool`` run with its I/O.r)zlk.function_tool.name tool_namezlk.function_tool.argumentsNzlk.function_tool.output)rtrrf set_metadatarVset_tool_inputset_tool_output)rFrrargsoutputs r7rz*LiveKitLangSmithSpanProcessor._handle_tools v$$(()@A    {C N ;##$@A     &!!%%&?@    ! !& ) r9c|jd|jd|jdd|jdd|jdtdxsd ||j|<|j |y ) z?Mark the conversation root and defer it until the session ends.rjT ls_modalityaudiols_integrationlivekitls_integration_versionzlivekit-agentsrSN)rt set_root_spanrr r?rg)rFrrLs r7rz*LiveKitLangSmithSpanProcessor._handle_roots} w D! ='2 +Y7  $':;K'L'RPR 27$$X. H%r9Fforcec|jj|}|y|jj|}|s ||jvry|||jvry|jj |d|j ||j|||j|||j||j|y)uExport the deferred root once the session ended and audio is ready. Requires the root seen, the session ended, and no awaited recording. ``force`` skips the last two gates — the last-resort path at :meth:`shutdown` for a root that never completed. N) r?rfr~r@rAre_render_conversation_attach_audio_recording_attach_pending_audior_cleanup_trace)rFrLrrrs r7rgz,LiveKitLangSmithSpanProcessor._maybe_releases,,00: = ))--h7t999!f0P0P&P $$((48 !!%( $$U+    & &uf 5 U H%r9c|jj|jjjg}|syt |dDcgc]\}}| }}}|j |ycc}}w)zJSet the whole transcript (ordered by span start_time) as the root's input.Fc |dS)Nrr)rs r7zDLiveKitLangSmithSpanProcessor._render_conversation..sAaDr9)keyrT)r>rfrmrnrLsortedr)rFrentries_rmessagess r7rz2LiveKitLangSmithSpanProcessor._render_conversationsk++// 0B0B0K0KRP&,W.&IJ&IFAsC&IJ (+Ks A1c|jj|}|jj|d|j ||j j|d|f|j j|d|jj|d|jj|d|j|yyN) r~rfr>re_forget_thread_idr@rCrArBr})rFrLrs r7rz,LiveKitLangSmithSpanProcessor._cleanup_traces))--h7 !!%%h5 x( ""&&x6    ! ! % %fd 3  , , 0 0 >  ) ) - -fd ;  % %f - r9ct|jD]}|j|dt|jj D]}|j |t |y)uForce-export any still-held roots and user_speaking spans, then shut down. ``force_flush`` deliberately does not — a still-held root there is legitimately in progress, not a buffered export waiting to drain. TrN)listr?rgrDkeysr}r;shutdown)rFrLrrIs r7rz&LiveKitLangSmithSpanProcessor.shutdownse T99:H     5;477<<>?F  % %f -@ r9c"t||S)uEForce-flush the downstream — deferred root spans are NOT finalized.)r; force_flush)rFtimeout_millisrIs r7rz)LiveKitLangSmithSpanProcessor.force_flushsw">22r9c|jy|j}||jsy |j}|j ||j ||j y#t$rYywxYw)z>Embed audio from the local ``audio_path_provider`` file (dev).Nrb)r/exists read_bytes Exception _attach_audior_r=)rFr audio_path audio_bytess r7rz5LiveKitLangSmithSpanProcessor._attach_audio_recordings  # # + --/  Z%6%6%8  $//1K  ++     sA,, A87A8c|jj|d}|r|jdsy|j||d|d|dy)zCEmbed a recording supplied via :meth:`complete_recording` (egress).Nrcr_r`rb)rBrerfr)rFrrrs r7rz3LiveKitLangSmithSpanProcessor._attach_pending_audiosW//33FDAgkk&1   k*  r9ct|jjD]}|jds|j|}d|j dd}t |}|At ||jD]#\}}||jvs||j|<%||jvr||j|<t|jjdxs$t|jjd}|j|d|jvr=d |jvr.t|jd} | r|jd i| yyyy) zForward ``lk.*`` to ``langsmith.metadata.lk_*`` and normalize the provider. Scalars pass through; JSON-object blobs are flattened per field. Runs on every exported span, so it also covers spans no handler classified. zlk.zlangsmith.metadata..rNrz gen_ai.systemrzlangsmith.usage_metadatar) rrr startswithreplacerritemsrrfrrr) rFrrvalueflat_keyparsedr_valrrs r7 _pre_exportz)LiveKitLangSmithSpanProcessor._pre_exports ((--/0C>>%($$S)E,S[[c-B,CDH*51F!!EH"%'"ID#5#3#3314((. " 5+++).E  X &1$&    !7 8 G  0 0 4 4_ E F  8$ (5+;+; ;*%2B2BB*  !<=E(%( C C&2 . 3  $  ')r9r+N)0r __future__rcollections.abcrpathlibrtypingrrr cachetoolsr opentelemetry.sdk.tracer $langsmith._internal._package_versionr "langsmith._internal.voice._helpersr r r-langsmith._internal.voice.base_span_processorrr_helpersrrrrrrrrrr6rorqrsrurwrxr|rrrzrrr+rr9r7rs#***1D    #  # "34# "34    +% &! +!  $@)$>@)r9