o 0jK@sddlZddlZddlZddlZddlmZmZmZddl m Z m Z ddl m Z ddlmZddlmZgdZGd d d e Zd efd d ZddZGdddZdaddZddZddZd"defddZd#ddZedGd d!d!eZ dS)$N)AnyDictOptional) BaseModelmodel_validator)Literal)logging)class_requires_deps)fastdeploy-server vllm-server sglang-servermlx-vlm-serverllama-cpp-serverc@sdeZdZUdZeded<dZeeed<dZ e ed<dZ ee ee fed<ed d d d ZdS) GenAIConfignative)rr r r rrbackendN server_urlmax_concurrency client_kwargsafter)modecCs.|jtvr|jdurtdt|jd|S)Nz(`server_url` must not be `None` for the z backend.)rSERVER_BACKENDSr ValueErrorreprselfrf/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddlex/inference/models/common/genai.pycheck_server_url1s zGenAIConfig.check_server_url)__name__ __module__ __qualname__rr__annotations__rrstrrintrrrrr rrrrr$s  rreturncCs.|durdSt|tr |nt|}|jtvS)z:Return whether the config targets a remote server backend.NF) isinstancerZmodel_validaterr) genai_configZ validatedrrruses_server_backend:s r*cCs t| SN)r*)r)rrrneed_local_modelFs r,c@sTeZdZdZddZddZddZdd efd d Zd d Z ddZ e ddZ dS)_AsyncThreadManageraH Manages an asyncio event loop running in a dedicated background thread. This class provides a bridge between synchronous code and async operations, allowing sync code to submit coroutines to be executed in the background event loop. Thread Safety: - Only `run_async()` and `stop()` are designed to be called from other threads - All internal asyncio operations are executed within the event loop thread - Uses `run_coroutine_threadsafe()` and `call_soon_threadsafe()` for cross-thread communication as recommended by Python documentation Lifecycle: 1. `start()` - Creates a daemon thread running `loop.run_forever()` 2. `run_async(coro)` - Submits coroutines to the loop (returns Future) 3. `stop(timeout)` - Gracefully shuts down: waits for tasks, cancels remaining, then cleans up resources Example: manager = _AsyncThreadManager() manager.start() future = manager.run_async(some_async_function()) result = future.result(timeout=10) manager.stop() cCs0d|_d|_d|_t|_t|_d|_dS)NF)loopthreadstopped threadingEvent _event_start_event_cleanup_done_shutting_downrrrr__init__gs    z_AsyncThreadManager.__init__cs`rdSd_d_jjfdd}tj|dd_j j dS)a9 Start the background event loop thread. This method is idempotent - calling it multiple times has no effect if the loop is already running. The method blocks until the event loop is fully initialized and ready to accept tasks (synchronized via threading.Event). NFcsdt_tjjzjWjd_ dSjd_ w)NT) asyncioZnew_event_loopr.Zset_event_loopr3setZ run_forever_cleanup_loop_internalr4r0rrrr _run_loops       z,_AsyncThreadManager.start.._run_loopT)targetdaemon) is_runningr5r0r3clearr4r1Threadr/startwait)rr:rrrr@qs     z_AsyncThreadManager.startc Cs|jdurdSzwzBt|j}|r1tdt|d|D]}|q|jtj|ddi|j|j t |jdrI|j|j Wnt yd}zt d|WYd}~nd}~wwW|jtddSW|jtddS|jtdw) a# Perform cleanup operations within the event loop thread. IMPORTANT: This method MUST be called from the event loop thread (i.e., in the finally block of _run_loop) because asyncio operations like `all_tasks()`, `task.cancel()` are NOT thread-safe. Cleanup sequence: 1. Cancel all remaining tasks 2. Wait for cancellation to complete 3. Shutdown async generators (prevents ResourceWarning) 4. Shutdown default executor (Python 3.9+) 5. Close the event loop Nz Cancelling z pending tasks during cleanupreturn_exceptionsTshutdown_default_executorz!Error during event loop cleanup: zEvent loop closed successfully)r.r7 all_tasksr debuglencancelZrun_until_completegatherZshutdown_asyncgenshasattrrC Exceptionwarningclose)rpendingtaskerrrr9s6       z*_AsyncThreadManager._cleanup_loop_internal@timeoutc s$sdSd_fdd}zt|j}|jddWn/tjjy8t dddYnt yQ}zt d |WYd}~nd}~wwz j jj Wn tyeYnwjjd d}|stt d jdurjjddjrt d d_d_dS) a Gracefully stop the event loop. This method performs a graceful shutdown sequence: 1. Sets shutting_down flag to reject new tasks 2. Schedules a graceful shutdown coroutine in the event loop that: - Waits for pending tasks to complete (with timeout) - Cancels tasks that don't complete in time 3. Signals the loop to stop 4. Waits for cleanup to complete in the loop thread 5. Joins the background thread Args: timeout: Maximum seconds to wait for pending tasks to complete. Tasks not completed within this time will be cancelled. Default is 5.0 seconds. Thread Safety: This method is safe to call from any thread. It uses only thread-safe mechanisms (run_coroutine_threadsafe, call_soon_threadsafe, Events) to communicate with the event loop thread. NTcstfddtjD}|stddStdt|ddtj|tjdIdH\}}|r[t d t|d d |D]}| qItj |d d iIdHtddS)z Graceful shutdown coroutine that runs inside the event loop. All asyncio operations here are thread-safe because this coroutine executes in the event loop thread. cs g|] }|ur|s|qSr)done).0t current_taskrr s zH_AsyncThreadManager.stop.._graceful_shutdown..z,No pending tasks to wait for during shutdownNzGraceful shutdown: waiting for z pending tasks (timeout=zs))rQZ return_whenzGraceful shutdown: cancelling z$ tasks that did not complete within srBTz%Graceful shutdown coroutine completed) r7rVrDr.r rErFrAZ ALL_COMPLETEDrKrGrH)rMrRZ still_pendingrNrrQrUr_graceful_shutdowns4      z4_AsyncThreadManager.stop.._graceful_shutdowng@rQz"Graceful shutdown timed out after zs, forcing loop stopz Error during graceful shutdown: rPz-Event loop cleanup did not complete within 5szYBackground thread did not terminate in time. Some resources may not be properly released.)r=r5r7run_coroutine_threadsafer.result concurrentfutures TimeoutErrorr rKrJZcall_soon_threadsafestop RuntimeErrorr4rAr/joinis_alive)rrQrZfuturerOZcleanup_completedrrYrras>,     z_AsyncThreadManager.stopcCs0|std|jrtdt||j}|S)a Submit a coroutine to be executed in the background event loop. This is the primary method for bridging sync and async code. The coroutine will be scheduled to run in the background thread's event loop. Args: coro: A coroutine object to be executed Returns: concurrent.futures.Future: A future that can be used to: - Wait for the result: future.result(timeout=...) - Check completion: future.done() - Cancel the task: future.cancel() Raises: RuntimeError: If the event loop is not running or is shutting down Thread Safety: This method is safe to call from any thread. It uses `asyncio.run_coroutine_threadsafe()` which is explicitly documented as thread-safe. Example: future = manager.run_async(fetch_data()) result = future.result(timeout=30) zEvent loop is not runningzxEvent loop is shutting down, cannot accept new tasks. Please ensure all async operations complete before calling stop().)r=rbr5r7r\r.)rcororerrr run_async=sz_AsyncThreadManager.run_asynccCs|jduo|j o|j S)z Check if the event loop is currently running and accepting tasks. Returns: bool: True if the loop is running and not closed/stopped N)r. is_closedr0rrrrr=gsz_AsyncThreadManager.is_runningcC|jS)a Check if the event loop is in the process of shutting down. During shutdown, new tasks will be rejected but existing tasks are still being processed. Returns: bool: True if shutdown has been initiated )r5rrrris_shutting_downps z$_AsyncThreadManager.is_shutting_downNrP) r!r"r#__doc__r6r@r9floatrargr=propertyrjrrrrr-Ks )2q* r-cCstdurtatSr+)_async_thread_managerr-rrrrget_async_managersrpcCst}|o |j S)z Check if the async event loop is ready to accept tasks. Returns: bool: True if the event loop is running and not shutting down )rpr=rjmanagerrrris_aio_loop_readysrscCs*t}|s|t|jdSdS)aF Start the global async event loop if not already running. This function also registers an atexit handler to ensure graceful shutdown when the program exits. Note: The atexit handler calls stop() which performs graceful shutdown, waiting for pending tasks to complete before terminating. N)rpr=r@atexitregisterrarqrrrstart_aio_loops  rvrPrQcCs"t}|r|j|ddSdS)a Gracefully close the global async event loop. This function initiates a graceful shutdown sequence that: 1. Waits for pending tasks to complete (up to timeout) 2. Cancels any remaining tasks 3. Cleans up resources (async generators, executor) 4. Closes the event loop Args: timeout: Maximum seconds to wait for pending tasks to complete. Default is 5.0 seconds. r[N)rpr=ra)rQrrrrrclose_aio_loopsrwFc Cst}|s t|std|jrtd||}|r"|Sz|j|dWStjj y<t d|dt yP}z t d|d}~ww)ao Execute a coroutine in the background event loop. This is the main entry point for running async code from sync contexts. It automatically starts the event loop if not already running. Args: coro: The coroutine to execute return_future: If True, return a Future immediately without waiting. If False (default), block until the coroutine completes. timeout: Maximum seconds to wait for completion (only used when return_future=False). None means wait indefinitely. Returns: If return_future=True: concurrent.futures.Future If return_future=False: The result of the coroutine Raises: RuntimeError: If the event loop fails to start or is shutting down concurrent.futures.TimeoutError: If timeout is exceeded Exception: Any exception raised by the coroutine Example: # Blocking call result = run_async(fetch_data(), timeout=30) # Non-blocking call future = run_async(fetch_data(), return_future=True) # ... do other work ... result = future.result() zFailed to start event loopz4Event loop is shutting down, cannot accept new tasksr[zTask timed out after z secondszTask failed with error: N)rpr=rvrbrjrgr]r^r_r`r rKrJerror)rf return_futurerQrrrerOrrrrgs(  rgopenaicsJeZdZ dfdd ZeddZddd d Zd d Zd dZZ S) GenAIClientrNc svddlm}t||_||_d|vrd|d<|dd|i||_|dur/t|dd}||_ t |j|_ dS) Nr) AsyncOpenAIZapi_keynullbase_url r[r) rzr|superr6rZ_max_concurrency_clientrg_get_model_name _model_namer7 Semaphore _semaphore)rrr~rZ model_namekwargsr| __class__rrr6s  zGenAIClient.__init__cCrir+)rrrrr openai_clientszGenAIClient.openai_clientFryc s*fdd}t|dj|d||dS)Nc s\j4IdHjjjj|i|IdHWdIdHS1IdHs'wYdSr+)rrZchatZ completionscreate)argsrrrr&_create_chat_completion_with_semaphores  0zRGenAIClient.create_chat_completion.._create_chat_completion_with_semaphore)modelmessagesrr)rgr)rrryrrrrrcreate_chat_completions z"GenAIClient.create_chat_completioncCst|jdddS)Nr[)rgrrLrrrrrL#szGenAIClient.closec sNz |jjIdH}Wnty }ztd||d}~ww|jdjS)Nz@Failed to get the model list from the OpenAI-compatible server: r)rmodelslistrJrbdataid)rrrOrrrr&s zGenAIClient._get_model_name)rN) r!r"r#r6rnrrrLr __classcell__rrrrr{s r{rk)FN)!r7rtconcurrent.futuresr^r1typingrrrZpydanticrrZtyping_extensionsrutilsr Z utils.depsr rrboolr*r,r-rorprsrvrmrwrgobjectr{rrrrs0     5  >