tjx dZddlZddlZddlZddlZddlZddlmZmZddl m Z ddl m Z ddl mZddlmZddlmZer2ddlZddlZddlZdd lmZmZmZmZmZdd lmZdd lmZdd l m!Z!d dl"m#Z#ej$e%Z&dZ'Gddej(Z)GddZ*Gdde+Z,Gdde-Z.Gdde/Z0ddiddddddd id!d"ddd#d$d%d&d'id%d(d)d*d+d,d"d-Z1d.d/d0e2dzfd1Z3d2e2d0e2fd3Z4d4e2d0e5e2dzfd5Z6d6gd7d%d&d'id&d'id8d9d:d;Z7dd?iZ8d`d.d/d0e2dzfd@Z9dAe-dBe2d0e:e-e-dzffdCZ;dDe5e<d0e=fdEZ>dFez z<|im_start|>assistant tool_callsTjson)opencloserepeatscontent)defaults start_anchorfields)stcetcschemaz9[^>\n]+)>(?P.*?)arrayobjecttypestringz<[^>\n]+)>\s*(?P.*?)\s*)r@zx-regex-key-valuename arguments)r@ properties)zx-regex-iteratorr@items)) qwen2 qwen2_moeqwen2_vl qwen2_5_vlqwen3 qwen3_moe qwen3_nextqwen3_vl qwen3_vl_moe)qwen3_5 qwen3_5_moemodelr returnc t|d|}t|dd}t|dd}t|dd}t|dd}d}|rF|rD|rBd|divr*id|ddid }d D]}||vr ||||<nnp|r|r|r|d d}n[|jj t fd t Dd} | dS| d | d| d}}}||} ||} || | dS)afReturn tool call config for the model, or ``None`` if tool calls are not supported. Returns a dict with: - ``schema`` (`dict`): Schema to pass to ``tokenizer.parse_response(block, schema)``. - ``stc_id`` (`int`): Token ID of the start-of-tool-call delimiter. - ``etc_id`` (`int`): Token ID of the end-of-tool-call delimiter. tokenizer stc_tokenN etc_tokenresponse_templateresponse_schemar2r:)r8r:)r9start_anchor_patternrEc3*K|] \}}|v |VdSr%r).0typesv model_types r z'get_tool_call_config..s2__xua:Y^K^K^K^K^K^K^__rr;r<r=)r=stc_idetc_id)getattrgetconfigr_next_TOOL_CALL_FALLBACKSrFconvert_tokens_to_ids) processorrRrUr;r<rXrYr= anchor_keyfallbackrarbr_s @r get_tool_call_configrls ; ::I )[$ / /C )[$ / /C +>EEi):DAAOF PsP(P\=N=R=RS[]_=`=`-`-`#%6x%@%NO   C  J...%6z%Bz"/ P P P .|<\, ____+?+E+E+G+G___aeff  4#E?HUOXh=O&S  , ,S 1 1F  , ,S 1 1F& A AAr tool_callc|d|}|di}|dt|tstj|n|dS)aNormalize a parsed tool call to ``{"name": str, "arguments": str}``. Different models return different structures from ``parse_response``: - Gemma: ``{"function": {"name": ..., "arguments": {...}}}`` (nested, arguments as dict) - Qwen: ``{"name": ..., "arguments": {...}}`` (flat, arguments as dict) The OpenAI API expects ``arguments`` as a JSON **string**, so we ``json.dumps`` it. functionrDrCrB)rd isinstancer)r3dumps)rmrorDs r _normalize_tool_callrrsc}}Z33H ["--I 2Parse tool calls from generated token IDs using ``tokenizer.parse_response``. Args: processor: The processor or tokenizer. generated_ids: Token IDs from generation. Passed directly to ``parse_response`` which decodes them internally, preserving special tokens that ``skip_special_tokens=True`` would strip (e.g. Gemma's ``<|tool_call>``). schema: The tool call schema (from ``response_schema`` or ``_TOOL_CALL_FALLBACKS``). Returns a list of ``{"name": str, "arguments": str}`` dicts, or ``None`` if none found. )prefixr2Nc,g|]}t|Sr)rr)r\rms r z$parse_tool_calls..s!JJJi&y11JJJr)parse_responserpdictlist)ri generated_idsr=parsedr2s r parse_tool_callsr}s % %mVB % G GF&$&LF$:$: % t fd # #JJ6JJJJ# -::-rzz)thinkingr7zK(?:)?(?P.*?)(?P.*?)(?:<\|[^|<>\s]+\|>)?\Z)r@rEzx-regex)startendr=gemma4)z <|channel>thought z )rrc t|d| |jjt fdt Dt} fd|dD} |d}t fd|Ds |d j fvrdSt dd}|r d |d vs td }|||d }|t|||d <|S)aReturn reasoning config for the model, or ``None`` if not supported. The config drives both streaming detection (token IDs) and post-hoc parsing (response schema). Returns a dict with: - ``start_ids`` (`list[int]`): Token ID sequence that opens a thinking block. - ``end_id`` (`int`): Token ID that closes the block. - ``schema`` (`dict`): Response schema with ``thinking`` / ``content`` properties for :func:`parse_reasoning`. - ``start_in_thinking`` (`bool`, only when ``input_ids`` is given): Whether the rendered prompt already opened an unclosed thinking block (prefilled by the template), so the model's output begins inside the block. rUc3.K|]\}}|k |VdSr%r)r\kr^r_s r r`z'get_reasoning_config..s+CCtq!1 ??????CCrc:g|]}|Sr)rh)r\trUs r rwz(get_reasoning_config..s'VVV0033VVVrrrc3.K|]}|djfvVdSr%) unk_token_id)r\tidrUs r r`z'get_reasoning_config..s0 F FS34/0 0 F F F F F FrNrYr~rEr=) start_idsend_idr=start_in_thinking) rcrer_lowerrf_THINKING_TOKENSrF_DEFAULT_THINKING_TOKENSrhanyr_starts_in_thinking) rirR input_idsthinking_tokensrrr=rer_rUs @@r get_reasoning_configrs> ; ::I(..00JCCCC'--//CCC OWVVV_W=UVVVI  , ,_U-C D DF F F F FI F F FFF&UY[d[qTrJrJrtY 14 8 8F 4zVL%999)(3!*fOOF&9)Y&O&O"# Mrr7reasoning_configc|||d}|r0|dd}|r|dd|fS|drd|fS|dfS)uSplit generated output into ``(content, reasoning_content)`` via ``parse_response``. If the schema's regex matches (closing marker present), use it. For prompts that prefill the opener (QwQ-32B, DeepSeek-R1) the entire output is reasoning until ```` arrives — when that's truncated, fall back to treating all decoded text as reasoning. Returns ``(content, None)`` otherwise. r=r~rtr7rN)rxrd)rir{r7rr| reasonings r parse_reasoningr%s % %m5Eh5O P PF 8JJz2..  8::i,,i7 7/007{ D=rrcbt|dr|}|r8t|dtrt |dkrdS|d}t |}dD]>}t |||zkr&t ||z }|||z ||krdS?dS)uyTrue if the rendered prompt ends with an unclosed thinking block. Some reasoning-model chat templates prefill the thinking opener as the final prompt tokens (e.g. DeepSeek-R1, QwQ-32B emit ``\n`` at the end when ``add_generation_prompt=True``). In those cases the model resumes *inside* the block, so its output contains only ``...reasoninganswer`` with no opening tag — the streamer must start with ``_inside_thinking=True``. The prefill always lands at the tail of the prompt (optionally followed by a single whitespace token like ``\n``), so we only inspect the last few tokens. tolistrrF)rrT)hasattrrrprzlen)rrntrailingrs r rr9sy(##'$$&& !Z ! d33! y>>Q  5aL  IA y>>Q\ ) )i..8+Cq3'944tt 5rtoken_idcR|jdS|jr||jkr d|_dSdS|jt|j}||kr g|_dS|j|t|jt|jkrd|_g|_dS)uUMutate ``streamer``'s thinking state; return ``True`` if ``token_id`` is a start or end token. Shared between :class:`DirectStreamer` and :class:`CBStreamer` — both track the same four attributes (``_thinking_start_ids``, ``_thinking_end_id``, ``_inside_thinking``, ``_thinking_prefix``) and need identical edge handling. NFT)_thinking_start_ids_inside_thinking_thinking_end_idr_thinking_prefixappend)streamerrexpecteds r _advance_thinking_staterUs#+u  x0 0 0(-H %4u+C0I,J,JKH8$&!u $$X... 8 $%%X-I)J)JJJ$(!$&! 4rcreZdZdZdedefdZdededzddfd Zded ededzddfd Z deddfd Z dd Z dS)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|_dSr%)rrRbarslast_emitted_current)r&rrs r r'zDownloadAggregator.__init__us%  79 04!!!rbar_idtotalNrScFd|f|j|<|dS)z6Register a new download bar with its total byte count.rNr_emit)r&rrs r registerzDownloadAggregator.register{s#J & rcurrentcF||f|j|<|dS)z>Update a bar's current byte count and emit aggregate progress.Nr)r&rrrs r updatezDownloadAggregator.updates$$e, & rcdSr%r)r&rs r r5zDownloadAggregator.close rc>td|jD}||jkrdS||_d|jD}|rt|nd}|d|jd||dddS)Nc3 K|] \}}|V dSr%r)r\c_s r r`z+DownloadAggregator._emit..s&;;1!;;;;;;rcg|] \}}|| Sr%r)r\rrs r rwz,DownloadAggregator._emit..sDDD1am!mmmrloadingdownloadrrstatusrRstageprogress)sumrvaluesrrrR)r& agg_currenttotals agg_totals r rzDownloadAggregator._emits;; (8(8(:(:;;;;; $3 3 3 F$/!DD 0 0 2 2DDD#)3CKKKt  ##(3iHH        rrSN) rrrr(rr)r'intrrr5rrrr rrns 55C5555 s3:$ S3sTzd  C D          rrcallbackrc\ddlm}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|dpd|_d|d<tj|i|d|_d|_|jdkr6t ||_|j|j dSdS)NunititTdisablerB) rdsse_unitsuperr'r last_emittedid_bar_idrr)r&argskwargs __class__download_aggregators r r'z7make_progress_tqdm_class..ProgressTqdm.__init__s"JJv..6$DM $F9  EGG d -f - - -DF "D }##!$xx #,,T\4:FFFFF$#rrc |d}|xj|z c_|jdkr(|j|j|jdS|j|jkr+|j|_dd|j|jdddSdSNrrrweightsrr)rrrrrr)r&rrrrs r rz5make_progress_tqdm_class..ProgressTqdm.updatesy FFaKFF}###**4<LLLLL4,,,$(F!"+!)!*04$L$L -,rc 3K|jD]}|xjdz c_|jdkr'|j|j|jn9|j|jkr)|j|_dd|j|jdd|VdSr)iterablerrrrrr)r&itemrrrs r __iter__z7make_progress_tqdm_class..ProgressTqdm.__iter__s   ! =C'''..t|TVTZPPPPVt000(,D%H&/%-%.48FTZ(P(P    rc|jdkr|jtdS)Nr)rr5rr)r&rrs r r5z4make_progress_tqdm_class..ProgressTqdm.closes;}###))$,777 GGMMOOOOOr)r)rrrr'rrr5 __classcell__)rrrrs@r ProgressTqdmrs G G G G G G        "       "          rr) tqdm.autorr)rr base_tqdmrrs`` @r make_progress_tqdm_classrsp,+++++,Xx@@000000000y000d rcreZdZdZ ddddejdejded edzd edzf d Z ddZ ddZ ddZ dS)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. TNrUtokenizers.Tokenizerloopqueueskip_special_tokens tool_configrcddlm}||_||_||_|g||_|r|dnd|_|r|dnd|_d|_|r|dnd|_ |r|dnd|_ t|o| d |_ g|_d |_t!j|_d|_g|_dS) 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. tool_config (`dict`, *optional*): Tool call config from ``get_tool_call_config``. When set, tokens between stc/etc delimiters (inclusive) are suppressed from the queue so tool call markup is never streamed to the client. reasoning_config (`dict`, *optional*): Thinking config from ``get_reasoning_config``. When set, tokens between start/end delimiters are wrapped as :class:`ReasoningText` so handlers route them to ``reasoning_content``. r DecodeStreamraNrbFrrrT)tokenizers.decodersr _tokenizer_loop_queue_decode_stream_stc_id_etc_id_inside_tool_callrrboolrdrr_first threadingEvent _cancelled total_tokensgenerated_token_ids)r&rUrrrrrrs r r'zDirectStreamer.__init__s. 544444#  *l2/BCC0;E{8,, 0;E{8,, !&DT#^#3K#@#@Z^ >N X 0 : :TX $%5%c:J:N:NOb:c:c d d+- #/++.0   rvalue torch.TensorrScD|jrt|jr d|_dS|D]}|xjdz c_|j|||jkrd|_ n||j krd|_ t||}|j |j|}||j s ||j ks|r|jrt!|}|j|jj|dS)zHCalled by ``model.generate()`` after each decode step with new token(s).FNrT)ris_setr+rrrr rrrrrrsteprrr.rcall_soon_threadsafer put_nowait)r&r ris_start_or_end_tokentexts r putzDirectStreamer.puts: ? ! ! # # )&(( ( ; DK F  J JH    "    $ + +H 5 5 54<'')-&&T\))).&$;D($K$K !&++DOXFFD|t5|T\9Q9QUj9Q$ +$T** J + +DK,BD I I I I! J JrcP|j|jjddS)z;Called by ``model.generate()`` when generation is complete.N)rrrrr&s r rzDirectStreamer.end,s% '' (>EEEEErc8|jdS)zWSignal cancellation. The next ``put()`` call will raise and abort ``model.generate()``.N)rsetrs r cancelzDirectStreamer.cancel0s r)TNN)r r rSNr) rrrr(asyncioAbstractEventLooprrryr'rrrrrr rrs%)#'(,'1'1)'1''1} '1 " '1 D[ '1+'1'1'1'1RJJJJ4FFFFrrcteZdZdZ ddddedddejd ejd edzd edzfd Z ddZ ddZ ddZ dS) 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. N cb_managerr request_idrUrrrrrcddlm}||_||_||_||_||_|gd|_|r|dnd|_|r|dnd|_ d|_ |r|dnd|_ |r|d nd|_ t|o|d |_g|_d|_d|_g|_dS) aY 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. tool_config (`dict`, *optional*): Tool call config (see ``DirectStreamer``). reasoning_config (`dict`, *optional*): Thinking config (see ``DirectStreamer``). rrTraNrbFrrr)rr_cb _request_idrrrrrrrrrrrdrr _prev_lenrr ) r&rrrUrrrrrs r r'zCBStreamer.__init__>s( 544444%  #*l2t440;E{8,, 0;E{8,, !&DT#^#3K#@#@Z^ >N X 0 : :TX $%5%c:J:N:NOb:c:c d d+-.0   routputrrSc|j|jd}t|j|_|D]}|xjdz c_|j|||jkrd|_n||jkrd|_t||}|j |j |}||js ||jks|r|j rt|}|j|dS)zLDecode new tokens from a CB ``GenerationOutput`` and push text to the queue.NrTF)generated_tokensr"rrr rrrrrrrrrr.rr)r&r# new_tokensrrrs r rzCBStreamer.putes,T^-=-=> V455" ) )H    "    $ + +H 5 5 54<'')-&&T\))).&$;D($K$K !&++DOXFFD|t5|T\9Q9QUj9Q$ +$T** K " "4 ( ( ( (! ) )rc:|jddS)zSignal end of stream.N)rrrs r rzCBStreamer.end{s t$$$$$rcD|j|jdS)zCancel the CB request.N)r cancel_requestr!rs r rzCBStreamer.cancels!  011111rNN)r#rrSNr) rrrr(r)rrrryr'rrrrrr rr5s$((,%1%1/%1%1* %1 ' %1 } %1D[%1+%1%1%1%1N)))),%%%%222222rrseedc.ddl}|j|dS)z8Set the PyTorch random seed for reproducible generation.rN)torch manual_seed)r+r-s r set_torch_seedr/s$LLLEdrcvddl}|jr|jdSdS)z+Empty the CUDA cache if a GPU is available.rN)r-cuda is_available empty_cache)r-s r reset_torch_cacher4sELLL z  !      !!rcBeZdZdZdZddZdefdZdejfdZ dS) 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|_|jdS)NT)targetdaemon)rrrThread_run_threadrrs r r'zInferenceThread.__init__s@"WW  'tyFFF  rrSNcR |j\}}}}} ||i|}|||j|n||nJ#t$r=}|||j|n||Yd}~nd}~wwxYwr%)rrdr set_result Exception set_exception)r&fnrrfuturerresultes r r;zInferenceThread._runs ,-1[__->-> *Bffd ,T,V,,#--f.?HHHH%%f--- , , ,#--f.BAFFFF((+++  , ,s;A B%(3B  B%c`t}|j||||df|S)ESubmit a callable to the inference thread. Returns a blocking Future.N)rrr)r&rArrrBs r submitzInferenceThread.submits/ T6648999 rctj}|}|j|||||f|SzOSubmit a callable to the inference thread. Returns an awaitable asyncio.Future.)rget_running_loop create_futurerr)r&rArrrrBs r async_submitzInferenceThread.async_submitsE'))##%% T6648999 rr) rrrr(r'r;rrGrrLrrr r6r6sy  , , , ,V 7>rr6ceZdZdZddZe dddd d d eddd ed edzdedzdee j dffdZ eddd d d eddd edeee e e ff dZeddZdS)BaseGenerateManageruBase class for generation managers. Subclasses: - :class:`GenerateManager` — sequential ``model.generate()`` on a persistent thread. - :class:`CBGenerateManager` — continuous batching with paged attention. rRr gen_configr rSNcdS)z:Initialize continuous batching. No-op for non-CB managers.Nrr&rRrOs r init_cbzBaseGenerateManager.init_cbrri(ProcessorMixin | PreTrainedTokenizerFastinputsrrrzDirectStreamer | CBStreamercdS)aStart 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. tool_config (`dict`, *optional*): Tool call config from ``get_tool_call_config``. When set, tool call tokens (between stc/etc) are suppressed from output. reasoning_config (`dict`, *optional*): Thinking config from ``get_reasoning_config``. When set, thinking tokens are wrapped as :class:`ReasoningText`. Returns: `tuple[asyncio.Queue, DirectStreamer | CBStreamer]`: A ``(queue, streamer)`` pair where *queue* yields ``str | _StreamError | None`` and *streamer* exposes ``.total_tokens`` and ``.cancel()``. Nr)r&rRrirUrOrrrs r generate_streamingz&BaseGenerateManager.generate_streamingrSrc KdS)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)``. Nr)r&rRrirUrOrs r generate_non_streamingz*BaseGenerateManager.generate_non_streamings rcdS)z/Stop the generation manager and free resources.Nrrs r stopzBaseGenerateManager.stoprSrrRr rOr rSNr*r)rrrr(rRrryr)tuplerrrWrrzrYr[rrr rNrNsPIIII$((,    >   '    D[ +  w};; <   ^ :   >   '     sCc" #   ^ *>>>^>>>rrNceZdZdZdZ dddddded d d ed edzd edzdeej e ffdZ ddddded d d edeee dff dZ dedefdZdedejfdZddZdS)GenerateManagerzFSequential generation via ``model.generate()`` on a persistent thread.c,t|_dSr%)r6r<rs r r'zGenerateManager.__init__s&(( rNrRr rirTrUrOr rrrrSc, tj tj t|d|j}t | ||} i|| ||d t drd d<d  fd } ||  | fS) zLStart streaming generation via ``model.generate()`` on the inference thread.rUrr)rgeneration_configrU has_talkerrgeneration_moderSNc   jdidS#t$rjdYdSt$r@}jt t |Yd}~dSd}~wwxYw)Nr)generater+rrr?r"r))rD gen_kwargsrrRrs r r;z0GenerateManager.generate_streaming.._runs R,,,,,,,' B B B))%*:DAAAAAA R R R))%*:LQ>Br)rrJrrcrrrrG)r&rRrirUrOrrrrust_tokenizerrr;rhrrs ` @@@r rWz"GenerateManager.generate_streamings'))&} KCCN! D%[Sc   onH:dmnnn 5, ' ' 3,2J( ) R R R R R R R R R Dhrr cKi|||d}t|drd|d<|j|jfi|d{V}|djd}|d|df} || d } | || fS) zNRun generation to completion via ``model.generate()`` on the inference thread.)rcrUrdrreNrrrTr)rrLrgshapedecode) r&rRrirUrOrgenerate_kwargs sequences input_lenr{rs r rYz&GenerateManager.generate_non_streaming's^V]*S\]]] 5, ' ' 817O- .+$+ENNNoNNNNNNNN ;'-b1 !!YZZ-0  4HHY --rrAc.|jj|g|Ri|S)rF)r<rGr&rArrs r rGzGenerateManager.submit;s'"t|"27777777rc.|jj|g|Ri|SrI)r<rLrrs r rLzGenerateManager.async_submit?s'(t|(=d===f===rcdSr%rrs r r[zGenerateManager.stopCrrr*r)rrrr(r'ryr)r]rrrrWrrYrrrGrLr[rrr r_r_saPP)))$((, >  '   D[+ w}n, -B. .>. . ' .  . sC' (....(88v8888>x>W^>>>>      rr_ceZdZdZdddZdd Zd efd Zd ed dfdZ dddddde dd d ede dzde dzd e e j effdZddddde dd d ed e eeeeff dZeddZddZdS)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. N cb_configContinuousBatchingConfig | Nonec"d|_||_dSr%)r  _cb_config)r&rws r r'zCBGenerateManager.__init__Vs59#rrRr rOr rSc|jdS|||j|_|jdS)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)rccontinuous_batching_config)r init_continuous_batchingrzrrQs r rRzCBGenerateManager.init_cbZsN 8  F11(T_2   rc0|jdup |jjduS)zJWhether the CB worker is healthy. ``True`` before ``init_cb()`` is called.N)r  fatal_errorrs r is_alivezCBGenerateManager.is_aliveksx4?48#74#??rrcn|j+|jj!td|d|jjdSdS)uRaise :class:`CBWorkerDeadError` if the CB worker has died. Called at request entry to fail fast — submitting to a dead worker would otherwise enqueue the request into a void where it never gets processed. Nz,CB worker is dead and cannot accept request : )r rr1)r&rs r _check_alivezCBGenerateManager._check_aliveosN 8 DH$8$D#czccTXMacc  $D$DrrirTrUrrc  |j}|td||tj} tj|d} || |d|j|j}t|d|j } t|j|| | || fd} | ||  fS) zFStart streaming CB generation. Registers a per-request output handler.N3CB manager not initialized. Call `init_cb()` first.rT)r streamingmax_new_tokens eos_token_idrUrbc ||j=t|jdS|rdSdS#t $r:}tt|Yd}~dSd}~wwxYwr%)rerrorrr"r is_finishedr?r))r#rDr text_queues r _on_outputz8CBGenerateManager.generate_streaming.._on_outputs < V$$$<+)),v|*D*DEEELLNNNNN''))#LLNNNNN## < < <%%l3q66&:&:;;;;;;;;; )rCrBs r _on_resultz._on_results7;;== *!!&))))) * *rF)rrrrzCB worker died during request rzCB generation failed for Trk)r rrrrrJrKrrrrrrr1r%rm)r&rRrirUrOrrrrprrrCr{rrBs @r rYz(CBGenerateManager.generate_non_streamingseX :TUU U *%%%;'  NN '))##%% * * * * * "":z:::  !%4#0     < #~)'(e(e(eW]Wc(e(efffW:WWWWXX X/  4HHY --rrch|j |jjtd|jjjS)z*The CB scheduler (for testing/monitoring).Nz.Continuous batching processor not initialized.)r batch_processorr schedulerrs r rzCBGenerateManager.schedulers3 8 tx7?OPP Px'11rcP|j|jdddSdS)NT)blocktimeout)r r[rs r r[zCBGenerateManager.stops0 8  HMMaM 0 0 0 0 0 rr%)rwrxr\r*)rSrr)rrrr(r'rRrrr)rryr]rrrrWrrzrYpropertyrr[rrr rvrvGs  $$$$$"@$@@@@ s t    $$((,8$8$ 8$>8$ 8$ ' 8$  8$D[8$+8$ w}j( )8$8$8$8$t/. /.>/. /. ' /.  /. sCc" #/./././.b222X2 111111rrvcleZdZdZ ddededdfdZd d d ed efd Zddeded e fdZ ddZ d efdZ dS)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. FNcontinuous_batchingcompilerwrxcZ||_||_||_i|_d|_d|_dSr%)_continuous_batching_compilerz_generate_managers _cb_manager _cb_model_id)r&rrrws r r'zGenerationState.__init__s8 %8! #>@59(,rrRr modalityrSc|jsdSt|do|tjk}|s't|jjd|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.)rrrrlogger warning_oncerr)r&rRrcans r use_continuous_batchingz'GenerationState.use_continuous_batching sn( 5e788UX=U    ?+999    rruse_cbc |ra|j|kr'|j |jd|_|j!t|j|_||_|jS||jvrt |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)rw)rrr[rvrzrr_)r&rrs r get_managerzGenerationState.get_manager s  $ H,,#/$))+++'+D$'#4t#O#O#O $,!# # 42 2 20?0A0AD #H -&x00rcX|j"|jd|_dSdS)z$Stop any active generation managers.N)rr[rs r shutdownzGenerationState.shutdown7s6   '   ! ! # # ##D    ( 'rcF|jdup|jS)zTWhether the CB worker is healthy. ``True`` if CB is disabled or not yet initialized.N)rrrs r is_cb_alivezGenerationState.is_cb_alive=s$4'F4+;+D+D+F+FFr)FFNFr) rrrr(rr'rrr)rNrrrrrr rrs  %*7; - -! - -5 - - - -->(W[(11C11BU1111.$$$$ GTGGGGGGrrc eZdZUdZdZedzed<eZee ed< dddde de dzfd Z d e d dfd Z ed dd e fdZd e d ee ddffdZ dd e ddded dfdZedee ded ee fdZdS) 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_statechat_template_kwargsc4||_||_|pi|_dSr%)rrr)r&rrrs r r'zBaseHandler.__init__Rs' + 0$8$>B!!!rbodyrSc&ddlm}t|}|j7|t |jdtz }|r|dd|||jz}|rtd|dSdS) zMValidate request fields against the handler's params class and unused fields.r HTTPExceptionN__mutable_keys__iz"Unexpected fields in the request:  status_codedetailz,Ignoring unsupported fields in the request: ) fastapirrkeysrrcrrr)r&rr input_keys unexpectedunuseds r _validate_requestzBaseHandler._validate_request\s))))))%%  # /#gd.FHZ\_\a\a&b&bbJ o#mW3X3X Y Y   $ .E F F  /7;L;[^b;b;b37!0 88M " " .,1$}2E,F,F  )T-())S00.3!+ 88G   (&+DM&:&:  # 88F   ' 4< ( ( (  ) >.?.T.\5=  2  0*/  '! rmessagesrcg}|D]}|dgd}d|vrg}|dD]z}tj|}|dp|}t|dtrt j|d|d<||{||d<d|vr |d|d<d|vrgn|dpg}t|trd|d g}|D]} | d } | d vr%|dd| dd 4| d vr^|tj tj fvrD| d } t| tr| d} |dd| d| dkry|tj kri| d} t| tr*| d} | d}|rd|d| n| } n| } |dd| d| dkrF|tj tj fvr,|dd| ddda| dkr:|tj kr*|dd| ddd|tj kr(d d|dD|d<|||S)a=Convert OpenAI-format messages to the format expected by HF processors. All modalities extract text. VLM additionally handles ``image_url`` and ``video_url``. MULTIMODAL handles all of the above plus ``input_audio`` and ``audio_url``. For LLMs, the content parts are collapsed into a plain text string. Args: messages (`list[dict]`): OpenAI-format chat messages. modality (`Modality`): The model modality (LLM, VLM, or MULTIMODAL). Returns: `list[dict]`: Processor-compatible messages. role)rr7r2rorD tool_call_idr7r)r@rr@)r input_text output_text) image_url input_imagerurlimage)r@r input_audiodataformatz data:audio/z;base64,audio video_urlvideo audio_url c3&K|] }|dV dS)rNr)r\rs r r`zABaseHandler.get_processor_inputs_from_messages..s&,R,R1QvY,R,R,R,R,R,Rr)rrrdrpr)r3rrrrrryrjoin)rrprocessor_inputsmessager|r2tcrA raw_contentr7 content_typerr audio_b64fmts r "get_processor_inputs_from_messagesz.BaseHandler.get_processor_inputs_from_messagessi7 ,7 ,G%fo"==Fw&& !,/**Br**B ++1rB!"[/377F*.*R _*E*E;%%b))))'1|$(()0)@~&!- 7 7""gkk)>T>T>ZXZK+s++ F(. DDE & d d&v #HHH9%,,fgfo-V-VWWWW!%AAAhS[S_aiatRuFuFu!+.C!#t,,)!%j9%,,gc-J-JKKKK"]22x8CV7V7V")-"8K!+t44*$/$7 )ooh77HKZDCDDDDDQZ)9%,,gc-J-JKKKK![00X(,PXPcAd5d5d9%,,ggkFZ[`Fa-b-bcccc![00XAT5T5T9%,,ggkFZ[`Fa-b-bccc8<''$'HH,R,Ry@Q,R,R,R$R$Ry!  # #F + + + +rr%r)rrrr(rr@__annotations__rrr)rryr'r staticmethodrr]rrrrzrrrrr rrBs  (,+++"suuNCH$$$ -1 ??%?*?#Tk ???? Yd Yt Y Y Y YG6G3GGG\G *4*E#7HJt2t,u****,W\0!0!0!3E0!OS0! 0!0!0!0!dH T$ZH 8H X\]aXbH H H \H H H rrr%r)Mr(rrenumr3rabcrrcollections.abcrconcurrent.futuresrrrtypingrtransformers.utilsr pydantic tokenizersr-rr r r r r:transformers.generation.continuous_batching.continuous_apir4transformers.generation.continuous_batching.requestsr5transformers.generation.continuous_batching.schedulerrrr get_loggerrr X_REQUEST_IDEnumrr"r?r+r)r.rr1rgryrlrrrzr}rrrr]rrrrrrr@rrrr/r4r6rNr_rvrrrrr rs ########$$$$$$%%%%%% &&&&&&,OOOLLLedddddUUUUUUOOOOOO++++++  H % % tyPPPPP9PPPC 05)+#%    ( \ #X. (-l""     !!?11h(B+<(B(B(B(B(BVDT".t.T T@Q....6[ *)  b  , 777 MM +<QUX\Q\DsdW\]`behlbl]lWm(d3iD82( ( ( ( ( ( ( ( VDxD3D4DDDDNRRRRRRRRjL2L2L2L2L2L2L2L2^!!!!&&&&&&&&RA>A>A>A>A>#A>A>A>HD D D D D )D D D Ng1g1g1g1g1+g1g1g1TNGNGNGNGNGNGNGNGb~ ~ ~ ~ ~ ~ ~ ~ ~ ~ r