o i'@sdZddlmZddlmZddlZddlmZddlZddlm Z e e Z GdddeZ Gd d d e Z Gd d d e ZGd ddejZGdddejZddee effddZdS)z This file contains shared utilities for tracing streams in LLMobs integrations. Integrations should implement a StreamHandler and / or AsyncStreamHandler subclass to be passed into the make_traced_stream factory function along with the stream to wrap. )ABC)abstractmethodN)Union) get_loggerc@s:eZdZddZddZddZddZed d d Zd S) BaseStreamHandlercKs8||_||_||_||_||_||fg|_||_dSN) integration primary_span request_argsrequest_kwargsoptionsspansinitialize_chunk_storagechunks)selfrspanargskwargsr rd/home/ubuntu/.local/lib/python3.10/site-packages/ddtrace/llmobs/_integrations/base_stream_handler.py__init__s zBaseStreamHandler.__init__cCsgSrrrrrrrsz*BaseStreamHandler.initialize_chunk_storagecCs|j||fdS)z Add a span to the list of spans to be finished when the stream ends. This is useful for integrations that need to create multiple spans for a single stream like LiteLLM. N)r append)rrrrrradd_span"szBaseStreamHandler.add_spancCs|jr |jjtdSdS)z Handle exceptions that occur during streaming. Default implementation sets exception info on the primary span. Args: exception: The exception that occurred N)r set_exc_infosysexc_infor exceptionrrrhandle_exception)s z"BaseStreamHandler.handle_exceptionNcCtd)a* Finalize the stream and complete all spans. This method is called when the stream ends (successfully or with error). Implementations should: 1. Process accumulated chunks into final response 2. Set appropriate span tags 3. Finish all spans z3finalize_stream must be implemented by the subclassNotImplementedErrorrrrrfinalize_stream5s z!BaseStreamHandler.finalize_streamr) __name__ __module__ __qualname__rrrrrr#rrrrrs  rc@eZdZdZedddZdS) StreamHandlera Instances of StreamHandler and AsyncStreamHandler contain the logic for initializing chunk storage, processing chunks, handling exceptions, and finalizing a stream. The only methods that need to be implemented are process_chunk (a callback function that is called for each chunk in the stream) and finalize_stream (a callback function that is called when the stream ends). All other methods are optional and can be overridden if needed (for example, extra exception processing logic in handle_exception). Note that it is possible to pass in extra arguments via the options argument in case you need to access other information within the stream handler that is not covered by the existing arguments. NcCr )a; Process a single chunk from the stream. This method is called for each chunk as it's received. Implementations should extract and store relevant data. Args: chunk: The chunk object from the stream iterator: The sync iterator object from the stream 1process_chunk must be implemented by the subclassr!rchunkiteratorrrr process_chunkOs zStreamHandler.process_chunkrr$r%r&__doc__rr-rrrrr(Cs r(c@r')AsyncStreamHandlerz) Async version of StreamHandler. Ncs td)a< Process a single chunk from the stream. This method is called for each chunk as it's received. Implementations should extract and store relevant data. Args: chunk: The chunk object from the stream iterator: The async iterator object from the stream r)r!r*rrrr-cs z AsyncStreamHandler.process_chunkrr.rrrrr0^sr0cTeZdZdZddeffdd ZddZdd Zd d Zd d Z e ddZ Z S) TracedStreama The TracedStream and AsyncTracedStream classes are wrappers around the underlying stream object that deal with iterating over the stream and calling the stream handler to process chunks, handle exceptions, and finalize the stream. Because each library's streamed response is different, these traced stream classes are meant to be generic enough to work with iterables, iterators, generators, and context managers. Nhandlerc$t|||_||_|j|_dS)a Wrap a stream object to trace the stream. Args: wrapped: The stream object to wrap handler: The StreamHandler instance to use for processing chunks on_stream_created: In the case that the stream is created by a stream manager, this callback function will be called when the underlying stream is created in case modifications to the stream object are needed N)superr _self_handler_self_on_stream_created __wrapped___self_stream_iterrwrappedr3on_stream_created __class__rrrz  zTracedStream.__init__c cszd}z2z|jD] }|j||j|VqWnty,}z |}|j|d}~wwW|j|dS|j|wr)r9r6r- Exceptionrr#rexcr+errr__iter__s  zTracedStream.__iter__c Cslz|j}|j||j|WSty|jty5}z |j||j|d}~wwr)r9__next__r6r- StopIterationr#r@rrr+rCrrrrEs     zTracedStream.__next__cCsB|j}||jur |S||_t||j|j}|jr|||S)a Enter the context of the stream. If the stream is wrapped by a stream manager, the stream manager will be entered and the underlying stream will be wrapped in a TracedStream object. We retain a reference to the underlying stream object to be consumed. If the stream is not wrapped by a stream manager, the stream will be returned as is. )r8 __enter__r9r2r6r7rresult traced_streamrrrrHs  zTracedStream.__enter__cCs|j|||Sr)r8__exit__rexc_typeexc_valexc_tbrrrrLszTracedStream.__exit__cC|jSrr6rrrrr3zTracedStream.handlerr) r$r%r&r/r(rrDrErHrLpropertyr3 __classcell__rrr=rr2rs  r2cr1)TracedAsyncStreamz( Async version of TracedStream. Nr3cr4)a Wrap an async stream object to trace the stream. Args: wrapped: The stream object to wrap handler: The AsyncStreamHandler instance to use for processing chunks on_stream_created: In the case that the stream is created by a stream manager, this callback function will be called when the underlying stream is created in case modifications to the stream object are needed N)r5rr6r7r8_self_async_stream_iterr:r=rrrr?zTracedAsyncStream.__init__c Csd}z:z|j2z3dHW}|j||jIdH|Vq6Wnty4}z |}|j|d}~wwW|j|dS|j|wr)rWr6r-r@rr#rArrr __aiter__s zTracedAsyncStream.__aiter__c szz|jIdH}|j||jIdH|WSty$|jty<}z |j||j|d}~wwr)rW __anext__r6r-StopAsyncIterationr#r@rrGrrrrYs    zTracedAsyncStream.__anext__csJ|jIdH}||jur|S||_t||j|j}|jr#|||S)a Enter the context of the stream. If the stream is wrapped by a stream manager, the stream manager will be entered and the underlying stream will be wrapped in a TracedAsyncStream object. We retain a reference to the underlying stream object to be consumed. If the stream is not wrapped by a stream manager, the stream will be returned as is. N)r8 __aenter__rWrVr6r7rIrrrr[s  zTracedAsyncStream.__aenter__cs|j|||IdHSr)r8 __aexit__rMrrrr\szTracedAsyncStream.__aexit__cCrQrrRrrrrr3rSzTracedAsyncStream.handlerr) r$r%r&r/r0rrXrYr[r\rTr3rUrrr=rrVs  rVr3cCs"t|tr t|||St|||S)zf Create a TracedStream or TracedAsyncStream object from a stream object and a stream handler. ) isinstancer0rVr2)r;r3r<rrrmake_traced_stream s   r^r)r/abcrrrtypingrwraptddtrace.internal.loggerrr$logrr(r0 ObjectProxyr2rVr^rrrrs    /NK