qiddlmZddlmZddlmZddlmZddlZddl m Z ddl m Z d d l mZd d lmZmZd d lmZmZmZmZeGd dZGddZGddZGddZy)) dataclass)partial)count)AnyN)PretrainedConfig)traced)PagedAttentionCache) TMP_TOKEN_IDFutureRequestState)CudaGraphBufferaligned_divideattn_mask_is_neededbuild_attention_maskceZdZUdZej ed<ej eeej fzdzed<ej ed<ej ed<ej eeej fzed<e ed<e eee fzed <e ej ed <e ej ed <ej ed <e ed <dZ e ed<deeeffdZy)PagedAttentionArgsaDataclass containing the keyword arguments for a forward pass using paged attention. Attributes: input_ids: Input token IDs tensor of shape `(1, total_query_tokens)`. attention_mask: Attention mask tensor or dictionary mapping layer types to masks. Can be `None` if the attention implementation doesn't require explicit masks. position_ids: Position IDs tensor of shape `(1, total_query_tokens)`. cu_seq_lens_q: Cumulative sequence lengths for queries, used for variable-length batching. cu_seq_lens_k: Cumulative sequence lengths for keys/values. Can be a tensor or dictionary mapping layer types (e.g., "full_attention", "sliding_attention") to tensors for hybrid models. max_seqlen_q: Maximum query sequence length in the batch. max_seqlen_k: Maximum key/value sequence length. Can be an int or dictionary for hybrid models. write_index: List of tensors indicating where to write new KV states in the cache, one per attention group. read_index: List of tensors indicating which cache positions to read from, one per attention group. logits_indices: Tensor indicating which positions in the output should be used for next-token prediction. cache: The [`PagedAttentionCache`] instance managing the KV cache. use_cache: Whether to use caching (always `False` in continuous batching as the cache is managed externally). input_idsNattention_mask position_ids cu_seq_lens_q cu_seq_lens_k max_seqlen_q max_seqlen_k write_index read_indexlogits_indicescacheF use_cachereturnc |j|j|j|j|j|j |j |j|j|j|j|jd S)N rrrrrrrrrrrrr"selfs k/opt/pipecat/venv/lib/python3.12/site-packages/transformers/generation/continuous_batching/input_outputs.pyasdictzPagedAttentionArgs.asdict?sp"11 --!//!// -- --++//"11ZZ  )__name__ __module__ __qualname____doc__torchTensor__annotations__dictstrintlistr rboolrr&r'r%rrs&||LL4U\\(9#::TAA,,<<<<$sELL'8"999S#X&&ell##U\\""LL It S#X r'rc eZdZdZ ddededejdejde ddf d Z e d dd Z d d ddejjdeddfdZe ej"d deddfdZdeej(eeej(fffdZdee e e ee ee ffdZdej(ddfdZddZdeeeee ffdZe deeddfdZd!de de deeeffdZ y)"ContinuousBatchingIOszA class to hold inputs and outputs for a continuous batching forward pass, using static tensors as storage. The class is meant to be self-contained, so once a set of inputs have been created, the class can be used to update the batch alone. rconfigdevice model_dtype max_graphsr Ncf||_||_||_||_t |dddn |j |_d|_d|_d|_t|jDcgc]}dc}|_ t|jDcgc]}dc}|_ g|_ i|_t||_|j#|j%d|j&dk(r0t(j*j-|j|_yd|_ycc}wcc}w) aInitialize the continuous batching I/O manager. Args: - cache: The [`PagedAttentionCache`] instance managing the KV cache. Meant to be unique. - config: The model's pretrained configuration. - device: The device to allocate tensors on. If the device is CPU, then the memory is pinned. - model_dtype: The data type for model computations. - max_graphs: Maximum number of CUDA graphs to cache. Uses LRU eviction when full. sliding_windowNr rT) full_resetcudar8)rr8r7r9getattrr<actual_query_lengthactual_key_lengthactual_batch_sizerange num_groupsactual_read_sizesactual_write_sizesrequests_in_batchreq_id_to_new_token_positionrgraphs_setup_static_tensors_reset_static_tensorstyper,r>Streamcompute_streamr$rr7r8r9r:_s r%__init__zContinuousBatchingIOs.__init__Vs     &#*63CT#J#RaX^XmXm#$ !"!"-253C3C-D!E!!E.3E4D4D.E"F1"F;=<>)'6z'B  ""$ ""d"3GM{{V\G\ejj//t{{/Cbf"F"Fs ) D) D.T) standalonec|jj}|jj}|jj|jjz}|j j dk(}t|dzdd}tjd|ftj|j ||_ |jdd|f|_ |jdd|f|_ |jdd|dzf|_|jd d|f|_|jd d|dzf}|jd d|dzf}|jd d|f|_i|_|jj$r||j"d <|jj&r||j"d<tj|dzftj|j ||_|j(j+d|_d|_t0j3|j"j5d|_t9|j:rhi|_|j"j5D]C}tjdd|||zf|j>|j ||j<|<End|_tj||ftj|j ||_ tj|||zftj|j ||_!y)aAllocates static tensors for generation inputs and outputs. This is called only once at init time, to avoid repeated allocations and enable CUDA graphs. All tensors are allocated with maximum possible sizes. The allocated tensors are: - `_bulk_input_tensor`: Storage for all the small inputs: `input_ids`, `position_ids`, `cumulative_seqlens_q`, `logits_indices`, `cumulative_seqlens_k`, `carry_over_ids`. - `attention_mask`: Optional attention masks (only for eager/SDPA implementations) - `write_index` and `read_index` storage: Cache indexing tensors for each attention group - `output_ids`: Storage for generated token IDs cpur )dtyper8 pin_memoryrNrfull_attentionsliding_attention)sizerXr8rY)"rrEmax_batch_tokens num_blocks block_sizer8rMrr,emptyint32_bulk_input_tensorrrcumulative_seqlens_qrcarry_over_idscumulative_seqlens_knum_full_attention_groupsnum_sliding_attention_groups output_idszero_total_seqlen_qrr/fromkeyskeysrrr7rr9write_index_storageread_index_storage) r$rEra num_pagesrY bulk_size#full_attention_cumulative_seqlens_k&sliding_attention_cumulative_seqlens_k layer_types r%rKz+ContinuousBatchingIOs._setup_static_tensorszs ZZ** ::66JJ))DJJ,A,AA [[%%. ##3a#7B? "'++  N%++dkkj# 004E5E4E1EF 33A7H8H7H4HI$($;$;A?UAQTUAU?U@! :: / /:]D % %&6 7 :: 2 2=cD % %&9 : ++  ! #5;;t{{Wa   MM$*C*C*H*H*JAN t{{ +"$D "77<<>  27++Q 0)>N2NO**;;) 3##J/ #'D $);; ) *%++dkk^h$  #(++ %55 6ekkRVR]R]jt# r'otherstream non_blockingch|j|_|j|_|j|_|jdd|_|jdd|_|j |_|j |_t|jj|_tjj|5|jj|j||jj|j||j j|j ||j"X|j"L|j"j%D]/}|j"|j|j"||1dddy#1swYyxYw)Nrz)rArBrCrFrGrnrr/ritemsr,r>ryrfcopy_rqrrrrp)r$rxryrzrws r%_transfer_inputsz&ContinuousBatchingIOs._transfer_inputssx%)$<$<!"&"8"8"&"8"8"&"8"8";#'#:#:1#= #22!..!$"3"3"9"9";< ZZ  v & w  $ $ * *4+B+BQ] * ^  % % + +D,D,DS_ + `  $ $ * *4+B+BQ] * ^"".53G3G3S"&"5"5":":"<wJ((4::4;N;Nz;Ziu:vw  w w ws CF((F1r=c|r|jjdn |j}|r|jjdn |j}|j ddd|dzfj d|_|jd|j |jd|j |jD]r}d|j|<|j|j|ddddd|d|fjtj|j j"t|jddd|fjd|jddd||zfjdy)aAReset static tensors for the next batch. For efficiency, this only resets the portions of tensors that were actually used in the previous batch, using the attributes actual_query_length, actual_key_length, and actual_batch_size. If a (full_reset) is requested, the entire tensor storage is reset. Nr r)rqr`rArrrBrfrmrrrlrirrfill_r,finfor9min)r$r=q_lenk_lenrws r%rLz+ContinuousBatchingIOs._reset_static_tensorssg6@((--b1TE]E]4>'',,R0DDZDZ ;UQY;/557 FU#))+ %%'33 oJ,-D  j )"".##J/1fuffuf0DEKKEKKX\XhXhLiLmLmn o   FUF+11"5 ?UU]? 2399"=r'c2|j|jfS)z:Get the cumulative sequence lengths for the current batch.)rgrir#s r%get_cumulative_seqlensz,ContinuousBatchingIOs.get_cumulative_seqlenss(($*C*CCCr'ct|j|j|j|j|jfSN)rArBrCrFrGr#s r%get_actual_lengthsz(ContinuousBatchingIOs.get_actual_lengthss9  $ $  " "  " "  " "  # #   r'rcyrr4)r$rs r%carry_over_tokensz'ContinuousBatchingIOs.carry_over_tokenss r'cR|j|jjyyr)rO synchronizer#s r%retrieve_device_outputsz-ContinuousBatchingIOs.retrieve_device_outputss%    *    + + - +r'c|j}|jdt|jj}||fSr)rHrllentolist)r$rH new_tokenss r%prepare_batch_updatez*ContinuousBatchingIOs.prepare_batch_update s> 22__%Bs4+A+A'BCJJL  *,,r'rHcH |s td|jd|_d|_d|_t |j jDcgc]}dc}|_t |j jDcgc]}dc}|_ g|_ i|_ g}g}dg}g}|jjDcic]}|dg}}t |j jDcgc]}g} }t |j jDcgc]}g} }|D] } | j} | j} t!| j"}|j j%| |}|xj|z c_|xjt'|j)z c_|xjdz c_| xj|z c_|j+| j"|j+t | | |z|j-|d|zt'|j.||_|j1D]H\}}||j-||d|zt'|j2|||j2|<J|j j5| j6| || | | j8r?|j-|ddz t:g| _|d|j| j6<|jj-| t=t>j@t>jB|jD}|||jFdt!||||jHdt!||||jJdt!||||jLdt!||d|_'|j1D]`\}}|||j|dt!||jP4tS|jP||||dk(r |jTndbg|_+g|_,t[t]| | D]z\}}}|||j^|dt!|f<|||j`|dt!|f<t!||j|<t!||j|<|ycc}wcc}wcc}wcc}wcc}w) aPrepare tensors and metadata for the next model forward pass, using the given requests as data. This method: 1. Resets the static tensors from the previous batch 2. Iterates through requests to accumulate input_ids, position_ids, and sequence lengths 3. Extends read/write indices for cache management 4. Builds attention masks if needed (for eager/SDPA implementations) 5. Converts accumulated lists to tensors and copies them to static storage This method also modifies the `position_offset` attribute of each request to track progress and adds a temporary token at the end of the requests for which there will a new token. zNo requests in batchrr r)rXr8Nr_)rrgrir<)1 ValueErrorrLrArBrCrDrrErFrGrHrIrirpstateposition_offsetrtokens_to_process get_seqlens_kmaxvaluesextendappendrr}rextend_read_and_write_indices request_id has_new_tokenr rr,tensorrer8rrrgrrnrrr<rrziprrrrq)r$rHrQrrrgrrwrirr future_stater past_length query_length seqlens_klayer_type_seqlen_k to_tensorlayer_type_seqlens_kigroup_read_indicesgroup_write_indicess r%prepare_batch_tensorsz+ContinuousBatchingIOs.prepare_batch_tensorss!34 4 ""$#$ !"!"-24::3H3H-I!J!!J.3DJJ4I4I.J"K1"K!#,.)  !sBFB[B[B`B`BbcJ QCcc"' (=(=">?Qb? ?#()>)>#?@ar@ @.% 8L &&E//Ku667L 00lKI  $ $ 4 $  " "c)*:*:*<&= = "  " "a ' "  ! !\ 1 !   U44 5   k;3M N O ' '(&BC+7.'FTUWFX11%2B2BC  " " ) ), 7K% 8RELL DKKP ,5Y+?'Y(1:<1H-C -.AJK_A`!!"=C(<$=>5>~5N1c.122261E0J0J0L  ,J,QZ[oQpD % %j 12MC8L4M N"".$#'#6#6z#B)=)=:DH[:[4#6#6ab  :=egzS^:_ B 6A!#6DMN`DaD # #A'@-?)@'@$@ AFOPcFdD $ $Q(B#.A*B(B%B C(+,>(?D " "1 %),-@)AD # #A &  B_"K"K d?@s T  T T8 T$ T padded_q_sizepadded_kv_cache_sizec.|dkDxr|dkD}|r|n |j}|r|n |j}||z}t|jd|j d|j d|j d|j d|dz|j|jd|iiigg|jd }|rUt|j||jz |_|j|_||j |jdzdt|jjD]}|r|n|j|} |r|n|j|} |j j#|j$|d| f|j&j#|j(|d| ft+|j,j/} t1| dkDri|_i|_i|_|j,j9D]r\} } | d|dz|j4| <|j2| |j2| <|j6D|r|n| |}|j6| dd|d|f|j6| <tns| d} |j,| d|dz|_|j2| |_|j64|r|n|j,| |}|j6| dd|d|f|_|j6d|_|j;S)a6Get model keyword arguments for the current batch, eventually padding the query dimension to (padded_q_size) and the keys/values dimension to (padded_kv_cache_size). The padding is only useful if we want static shapes, like when using cuda graphs AND only activated if both Q and KV are padded.rNr F) rrrrrrrrrrrr.)rArCrr unsqueezerrgrrrrrnrDrErFrGrrrrrrqr2rirprrrrr}r&)r$rr use_paddingrb_sizepadded_kv_sizekwargsrread_index_sizewrite_index_size layer_typesrwrrs r%get_model_kwargsz&ContinuousBatchingIOs.get_model_kwargszsT $a'D,@1,D !, $2J2J"-43I3I&)==$nnVe,66q9**6E2<&>qBSCSBS?S&T U  V44499;< { a 24F <>F =?F !)-)B)B)H)H)J m% I3<\vz3J$$Z0262C2CJ2O##J/&&2.9Ny?PE8<8K8KJ8WX[]c^c]cekfkekXk8lF))*5  m%QJ#'#<#rNr3rno_gradrLtupler-r/r0rr2rrrr rrrrr4r'r%r6r6Ps "g""g!"g "g [[ "g  "g "gH tC C N_dw,w6;jj6G6GwW[w w, U]]_>>> >8DellDellAREventh2d_over compute_overd2h_over)r$rr7r8r9r:s r%rRzHostDeviceIOPair.__init__su-UFELLryrrlr~rrs r%transfer_outputs_d2hz%HostDeviceIOPair.transfer_outputs_d2hsU ZZ  v & X LL # # ) )$..*C*CRV ) W X X Xs rNrrr4r'r%rrs +" +! + + [[ +  +  +X%***;*;XXX5::+<+<XXr'rc eZdZdZ ddededejdejde ddf d Z de eje eejfffd Zde e e e ee ee ffd Zd eeddfd ZdejfdZdde de de eeffdZdejddfdZedejfdZedefdZddZde eeee ffdZy)ContinuousBatchingAsyncIOsu A class to handle the inputs and outputs for the asynchronous API. It uses two IO pairs to avoid race conditions between the two batches, which means twice as more VRAM is used for static input tensors and CUDA graph. If your GPU is large enough or you want to generate long sequences, this is a good trade-off to make. Asynchronous batching works by creating two pairs of host - device inputs and ouputs: inputs ┌──────────┐ ────────► ┌────────────┐ IO pair object: │ Host IOs │ │ Device IOs │ (for a CUDA sytem, Host = CPU and Device = GPU) └──────────┘ ◄──────── └────────────┘ outputs Each pair is separate from the other. This means that each pairs has its own CUDA graphs set, because CUDA graphs need to have static adresses for input tensors. To have a unique set of CUDA graph, we would need to copy the input tensors to a third device-side buffer. This could limit the memory cost of CUDA graphs but would slow down the forward pass. But the CUDA streams orchestrating the transfer from host to device (H2D) and device to host (D2H) are the same for both pairs. Same for the compute stream. The order of steps in async batching looks like this (for 3 batches of compute): │ ┌────┬────┐ ┌────┬────┐ ┌────┬────┐ ┌────┐ ┌────┐ CPU │ │PR 0│PR 1│ │UP 0│PR 2│ │UP 1│PR 3│ │UP 2│ │UP 3│ │ └────┼───┬┴──┐ └────┴────┼───┐ └────┴────┼───┐ └────┘ └────┘ H2D │ │0->│1->│ ¦ │2->│ ¦ │3->│ ¦ ¦ │ └───┼───┴───────────┬─────────────┴─┬─┼───────────┴───┼───────────────┐ ¦ GPU │ │ COMPUTE 0 │ COMPUTE 1 │█│ COMPUTE 2 │ COMPUTE 3 │ ¦ │ └───────────────┼───┬───────────┼─┴─┬─────────────┼───┬───────────┼───┤ D2H │ │0<-│ │1<-│ │2<-│ │3<-│ │ └───┘ └───┘ └───┘ └───┘ with: - CPU: actions happening on the CPU (host-side) - GPU: actions happening on the GPU (device-side) - H2D: host to device transfer - D2H: device to host transfer and: - PR N: preparation of batch N - ->N: host to device transfer of batch N - COMPUTE N: compute step for batch N - <-N: device to host transfer of batch N - UP N: update of batch N You can see that the GPU is almost always busy, execpt where the █ is. Proper ordering of steps is ensured through the use of CUDA events and streams. rr7r8r9r:r Nc bd|_tdDcgc]}t|||||c}|_tj j ||_tj j ||_tj j ||_ d|jdj_ d|jdj_ d|jdj_ d|jdj_ |j|_ ycc}w)NrrZr?r ) current_pairrDrio_pairsr,r>rN h2d_stream d2h_streamrOrrrarPs r%rRz#ContinuousBatchingAsyncIOs.__init__schijckl^_)%jYl **++6+:**++6+:#jj//v/>26 a  /48 a""126 a  /48 a""1 % 6 6msD,cd|j|jjjSr)rrrrr#s r%rz1ContinuousBatchingAsyncIOs.get_cumulative_seqlenss&}}T../77NNPPr'cd|j|jjjSr)rrrrr#s r%rz-ContinuousBatchingAsyncIOs.get_actual_lengthss&}}T../77JJLLr'rHc|j|j}|jj||jjj |j yr)rrrrrhr~infer_carry_over_ids)r$rHio_pairs r%rz0ContinuousBatchingAsyncIOs.prepare_batch_tensors#sJ-- 1 12--.?@&&,,T-F-F-HIr'c|j|jjj}|jd|jz jj}t |j Dcgc]}d}}t |jD]\}}|j|}||||< tj|tjScc}w)aInfers the ids of the tokens to carry over from batch N to batch N+1. In asynchronous batching mode, we can schedule a request for batch N+1 without knowing the token predicted for that request in batch N. For that reason, we might need to carry over tokens just predicted in batch N before launching the forwar pass of batch N+1. This method computes the ids of the tokens to carry over.r r)rX) rrrrIrDra enumeraterpgetr,rre)r$!next_req_id_to_new_token_position!prev_req_id_to_new_token_positionrQrhrreq_idnew_token_positions r%rz/ContinuousBatchingAsyncIOs.infer_carry_over_ids(s -1MM$:K:K,L,T,T,q,q),0MM!d>O>O:O,P,X,X,u,u)&+D,A,A&BC"CC ##D#I#I#KL 7IAv!B!F!Fv!N !-5612 7||N%++>>Ds5 Crrc6|j|j}|j|j|jj |j |j j|j |jj||Sr) rrrr record_eventrrO wait_eventrr)r$rrrs r%rz+ContinuousBatchingAsyncIOs.get_model_kwargs<st-- 1 12##DOO4 $$W%5%56 &&w'7'78  11-AUVVr'rcf|jd|jz jj}|j|jjj}||}|dk7j }|d|j d}|d|j d}||z|dd|z zz|d<y)aAs explained in the infer_carry_over_ids method, we might need to carry over tokens just predicted in batch N before launching the forwar pass of batch N+1. This method performs the carry over, and is recorded in CUDA graphs if they are enabled.r rNr)rrrrlrhr1r`)r$rprev_output_idsrhcarried_over_idscarried_over_masks r%rz,ContinuousBatchingAsyncIOs.carry_over_tokensCs --D,=,=(=>HHSSt'8'89CCRR*>:+r1668+,?innQ.?@-.A q0AB'*;;ilaRcNc>dd ! r'c\|j|jjjSr)rrrrlr#s r%rlz%ContinuousBatchingAsyncIOs.output_idsUs%}}T../99DDDr'c\|j|jjjSr)rrrrJr#s r%rJz!ContinuousBatchingAsyncIOs.graphsZs#}}T../99@@@r'cr|j|j}|jj|j|j j |j|j|j |j j|jd|jz |_y)Nr ) rrrOrrrrrrr$rs r%rz2ContinuousBatchingAsyncIOs.retrieve_device_outputs_s-- 1 12 (()=)=> ""7#7#78$$T__5 $$W%5%56 1 11r'c|j|j}|jj|jj Sr)rrrrrrrs r%rz/ContinuousBatchingAsyncIOs.prepare_batch_updateks;-- 1 12$$&3355r'rrr)r(r)r*r+r rr,r8rXr1rRrr-r/r0rr2rr rrrrrpropertyrlrrJrrr4r'r%rrs{+f 7"7!7 7 [[ 7  7 70QellDellARrsg" =#&6]] / /  / dkk\ XX0X6X6r'