&`i ddlmZmZmZmZmZmZmZddlZddl m Z er ddlm Z ddl m Z dZdZdd d d d ed edeeeedfdffdZe deddd d d ed edeeedffdZe deedddddeedeed ed edeeedff dZdS)) TYPE_CHECKINGAnyIterableIteratorOptionalSequenceUnionN) PublicAPI) ObjectRef)RemoteFunction dF)yield_obj_refsrefszSequence[ObjectRef] chunk_sizerreturnr zlist[ObjectRef]c |dkrtd|pi}tj|fdt|t |i|\}}|stj||fS||fS)a Call ray.wait and explicitly return the ready objects/results and remaining Ray remote refs. Args: refs: A list of Ray object refs. chunk_size: The `num_returns` parameter to pass to `ray.wait()`. yield_obj_refs: If True, return Ray remote refs instead of results (by calling :meth:`~ray.get`). **kwargs: Additional keyword arguments to pass to `ray.wait()`. Returns: A tuple of two lists, ready and not ready. This is the same as the return value of `ray.wait()`. `chunk_size` must be >= 1 num_returns) ValueErrorraywaitminlenget)rrrkwargsreadys d/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/helpers.py_wait_and_get_single_batchr s(A~~4555 \rF(  CII.. KE4 $wu~~t## $;alpha) stabilityrrc+K|dkrtdd|vrtd|rt|f||d|\}}|Ed{V|dSdS)aGiven a list of Ray task references, yield results as they become available. Unlike calling :meth:`~ray.get` on a list of references (i.e., `ray.get(refs)`) which waits for all results to be ready, this function begins to yield result as soon as a batch of `chunk_size` results are ready. .. note:: Generally there is no guarantee on the order of results. For example, the first result is not necessarily the first one completed, but rather the first one submitted in the first available batch (See :meth:`~ray.wait` for more details about preservation of submission order). .. note:: Use this function instead of calling :meth:`~ray.get` inside a for loop. See https://docs.ray.io/en/latest/ray-core/patterns/ray-get-loop.html for more details. Example: Suppose we have a function that sleeps for x seconds depending on the input. We expect to obtain a partially sorted list of results. .. testcode:: python import ray import time @ray.remote def f(x): time.sleep(x) return x refs = [f.remote(i) for i in [10, 4, 6, 8, 2]] for x in ray.util.as_completed(refs, chunk_size=2): print(x) .. testoutput:: :options: +MOCK # Output: 4 2 6 8 10 Args: refs: A list of Ray object refs. chunk_size: The number of tasks to wait for in each iteration (default 10). The parameter is passed as `num_returns` to :meth:`~ray.wait` internally. yield_obj_refs: If True, return Ray remote refs instead of results (by calling :meth:`~ray.get`). **kwargs: Additional keyword arguments to pass to :meth:`~ray.wait`, e.g., `timeout` and `fetch_local`. Yields: Union[Any, ObjectRef]: The results (or optionally their Ray references) of the Ray tasks as they complete. rrrz6Use the `chunksize` argument instead of `num_returns`.r$N)rr )rrrrresultss r as_completedr'8s|A~~4555QRRR 2  !)       r!)backpressure_sizerrfnr itemsr(c+K|td}n|dkrtd|dkrtdd|vrtdg}|D]X}|||t ||krt |f||d |\}}|Ed{VYt |f||d |Ed{VdS) a Apply a Ray remote function to a list of items and return an iterator that yields the completed results as they become available. This helper function applies backpressure to control the number of pending tasks, following the design pattern described in https://docs.ray.io/en/latest/ray-core/patterns/limit-pending-tasks.html. .. note:: There is generally no guarantee on the order of results. Example: Suppose we have a function that sleeps for x seconds depending on the input. We expect to obtain a partially sorted list of results. .. testcode:: python import ray import time @ray.remote def f(x): time.sleep(x) return x # Example 1: chunk_size=2 for x in ray.util.map_unordered(f, [10, 4, 6, 8, 2], chunk_size=2): print(x) .. testoutput:: :options: +MOCK 4 2 6 8 10 .. testcode:: python # Example 2: backpressure_size=2, chunk_size=1 for x in ray.util.map_unordered(f, [10, 4, 6, 8, 2], backpressure_size=2, chunk_size=1): print(x) .. testoutput:: :options: +MOCK 4 10 6 8 2 Args: fn: A remote function to apply to the list of items. For more complex use cases, use Ray Data's :meth:`~ray.data.Dataset.map` / :meth:`~ray.data.Dataset.map_batches` instead. items: An iterable of items to apply the function to. backpressure_size: Maximum number of in-flight tasks allowed before calling a blocking :meth:`~ray.wait` (default 100). If None, no backpressure is applied. chunk_size: The number of tasks to wait for when the number of in-flight tasks exceeds `backpressure_size`. The parameter is passed as `num_returns` to :meth:`~ray.wait` internally. yield_obj_refs: If True, return Ray remote refs instead of results (by calling :meth:`~ray.get`). **kwargs: Additional keyword arguments to pass to :meth:`~ray.wait`, e.g., `timeout` and `fetch_local`. Yields: Union[Any, ObjectRef]: The results (or optionally their Ray references) of the Ray tasks as they complete. .. seealso:: :meth:`~ray.util.as_completed` Call this method for an existing list of Ray object refs. :meth:`~ray.data.Dataset.map` Use Ray Data APIs (e.g., :meth:`~ray.data.Dataset.map` and :meth:`~ray.data.Dataset.map_batches`) for better control and complex use cases, e.g., functions with multiple arguments. .. note:: This is an altenative to `pool.imap_unordered()` in Ray's Actor-based `multiprocessing.Pool`. See https://docs.ray.io/en/latest/ray-more-libs/multiprocessing.html for more details. Ninfrz#backpressure_size must be positive.rrrz7Use the `chunk_size` argument instead of `num_returns`.r$)floatrappendremoterr r') r)r*r(rrrritemr&s r map_unorderedr1sNz #(<< a  >???A~~4555RSSS D   BIIdOO$$$ t99) ) )6%- MGT          !)              r!)typingrrrrrrr rray.util.annotationsr r ray.remote_functionr DEFAULT_CHUNK_SIZEDEFAULT_BACKPRESSURE_SIZEintbooltuplelistr r'r1r!rr<sTTTTTTTTTTTTTTTTTT ******3222222 ! ### ## #  4c;&' (*; ;< ####L W) JJJ JJ J eC$%& JJJJZ W (A( y y y y C=y  } y  y  y eC$%&y y y y y y r!