iQdZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZer2ddlZddlZddlZdd lmZmZmZmZm Z dd l!m"Z"dd l#m$Z$dd l%m&Z&ddl'm(Z(ejRe*Z+dZ,GddejZZ.GddZ/Gdde0Z1ddddiZ2ddde3dzfdZ4Gdd Z5Gd!d"Z6d#e d$e7de8fd%Z9Gd&d'Z:Gd(d)Z;d*eGd-d.Z?Gd/d0e Z@Gd1d2e@ZAGd3d4e@ZBGd5d6ZCGd7d8ZDy):z? Shared types, constants, and utilities for the serving layer. N)ABCabstractmethod)Callable)Future)BytesIO)Queue) TYPE_CHECKING)logging)ContinuousBatchingConfigGenerationConfigPreTrainedModelPreTrainedTokenizerFastProcessorMixin)ContinuousBatchingManager)GenerationOutput) Scheduler) ModelManagerz x-request-idceZdZdZdZdZdZy)ModalityLLMVLMSTTTTSN)__name__ __module__ __qualname__rrrro/var/www/vps2.regionflexible.com/Desarrollo/venv/lib/python3.12/site-packages/transformers/cli/serving/utils.pyrr=s C C C CrrceZdZdZdefdZy) _StreamErrorz5Sentinel to signal an error from the generate thread.msgc||_yN)r#)selfr#s r __init__z_StreamError.__init__Gs rN)rrr__doc__strr'rrr r"r"Ds?Crr"ceZdZdZy)_GenerationCancelledzERaised inside ``DirectStreamer.put()`` to abort ``model.generate()``.N)rrrr(rrr r+r+KsOrr+qwenz z )startendmodelr returnc|jjdj}tD]}||vst|cSy)a.Return the tool call token format for a model, if supported. Args: model (`PreTrainedModel`): The loaded model. Returns: `dict | None`: A dict ``{"start": str, "end": str}`` with the model's tool call delimiters, or ``None`` if the model family is not recognized. rN)config architectureslower_TOOL_CALL_TOKENS)r/ architecturefamilys r detect_tool_formatr8[sH<<--a0668L#- \ !$V, ,- rceZdZdZdefdZeZdedeezdzfdZ e dede eefdzfd Z e dedede edzfd Zdededzfd Zy) ToolCallParseru]Parses tool calls from model output. The model emits tool calls as structured text between start/end tokens (e.g. ``{"name": "fn", "arguments": {...}}``). **Streaming** (``feed``): buffers tokens between start/end markers, parses the complete block when the end marker is seen, returns a ``ChoiceDeltaToolCall``. **Non-streaming** (``parse``): extracts all tool call blocks from complete text. Usage:: parser = ToolCallParser(tool_format={"start": ..., "end": ...}) for text_chunk in streamer: result = parser.feed(text_chunk) if result is None: # Normal text — emit as content elif result is ToolCallParser.CONSUMED: # Buffering — skip else: # result is a ChoiceDeltaToolCall — emit it tool_formatc.||_d|_d|_y)NF)_tokens_inside_buffer)r&r;s r r'zToolCallParser.__init__s"   rtextr0Nc|j|jdk(rd|_d|_|jS|j|jdk(rGd|_|jj}d|_|j |xs |jS|jr!|xj|z c_|jSy)uFeed a text chunk (streaming). Returns: - ``None`` — normal text, not a tool token. Emit as content. - ``CONSUMED`` — token consumed internally (buffering/markers). Skip. - A ``ChoiceDeltaToolCall`` — emit as a tool call delta. r-Tr=r.FN)stripr>r?r@CONSUMED _parse_block)r&rAblocks r feedzToolCallParser.feeds ::<4<<0 0DLDL== ::<4<<. . DLLL&&(EDL$$U+q q ::r)rrrr(dictr'objectrDr)rG staticmethodtuplerPlistr[rErrr r:r:ls.D xH$!52 +c +eCHo.D + +2C2d2tDzD/@22*;#;$+;rr:cpeZdZdZdedefdZdededzddfd Zded ededzddfd Z deddfd Z dd Z y)DownloadAggregatora Aggregates byte-progress across multiple concurrent download tqdm bars. huggingface_hub opens one tqdm bar per file shard. This class tracks them all and emits a single aggregate ``{"stage": "download", "progress": {...}}`` event whenever any updates. enqueuemodel_idc<||_||_i|_d|_yr%)rdr/barslast_emitted_current)r&rdres r r'zDownloadAggregator.__init__s  79 04!rbar_idtotalNr0cFd|f|j|<|jy)z6Register a new download bar with its total byte count.rNrg_emit)r&rirjs r registerzDownloadAggregator.registersJ & rcurrentcF||f|j|<|jy)z>Update a bar's current byte count and emit aggregate progress.Nrl)r&rirorjs r updatezDownloadAggregator.updates$e, & rcyr%r)r&ris r closezDownloadAggregator.close rcTtd|jjD}||jk(ry||_|jjDcgc] \}}| | }}}|r t|nd}|j d|j d||ddycc}}w)Nc3&K|] \}}| ywr%r).0c_s r z+DownloadAggregator._emit..s;1!;sloadingdownloadrorjstatusr/stageprogress)sumrgvaluesrhrdr/)r& agg_currentryttotals agg_totals r rmzDownloadAggregator._emits; (8(8(:;; $33 3 $/! $ 0 0 2D1am!DD#)CKt  ##(3iH   Es  B$*B$r0N) rrrr(rr)r'intrnrqrsrmrrr rcrcsu 55C5 s3:$ S3sTzd  C D  rrccallbackrecNddlm}tGfdd|}|S)u Create a tqdm subclass that routes progress to a callback. Bars with ``unit="B"`` are download bars — aggregated via ``DownloadAggregator``. Other bars (e.g. "Loading weights") emit ``weights`` stage events. Args: callback (`callable`): Called with a dict payload ``{"status": "loading", "model": ..., "stage": ..., "progress": ...}``. model_id (`str`): The model ID (included in progress payloads). Returns: A tqdm subclass that forwards progress to *callback*. r)tqdmcLeZdZfdZdfd ZfdZfdZxZS).make_progress_tqdm_class..ProgressTqdmc |jdxsd|_d|d<t| |i|d|_d|_|jdk(r7t ||_j|j|jyy)NunititTdisablerB) rMsse_unitsuperr'n last_emittedid_bar_idrnrj)r&argskwargs __class__download_aggregators r r'z7make_progress_tqdm_class..ProgressTqdm.__init__ sw"JJv.6$DM $F9  G d -f -DF "D }}#!$x #,,T\\4::F$rcX|d}|xj|z c_|jdk(r2j|j|j|jy|j|j k7r6|j|_dd|j|jddyyNrrr{weightsr}r~)rrrqrrjr)r&rrrres r rqz5make_progress_tqdm_class..ProgressTqdm.updatesy FFaKF}}##**4<<L4,,,$(FF!"+!)!*04$L -rc 3K|jD]}|xjdz c_|jdk(r2j|j|j|j nN|j|j k7r5|j|_dd|j|j dd|ywr)iterablerrrqrrjr)r&itemrrres r __iter__z7make_progress_tqdm_class..ProgressTqdm.__iter__(s  ! ==C''..t||TVVTZZPVVt000(,D%&/%-%.48FFTZZ(P   sB;B>cv|jdk(rj|jt|y)Nr)rrsrr)r&rrs r rsz4make_progress_tqdm_class..ProgressTqdm.close9s*}}##))$,,7 GMOr)r)rrrr'rqrrs __classcell__)rrrres@r ProgressTqdmr s G " "  rr) tqdm.autorrc)rre base_tqdmrrs`` @r make_progress_tqdm_classrs/,,Xx@00y0d rc leZdZdZ d dddej dej defdZd d Z dd Z dd Z y)DirectStreameraStreamer for ``model.generate()`` (used by :class:`GenerateManager`). Implements the ``put``/``end`` protocol that ``model.generate()`` expects: generate calls ``put(token_tensor)`` after each decode step, and ``end()`` when generation is complete. Tokens are decoded incrementally via the Rust ``DecodeStream`` (O(1) per token) and pushed as text to an asyncio.Queue. tokenizertokenizers.Tokenizerloopqueueskip_special_tokenscddlm}||_||_||_|g||_d|_tj|_ d|_ y)a Args: tokenizer: The Rust tokenizer (``tokenizer._tokenizer``). loop (`asyncio.AbstractEventLoop`): The event loop to push decoded text to. queue (`asyncio.Queue`): The queue that receives decoded text chunks. skip_special_tokens (`bool`, *optional*, defaults to `True`): Whether to strip special tokens during decoding. r DecodeStreamTN) tokenizers.decodersr _tokenizer_loop_queue_decode_stream_first threadingEvent _cancelled total_tokens)r&rrrrrs r r'zDirectStreamer.__init__JsI 5#  *2/BC #//+rNcz|jjr t|jrd|_y|j D]p}|xj dz c_|j j|j|}|A|jj|jj|ry)zHCalled by ``model.generate()`` after each decode step with new token(s).FNr) ris_setr+rtolistrrsteprrcall_soon_threadsafer put_nowait)r&valuetoken_idrAs r putzDirectStreamer.putcs ?? ! ! #&( ( ;;DK   NH    " &&++DOOXFD // 0F0FM  Nrcd|jj|jjdy)z;Called by ``model.generate()`` when generation is complete.N)rrrrr&s r r.zDirectStreamer.endqs  '' (>(>Erc8|jjy)zWSignal cancellation. The next ``put()`` call will raise and abort ``model.generate()``.N)rsetrs r cancelzDirectStreamer.cancelus r)T)r torch.Tensorr0Nr) rrrr(asyncioAbstractEventLooprboolr'rr.rrrr rrAsR%) )''}}  " 2 NFrrc leZdZdZdddedddej dejf d Zdd Z dd Z dd Z y ) CBStreameraStreamer for continuous batching (used by :class:`CBGenerateManager`). Same ``put``/``end`` protocol as :class:`DirectStreamer`, but called manually by :class:`CBGenerateManager` instead of by ``model.generate()``: ``put(output)`` receives a CB ``GenerationOutput``, decodes new tokens, and pushes text to the asyncio.Queue. ``end()`` signals the stream is complete. cb_managerr request_idrrrrcddlm}||_||_||_||_||_|gd|_d|_d|_ y)a Args: cb_manager (`ContinuousBatchingManager`): The CB manager instance. request_id (`str`): The request ID to track in the CB scheduler. tokenizer: The Rust tokenizer (``tokenizer._tokenizer``). loop (`asyncio.AbstractEventLoop`): The event loop to push decoded text to. queue (`asyncio.Queue`): The queue that receives decoded text chunks. rrTN) rr_cb _request_idrrrr _prev_lenr)r&rrrrrrs r r'zCBStreamer.__init__sI 5%  #*2t4rNc*|j|jd}t|j|_|D][}|xjdz c_|jj |j |}|A|jj|]y)zLDecode new tokens from a CB ``GenerationOutput`` and push text to the queue.Nr) generated_tokensrrTrrrrrr)r&output new_tokensrrAs r rzCBStreamer.puts,,T^^-=> V445" -H    " &&++DOOXFD &&t,  -rc:|jjdy)zSignal end of stream.N)rrrs r r.zCBStreamer.ends t$rcN|jj|jy)zCancel the CB request.N)rcancel_requestrrs r rzCBStreamer.cancels  0 01r)rrr0Nr) rrrr(r)rrrr'rr.rrrr rrzsU/*  ''  }} 6-%2rrseedc0ddl}|j|y)z8Set the PyTorch random seed for reproducible generation.rN)torch manual_seed)rrs r set_torch_seedrsEdrcvddl}|jjr|jjyy)z+Empty the CUDA cache if a GPU is available.rN)rcuda is_available empty_cache)rs r reset_torch_cachers* zz   !rcJeZdZdZdZddZdefdZdej fdZ y) InferenceThreadzPersistent thread for ``model.generate()`` calls. ``torch.compile`` with CUDA graphs stores state in thread-local storage. All inference must run on the same thread to avoid corrupted graph state. ct|_tj|jd|_|j j y)NT)targetdaemon)rrrThread_run_threadr-rs r r'zInferenceThread.__init__s3"W  ''tyyF  rr0NcD |jj\}}}}} ||i|}||j|j|n|j|Z#t$r:}||j|j |n|j |Yd}~?d}~wwxYwr%)rrMr set_result Exception set_exception)r&fnrrfuturerrZrYs r rzInferenceThread._runs-1[[__-> *Bffd ,T,V,#--f.?.?H%%f- ,#--f.B.BAF((+  ,s8A B%0BBcZt}|jj||||df|S)ESubmit a callable to the inference thread. Returns a blocking Future.N)rrr)r&rrrrs r submitzInferenceThread.submits) T66489 rctj}|j}|jj |||||f|SzOSubmit a callable to the inference thread. Returns an awaitable asyncio.Future.)rget_running_loop create_futurerr)r&rrrrrs r async_submitzInferenceThread.async_submits>'')##% T66489 rr) rrrr(r'rrrrrrrr rrs-  ,V 7>>rrceZdZdZedddddeddd ed eejd ff d Z edddddeddd ed eee e e ff d Z eddZy)BaseGenerateManageruBase class for generation managers. Subclasses: - :class:`GenerateManager` — sequential ``model.generate()`` on a persistent thread. - :class:`CBGenerateManager` — continuous batching with paged attention. r/r processor(ProcessorMixin | PreTrainedTokenizerFastinputs gen_configr rr0zDirectStreamer | CBStreamercy)axStart streaming generation. Args: model (`PreTrainedModel`): The loaded model. processor: The processor or tokenizer for decoding. inputs (`dict`): Tokenized inputs (tensors for sequential, lists for CB). gen_config (`GenerationConfig`): Generation parameters. request_id (`str`): Unique request identifier. Returns: `tuple[asyncio.Queue, DirectStreamer | CBStreamer]`: A ``(queue, streamer)`` pair where *queue* yields ``str | _StreamError | None`` and *streamer* exposes ``.total_tokens`` and ``.cancel()``. Nrr&r/rrrrs r generate_streamingz&BaseGenerateManager.generate_streamingrcy)aRun generation to completion. Args: model (`PreTrainedModel`): The loaded model. processor: The processor or tokenizer for decoding. inputs (`dict`): Tokenized inputs (tensors for sequential, lists for CB). gen_config (`GenerationConfig`): Generation parameters. request_id (`str`): Unique request identifier. Returns: `tuple[str, int, list[int]]`: ``(text, input_len, generated_ids)``. Nrrs r generate_non_streamingz*BaseGenerateManager.generate_non_streaming r rNcy)z/Stop the generation manager and free resources.Nrrs r stopzBaseGenerateManager.stopr rr)rrrr(rr]r)r`rrr rrar rrrr rrs   >   '     w}};; <  .   >   '     sCc" #  *>>rrceZdZdZdZdddddedd d ed eeje ff d Z dddddedd d ed eee d ff dZ ded efdZded ejfdZddZy)GenerateManagerzFSequential generation via ``model.generate()`` on a persistent thread.c"t|_yr%)rrrs r r'zGenerateManager.__init__'s &( rr/r rrrrr rr0c tj tj t|j d}i||||dd  fd }|j | |fS)zLStart streaming generation via ``model.generate()`` on the inference thread.Tr)streamergeneration_configrc  jdiy#t$rjjdYyt$r8}jjt t |Yd}~yd}~wwxYw)Nr)generater+rrrr"r))rY gen_kwargsrr/rs r rz0GenerateManager.generate_streaming.._run8sm R,,' B))%*:*:DA R))%*:*:LQ  '    w}}n, -2. .>. . ' .  . sC' (."88v8>x>W^^> rrceZdZdZdddZdd Zddd d d eddd edee je ff dZ ddd d d eddd edeee ee ff dZeddZddZy)CBGenerateManagerauContinuous batching generation via paged attention. Translates between the handler's text-level asyncio.Queue and CB's token-level interface. Per-request: ``max_new_tokens``, ``eos_token_id``. The CB manager is initialized lazily on the first request via :meth:`ensure_initialized`, using that request's ``gen_config`` for shared sampling params (temperature, top_p, do_sample). .. todo:: Remove :meth:`init_cb` when CB supports per-request generation config. At that point, ``gen_config`` can be passed directly to ``add_request`` and the CB manager no longer needs a shared config. Nc d|_||_yr%)r _cb_config)r& cb_configs r r'zCBGenerateManager.__init__os#rr/r rr r0c|jy|j||j|_|jjy)atInitialize the CB manager on first call with the request's generation config. .. todo:: Remove when CB supports per-request generation config. Args: model (`PreTrainedModel`): The loaded model (must support ``init_continuous_batching``). gen_config (`GenerationConfig`): Generation config used for shared sampling params. N)rcontinuous_batching_config)rinit_continuous_batchingr'r-)r&r/rs r init_cbzCBGenerateManager.init_cbssA 88  11(T__2  rrrrrcZ tj}tj |d}|jj ||d|j |j }t|j||j|  fd}|jj|| fS)zFStart streaming CB generation. Registers a per-request output handler.rT)r streamingmax_new_tokens eos_token_idc j||jrjyy#t$r-}j t t |Yd}~yd}~wwxYwr%)r is_finishedr.rrr"r))rrYr text_queues r _on_outputz8CBGenerateManager.generate_streaming.._on_outputsX < V$%%'LLN( <%%l3q6&:;; ._on_results;;=!!&)!rF)rr/r.r0Nz1CB manager stopped before producing a result for Tr) rTrrrrr6r5r/r0 RuntimeErrorrr)r&r/rrrrrrrr:rZrrArs @r r z(CBGenerateManager.generate_non_streamings;'  N '')##% * ((Z@  !%44#00   >!RS]R^_` `//  4HY -- sBC C 9C cB|jjjS)z*The CB scheduler (for testing/monitoring).)rbatch_processor schedulerrs r r>zCBGenerateManager.schedulersxx''111rcX|j|jjddyy)NT)rFtimeout)rrrs r rzCBGenerateManager.stops% 88  HHMMaM 0 rr%)r(ContinuousBatchingConfig | None)r/r rr r0N)r0rr)rrrr(r'r,r]r)r`rrrr rrar propertyr>rrrr r%r%`s $"!$ !$>!$ !$ ' !$  !$ w}}j( )!$F". ".>". ". ' ".  ". sCc" #".H221rr%c^eZdZdZ ddededdfdZdd d ed efd Zdd eded e fdZ ddZ y)GenerationStatea'Shared generation state across all handlers. Manages per-model :class:`GenerateManager` instances (each with its own :class:`InferenceThread` so different models can run concurrently while ``torch.compile`` / CUDA graphs require same-model-same-thread) and a single :class:`CBGenerateManager` for continuous batching. Args: continuous_batching (`bool`, *optional*, defaults to `False`): Whether to use continuous batching with paged attention instead of sequential ``model.generate()`` calls. Ncontinuous_batchingcompiler(rBcX||_||_||_i|_d|_d|_yr%)_continuous_batching_compiler'_generate_managers _cb_manager _cb_model_id)r&rFrGr(s r r'zGenerationState.__init__s2 %8! #>@59(,rr/r modalityr0c|jsyt|dxr|tjk(}|s,tj |j jd|S)aWCheck if continuous batching can be used for this model and modality. Args: model (`PreTrainedModel`): The loaded model. modality (`Modality`): The detected model modality (LLM, VLM, etc.). Returns: `bool`: ``True`` if CB is enabled and the model supports it, ``False`` otherwise. Fr+zM does not support continuous batching. Falling back to sequential generation.)rIhasattrrrlogger warning_oncerr)r&r/rNcans r use_continuous_batchingz'GenerationState.use_continuous_batchings]((e78UX=U   ??++,-99  rreuse_cbcZ|rv|j|k7r-|j!|jjd|_|j"t|j|_||_|jS||j vrt |j |<|j |S)afReturn a per-model generation manager, lazily created on first request. Args: model_id (`str`): The model ID in ``'model_id@revision'`` format. use_cb (`bool`): Whether to return a CB manager or a sequential one. Returns: `BaseGenerateManager`: Either a `GenerateManager` or `CBGenerateManager`. N)r()rMrLrr%r'rKr)r&rerUs r get_managerzGenerationState.get_managers   H,##/$$))+'+D$'#4t#O $,!## # 422 20?0AD # #H -&&x00rc`|j"|jjd|_yy)z$Stop any active generation managers.N)rLrrs r shutdownzGenerationState.shutdowns-    '    ! ! ##D  (r)FFNFr) rrrr(rr'rrTr)rrWrYrrr rErEso %*7; -! - -5 -->(W[(1C11BU1.$rrEc eZdZUdZdZedzed<eZee ed<ddde fdZ d e d dfd Z ed d d e fdZd e d ee ddffdZ dd e ddded dfdZedee ded ee fdZy) BaseHandleraShared logic for chat completion and responses handlers. Provides model resolution, generation config building, and SSE formatting. Generation is delegated to the shared :class:`GenerationState`. Args: model_manager (`ModelManager`): Handles model loading, caching, and lifecycle. generation_state (`GenerationState`): Shared state managing per-model generation managers. N_valid_params_class_unused_fields model_managerrgeneration_statec ||_||_yr%)r_r`)r&r_r`s r r'zBaseHandler.__init__2s + 0rbodyr0cddlm}t|j}|j(||jj z }|r |dd|||j z}|rtjd|yy)zMValidate request fields against the handler's params class and unused fields.r) HTTPExceptionNiz"Unexpected fields in the request: ) status_codedetailz,Ignoring unsupported fields in the request: ) fastapirdrkeysr]__mutable_keys__r^rQrR)r&rbrd input_keys unexpectedunuseds r _validate_requestzBaseHandler._validate_request:s)%  # # /#d&>&>&O&OOJ#>`ak`lW3X Y  $ .E F  //7;L;[;[^b;b37!0 88M " .,1$}2E,F  )T-()S0.3!+ 88G  (&+DM&:  # 88F  ' 4< (  ) ).?.T.T.\5=  2 */  '! rmessagesrNc g}|D]}|dgd}|tjk(ret|dtr |d|d<net|dtrQ|dDcgc]}|ddk(s |d}}dj ||d<n|tj k(r t|dtr|djd|ddn|dD]}|ddk(r|dj| |ddk(s)d d lm }|dd } d | vrtjd d| } |jttj| } t!j"dd} | j%| j&| j&} |djd| d|j||Scc}w)aConvert OpenAI-format messages to the format expected by HF processors. For LLMs, collapses list content blocks into plain text. For VLMs, converts ``image_url`` content parts (including base64) into ``{"type": "image", "url": ...}`` entries that HF processors understand. Args: messages (`list[dict]`): OpenAI-format chat messages. modality (`Modality`): Whether the model is an LLM or VLM. Returns: `list[dict]`: Processor-compatible messages. role)rcontentrtyperA )rrA image_urlr)Imageurlbase64z^data:image/.+;base64,r=z.pngF)suffixdeleteimage)rr)rrrqr)rajoinrrUPILrresubopenrr b64decodetempfileNamedTemporaryFilesaverI) rrNprocessor_inputsmessagerOrxtextsrrr image_datarfiles r "get_processor_inputs_from_messagesz.BaseHandler.get_processor_inputs_from_messagess ,G%fo"=F8<<'gi0#6(/ (:F9% 2D907 0BZ1aiSYFYQvYZEZ(+F9%X\\)gi0#69%,,fgiFX-YZ#*9#5 T"6?f4"9-44W=$V_ ;1")+"6u"=C'3-/VV4LbRU-V (- 76;K;KJ;W3X(Y'/'B'B&Y^'_ % 499 5&*ii"9-44gc5RS T  # #F +; ,< /[s  F<*F<rZ)rrrr(r]r__annotations__rr^r)rEr'r]rmr_rtr`ryrrrarrrrr r\r\"s (,+"uNCH$1%1*1 Yd Yt YG6G3GG *4 *E#7HJt2t,u *W\0!0!3E0!OS0! 0!d. T$Z. 8. X\]aXb. . rr\r)Er(rrrenumrKrrrabcrrcollections.abcrconcurrent.futuresriorrrtypingr transformers.utilsr pydantic tokenizersrrr r r rr:transformers.generation.continuous_batching.continuous_apir4transformers.generation.continuous_batching.requestsr5transformers.generation.continuous_batching.schedulerrr_r get_loggerrrQ X_REQUEST_IDEnumrr"rr+r5r]r8r:rcr)rrrrrrrrrrr%rEr\rrr rs #$% &eUO+   H % tyyP9P  /D4K"`;`;F( ( VDxD3D4DN66r4242n!&&R8>#8>v9 )9 xr1+r1jJ$J$ZZ Z r