
    j                     h   d dl mZmZmZ d dlmZmZ d dlmZ ddl	m
Z
mZ ddl	mZmZ  G d d	e
j                        Z G d
 de
j                        Z G d de
j                        Z G d de      Z G d de      Z G d de      Z G d de
j                        Z G d de
j                        Zy)    )UnioncastList)
span_writetrace_writeN   )base_batchersequence_splitter   )messagesencoder_helpersc                   b     e Zd Zdeej
                     fdZdej                  ddf fdZ xZ	S )CreateSpanMessageBatcherreturnc                    g }| j                   D ]  }|j                         }t        j                  |      }t	        j
                  |t        j                  j                         d      }|j                  t        j                  di |        t        j                  || j                        }g }|D ]'  }|j                  t        j                  |             ) |S )Nspanfields_to_anonymizeobject_typeitemsmax_payload_size_MBbatch )_accumulated_messagesas_payload_dict
dict_utilsremove_none_from_dictr   encode_and_anonymizer   CreateSpansBatchMessager   appendr   	SpanWriter
   split_into_batches_batch_memory_limit_mb)self
rest_spansitemspan_write_kwargscleaned_span_write_kwargsmemory_limited_batchesbatchesr   s           /Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/batching/batchers.py)_create_batches_from_accumulated_messageszBCreateSpanMessageBatcher._create_batches_from_accumulated_messages   s     24
..D $ 4 4 6(2(H(H!)% )8(L(L)$,$D$D$X$X$Z")%
 j22O5NOP / "3!E!E $ ; ;"

 +ENN8;;%HI ,     messageNc                 d    j                   | j                  fd       t        |         S )Nc                 6    | j                   j                   k(  S N)span_idxr0   s    r-   <lambda>z.CreateSpanMessageBatcher.add.<locals>.<lambda>*   s    QYY'//5Qr/   end_time_remove_matching_messagessuperaddr&   r0   	__class__s    `r-   r<   zCreateSpanMessageBatcher.add'   s-    '**+QRw{7##r/   )
__name__
__module____qualname__r   r   r!   r.   CreateSpanMessager<   __classcell__r>   s   @r-   r   r   
   s9    	h..	/8$855 $$ $ $r/   r   c                   b     e Zd Zdeej
                     fdZdej                  ddf fdZ xZ	S )CreateTraceMessageBatcherr   c                    g }| j                   D ]  }|j                         }t        j                  |      }t	        j
                  |t        j                  j                         d      }|j                  t        j                  di |        t        j                  || j                        }g }|D ]'  }|j                  t        j                  |             ) |S )Ntracer   r   r   r   )r   r   r   r   r   r    r   CreateTraceBatchMessager   r"   r   
TraceWriter
   r$   r%   )r&   rest_tracesr(   trace_write_kwargscleaned_trace_write_kwargsr+   r,   r   s           r-   r.   zCCreateTraceMessageBatcher._create_batches_from_accumulated_messages0   s     57..D!%!5!5!7)3)I)I"*& *9)M)M*$,$D$D$X$X$Z#*&
 {55S8RST / "3!E!E $ ; ;"

 +ENN8;;%HI , r/   r0   Nc                 d    j                   | j                  fd       t        |         S )Nc                 6    | j                   j                   k(  S r3   )trace_idr5   s    r-   r7   z/CreateTraceMessageBatcher.add.<locals>.<lambda>O   s    QZZ7CSCS5Sr/   r8   r=   s    `r-   r<   zCreateTraceMessageBatcher.addL   s-    '**+STw{7##r/   )
r?   r@   rA   r   r   rI   r.   CreateTraceMessager<   rC   rD   s   @r-   rF   rF   /   s9    	h..	/8$866 $4 $ $r/   rF   c                        e Zd Zdeeej                  ej                  ej                  f      f fdZ	deej                  ej                  ej                  f   ddfdZ
 xZS )(BaseAddFeedbackScoresBatchMessageBatcherr   c                      t         |          S r3   )r;   r.   )r&   r>   s    r-   r.   zRBaseAddFeedbackScoresBatchMessageBatcher._create_batches_from_accumulated_messagesU   s     w@BBr/   r0   Nc                 *   |j                   }t        |      }t        | j                        }||z   | j                  k\  r@| j                  |z
  }|d | }||d  }| xj                  |z  c_        |}| j	                          | xj                  |z  c_        y r3   r   lenr   _max_batch_sizeflushr&   r0   new_messagesn_new_messagesn_accumulated_messagesfree_space_in_accumulatormessages_that_fit_in_batchmessages_that_dont_fit_in_batchs           r-   r<   z,BaseAddFeedbackScoresBatchMessageBatcher.add`   s     }}\*!$T%?%?!@22d6J6JJ(,(<(<?U(U%)56P7P)Q&.:;T;U.V+&&*DD&:LJJL""l2"r/   )r?   r@   rA   r   r   r   !AddSpanFeedbackScoresBatchMessage"AddTraceFeedbackScoresBatchMessage$AddThreadsFeedbackScoresBatchMessager.   r<   rC   rD   s   @r-   rS   rS   T   s{    	C	667799;	

	C3667799;
3 
3r/   rS   c                   2    e Zd Zdeej
                     fdZy)(AddSpanFeedbackScoresBatchMessageBatcherr   c                 F    t        j                  | j                  d      gS NF)r   supports_batching)r   ra   r   r&   s    r-   r.   zRAddSpanFeedbackScoresBatchMessageBatcher._create_batches_from_accumulated_messages|   (     6600"'
 	
r/   N)r?   r@   rA   r   r   ra   r.   r   r/   r-   re   re   y   s    
	h88	9
r/   re   c                   2    e Zd Zdeej
                     fdZy))AddTraceFeedbackScoresBatchMessageBatcherr   c                 F    t        j                  | j                  d      gS rg   )r   rb   r   ri   s    r-   r.   zSAddTraceFeedbackScoresBatchMessageBatcher._create_batches_from_accumulated_messages   s(     7700"'
 	
r/   N)r?   r@   rA   r   r   rb   r.   r   r/   r-   rl   rl      s    
	h99	:
r/   rl   c                   2    e Zd Zdeej
                     fdZy)+AddThreadsFeedbackScoresBatchMessageBatcherr   c                 F    t        j                  | j                  d      gS rg   )r   rc   r   ri   s    r-   r.   zUAddThreadsFeedbackScoresBatchMessageBatcher._create_batches_from_accumulated_messages   s(     9900"'
 	
r/   N)r?   r@   rA   r   r   rc   r.   r   r/   r-   ro   ro      s    
	h;;	<
r/   ro   c                   b     e Zd Zdeej
                     fdZdej
                  ddf fdZ xZS )GuardrailBatchMessageBatcherr   c                     g }| j                   D ]7  }t        t        j                  |      j                  }|j                  |       9 t        j                  |d      gS rg   )r   r   r   GuardrailBatchMessager   extend)r&   r   batch_messagebatch_of_message_items       r-   r.   zFGuardrailBatchMessageBatcher._create_batches_from_accumulated_messages   s]     !77M$(..%e " LL./	 8 ..UeTUUr/   r0   Nc                 "    t         |   |      S r3   )r;   r<   r=   s     r-   r<   z GuardrailBatchMessageBatcher.add   s    w{7##r/   )	r?   r@   rA   r   r   rt   r.   r<   rC   rD   s   @r-   rr   rr      s;    V	h,,	-V$899 $d $ $r/   rr   c                   V    e Zd Zdeej
                     fdZdej
                  ddfdZy)(CreateExperimentItemsBatchMessageBatcherr   c                 F    t        j                  | j                  d      gS rg   )r   !CreateExperimentItemsBatchMessager   ri   s    r-   r.   zRCreateExperimentItemsBatchMessageBatcher._create_batches_from_accumulated_messages   rj   r/   r0   Nc                 *   |j                   }t        |      }t        | j                        }||z   | j                  k\  r@| j                  |z
  }|d | }||d  }| xj                  |z  c_        |}| j	                          | xj                  |z  c_        y r3   rV   rZ   s           r-   r<   z,CreateExperimentItemsBatchMessageBatcher.add   s     }}\*!$T%?%?!@22d6J6JJ(,(<(<?U(U%)56P7P)Q&.:;T;U.V+&&*DD&:LJJL""l2"r/   )r?   r@   rA   r   r   r|   r.   r<   r   r/   r-   rz   rz      s5    
	h88	9
3AA3	3r/   rz   )typingr   r   r   opik.rest_api.typesr   r   opik.dict_utilsr    r	   r
   r   r   BaseBatcherr   rF   rS   re   rl   ro   rr   rz   r   r/   r-   <module>r      s    $ $ 7 $ - ("$|77 "$J"$ 8 8 "$J"3|/G/G "3J
,

,

,
$<#;#; $$3|/G/G 3r/   