~tj`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/d#S)$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.choiceceZdZdZ dEdddddeddFfdZdGfd ZdHdZdIdZddddJd#Z dKd'Z dLd(Z dLd)Z dLd*Z dLd+ZdMd/ZdKd0ZdNd2ZdOd3ZdPd4ZdLd5ZdQd6Zd7d8dRd:ZdKd;ZdSd<ZdTfd= ZdUdVfd@ ZdLdAZdWdCZdLdDZxZS)XLiveKitLangSmithSpanProcessorzIEnriches LiveKit Agents' OTel spans with LangSmith-compatible attributes.Nz audio/ogg)api_keyprojectendpointaudio_path_provideraudio_mime_typestate_ttl_secondsdownstream_processorOptional[SpanProcessor]r, Optional[str]r-r.r/&Optional[Callable[[], Optional[Path]]]r0strr1floatkwargsrreturnNonec ^tj|f|||d|||_||_dfd } | |_| |_| |_| |_| |_| |_ | |_ | |_ dS)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.r9rc0ttS)N)maxsizettl)r DEFAULT_STATE_MAXSIZE)r1sf/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/langsmith/integrations/livekit/processor.py_cachez6LiveKitLangSmithSpanProcessor.__init__.._cachegs$9?PQQQ QN)r9r) 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) selfr2r,r-r.r/r0r1r8rA __class__s ` r@rDz&LiveKitLangSmithSpanProcessor.__init__Hs*          $7 / R R R R R R FHH !MSFHH$@F"FLfhh(CI688%:@&(( FHH $JP&&&rBtrace_idint thread_idc`t||||j|<dS)uIAlso index thread→trace (so ``complete_recording`` can find the trace).N)rC_remember_thread_idrK)rNrPrRrOs r@rTz1LiveKitLangSmithSpanProcessor._remember_thread_id}s0 ##Hi888+3i(((rBsessioncN|ddfd }dS) 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_transcribedevrr9r:ct|ddr6tt|ddpddSdS)Nis_finalF transcript)getattr_record_user_transcriptr6)rXrNrRs r@_on_user_input_transcribedzTLiveKitLangSmithSpanProcessor.instrument_session.._on_user_input_transcribeds_r:u-- ,, NNGB b$A$A$GR  rBN)rXrr9r:)on)rNrUrRr_s` ` r@instrument_sessionz0LiveKitLangSmithSpanProcessor.instrument_sessionsK0 , - -       . -   rBc4d|jt|<dS)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)rIr6)rNrRs r@expect_recordingz.LiveKitLangSmithSpanProcessor.expect_recordings <@(Y888rBz recording.ogg)name mime_typedataOptional[bytes]rdrec8|r/|t||p|jd|jt|<|jt|d|jt|}|||dSdS)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. rdrfreN) bytesrErJr6rIpoprKget_maybe_release)rNrRrfrdrerPs r@complete_recordingz0LiveKitLangSmithSpanProcessor.complete_recordings  d &?$*?==D )#i.. 9 (,,S^^TBBB(,,S^^<<      ) ) ) ) ) rBtspanrboolc|jjj}|jj}|tkr||n|t kr||n|tvr| dn|tkr| |ne|tvr| dnE|tkr||n#|tkr||S|t"krb| dd|j|<||j|||n|dkr| dnz|t.kr||nY|t2kr| dn8|jj,t7|jr|||dSdS)NchainT eou_detectionllmF)spancontextrPrd _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_SPANrH_flush_user_speaking_thread_id_by_tracerlrm _TOOL_SPAN _handle_tool_REALTIME_METRICS_SPANparentr _handle_root)rNrorPrds r@ _dispatchz'LiveKitLangSmithSpanProcessor._dispatchs :%.z 9     U # # # # ( ( (  $ $U + + + + ' ' ' NN7 # # # # ( ( (   U # # # # ' ' ' NN7 # # # # Z     e $ $ $ $ ( ( (--e44 4 ] " " NN7 # # #37D &x 0  % %d&>&B&B8&L&L M M M    ) ) ) ) _ $ $ NN7 # # # # Z     e $ $ $ $ + + + NN5 ! ! ! ! Z  &?5:+F+F &   eX . . .5trBc|d||jd|t |jd|jd}|rY|td|dg|tt|g| d S) u6STT (``user_turn``): audio input → transcribed text.rtgen_ai.request.modelgen_ai.provider.namelk.user_transcriptz Audio for: ""prompt completionN) r| set_model attributesrl set_providerr set_messagesr r r6exclude_from_message_view)rNror[s r@rxz)LiveKitLangSmithSpanProcessor._handle_stts  u (,,-CDDEEE  u/334JKK L L   %))*>??  V   *+G*+G+G+GHHI       +B3z??+S+S*T  U U U '')))))rBc|dg}g}|jD]z}|jtkr$|t d|6t |jx}#|t ||{||pd|pdt|j d}| t|t|j d}|r |jdi|d|jD|jdd<dS)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. rtr(Nrrzlk.llm_metricscNg|]"}|jtk|jtv |#S)rd_LLM_CHOICE_EVENT_LLM_EVENT_ROLES).0es r@ zELiveKitLangSmithSpanProcessor._handle_llm_request..s=   v***qv=M/M/M /M/M/MrBr)r|eventsrdrappendrrrlrrrrrr set_usage)rNrorreventroleproviderusages r@rzz1LiveKitLangSmithSpanProcessor._handle_llm_requestsp u!# \ E EEz...!!":;"N"NOOOO*..uz:::$G 6tUCCDDD &.DZ=O4PPP3   !1 2 2   -h77888"%"2"6"67G"H"HII  % EO $ $e $ $ $  \    QQQrBc|d||jdp5|jdp|jdpd}|t t |gtd|dg||jd p&t|jd t|jd }| t|d S) uB``tts_request``: synthesize text → audio (an ``llm`` inference).rtz lk.input_textzlk.request.textzlk.textr\zGenerated audio for: "rrrzlk.tts_metricsN) r|rrrlrr r6r rrrrr)rNrotextrs r@r~z)LiveKitLangSmithSpanProcessor._handle_tts sf u '')))    1 1 ##$566 ##I..  &s4yy112/0P0P0P0PQQR       !7 8 8 U,U-=-A-ABR-S-STT   4   !1 2 2   -h7788888rBc|d|jvrdnd|jd}|jd}|jjj}|jj}|rJtt|}| |g| ||||rLtt|}| |g| |||dSdS) 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_metricsrtrrz lk.user_inputzlk.response.textrrN) r|rrlrurvrP start_timer r6r_append_transcriptr )rNro user_inputresponserPstartmsgs r@rz*LiveKitLangSmithSpanProcessor._handle_turn9s 0E4DDDEE'   %))/:: #''(:;;:%. %  :$S__55C   se  , , ,  # #Hc5 9 9 9  :)#h--88C   3%  0 0 0  # #Hc5 9 9 9 9 9 : :rBmessagedictsort_keyc|j|pg}|||f||j|<dS)zIAppend a message to the transcript the root rolls up, keyed for ordering.N)rFrlr)rNrPrr conversations r@rz0LiveKitLangSmithSpanProcessor._append_transcriptPsI044X>>D" Xw/000.:!(+++rBc |d|jd}|dSt|}|j|}|rj|d}|r ||j|<n|j|d|||||dS|j|pg}| |||j|<dS)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. rrzlangsmith.metadata.thread_idNTrF) r|rrlr6rMrk_apply_user_transcript_exportrLr)rNrothreadpendingr[helds r@rz3LiveKitLangSmithSpanProcessor._handle_user_speakingYs  w!%%&DEE >4V044V<<   QJ A9@.v66.2264@@@  ' 'z : : : LL   5+//77=2 E/3$V,urBr[ct|}|j|}|rj|d}|r ||j|<n|j|d|||||dS|j|pg}||||j|<dS)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)r6rLrlrkrrrMr)rNrRr[tidrrors r@r^z5LiveKitLangSmithSpanProcessor._record_user_transcriptvs )nn+//44  HHQKKE <48,S11,00d;;;  ' 'z : : : LL    F044S99?Rz""".5&s+++rBc|d|rb||jd<t|}||g||jjj||jjdSdS)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. rtrrN) r|rr rrrurvrPr)rNror[rs r@rz4LiveKitLangSmithSpanProcessor._apply_user_transcripts u  5?E 1 2$Z00C   se  , , ,  # # "+S%*2G        rBc|dSt|}|j|dpgD]}|||j|ddS)zJExport held ``user_speaking`` spans untranscribed (no transcript arrived).N)r6rLrkrrM)rNrRrros r@rz2LiveKitLangSmithSpanProcessor._flush_user_speakingss   F)nn155c4@@FB  E LL     &**355555rBcv|d|jd}|r#|dt ||jd}||||jd}|||dSdS)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)r|rrl set_metadatar6set_tool_inputset_tool_output)rNrorargsoutputs r@rz*LiveKitLangSmithSpanProcessor._handle_tools v$(()@AA  <   {C NN ; ; ;##$@AA     & & &!%%&?@@    ! !& ) ) ) ) )  rBc:|d|d|dd|dd|dtdpd ||j|<||d S) z?Mark the conversation root and defer it until the session ends.rrT ls_modalityaudiols_integrationlivekitls_integration_versionzlivekit-agentsr\N)r| set_root_spanrr rGrm)rNrorPs r@rz*LiveKitLangSmithSpanProcessor._handle_roots w D!!! ='222 +Y777  $':;K'L'L'RPR   27$X. H%%%%%rBFforcerc|j|}|dS|j|}|s||jvrdS| ||jvrdS|j|d||||||||| || |dS)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) rGrlrrHrIrk_render_conversation_attach_audio_recording_attach_pending_audior_cleanup_trace)rNrPrrors r@rmz,LiveKitLangSmithSpanProcessor._maybe_releases,00:: = F)--h77 t999!f0P&P&P $((4888 !!%((( $$U+++    & &uf 5 5 5 U H%%%%%rBc|j|jjjg}|sdSdt |dD}||dS)zJSet the whole transcript (ordered by span start_time) as the root's input.Fcg|]\}}|Srr)r_rs r@rzFLiveKitLangSmithSpanProcessor._render_conversation..sJJJFAsCJJJrBc|dS)Nrr)rs r@zDLiveKitLangSmithSpanProcessor._render_conversation..s AaDrB)keyrT)rFrlrurvrPsortedr)rNroentriesmessagess r@rz2LiveKitLangSmithSpanProcessor._render_conversationsm+// 0B0KRPP 5JJfW..&I&I&IJJJ (+++trBc|j|}|j|d|||j|d|h|j|d|j|d|j|d| |dSdSN) rrlrFrk_forget_thread_idrHrKrIrJr)rNrPrs r@rz,LiveKitLangSmithSpanProcessor._cleanup_traces)--h77 !%%h555 x((( "&&x666    ! % %fd 3 3 3  , 0 0 > > >  ) - -fd ; ; ;  % %f - - - - -  rBc t|jD]}||dt|jD]}||t dS)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)listrGrmrLkeysrrCshutdown)rNrPrrOs r@rz&LiveKitLangSmithSpanProcessor.shutdowns T9:: 6 6H     5 5 5 547<<>>?? . .F  % %f - - - - rB0utimeout_milliscFt|S)uEForce-flush the downstream — deferred root spans are NOT finalized.)rC force_flush)rNrrOs r@rz)LiveKitLangSmithSpanProcessor.force_flushsww"">222rBc|jdS|}||sdS |}n#t$rYdSwxYw|||j||jdS)z>Embed audio from the local ``audio_path_provider`` file (dev).Nri)r/exists read_bytes Exception _attach_audiordrE)rNro audio_path audio_bytess r@rz5LiveKitLangSmithSpanProcessor._attach_audio_recordings  # + F--//  Z%6%6%8%8  F $//11KK    FF   +      sA AArc|j|d}|r|dsdS|||d|d|ddS)zCEmbed a recording supplied via :meth:`complete_recording` (egress).Nrfrdreri)rJrkrlr)rNrorrs r@rz3LiveKitLangSmithSpanProcessor._attach_pending_audiosz/33FDAA gkk&11  F  k*      rBct|jD]}|ds|j|}d|dd}t |}|>%(( $S)EDS[[c-B-BDDH*511F!!EH""%''55ID#5#33314(.5+++).E X & &&   !7 8 8  G  0 4 4_ E E F F  8$$$ (5+; ; ;*%2BBB* !<=E )((%((((( < ;BB  ) )rBr)r2r3r,r4r-r4r.r4r/r5r0r6r1r7r8rr9r:)rPrQrRr6r9r:)rUrrRr6r9r:)rRr6r9r:) rRr6rfrgrdr6rer4r9r:)rorr9rp)rorr9r:)rPrQrrrrr9r:)rRr6r[r6r9r:)rorr[r6r9r:)rRr4r9r:)rorrPrQr9r:)rPrQrrpr9r:)rPrQr9r:)r9r:)r)rrQr9rp)rorrr6r9r:)__name__ __module__ __qualname____doc__DEFAULT_STATE_TTL_SECONDSrDrTrarcrnrrxrzr~rrrr^rrrrrmrrrrrrr __classcell__)rOs@r@r+r+EsSS9=3R"&!%"&FJ*#<3R3R3R3R3R3R3R3Rj444444B@@@@$#' ******6####N****     B99992::::.;;;;:6666* 6666 * * * * & & & &>C&&&&&&2 . . . .      3333333     $     ')')')')')')')')rBr+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_helpersrrrrrrrrrr?rwryr{r}rrrrrrrrr+rrBr@rs#"""""****************111111DDDDDD                     #  # "34# "34    +% &! +!  $@)@)@)@)@)$>@)@)@)@)@)rB