&`ir ddlZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z m Z mZddlZddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!e d Z"e!GddeZ#dS)N)chain) AnyCallableDictListLiteralOptionalTupleTypeVarUnion) DAGNodeBase)build_compiled_dag_from_ray_dag) _PyObjScanner)ChannelOutputType)AutoTransportType) Communicator)TorchTensorType)Device) DeveloperAPITceZdZdZdeedeeefdeeefdeeeffdZde dfdZ dDd e e ee fd eddedefdZedefdZejdeddfdZdeefdZdeeeffdZdeeeffdZdeeeffdZdefdZdeeeffdZdZ dEde ede ed ed!e ed"e ed#e ed$e e e efdd%fd&Zd d'd(ede ej d)ffd*Z!de dfd+Z"de dfd,Z# dFd/Z$d-d.de%fd0Z&dGd1Z'd2Z(dHd3Z)d4ed5e*d6e*fd7Z+de ej d)ffd8Z,d9e ed:eeefd;eeefdZ.d?Z/d@eeeffdAZ0dBefdCZ1dS)IDAGNodea5Abstract class for a node in a Ray task graph. A node has a type (e.g., FunctionNode), data (e.g., function options and body), arguments (Python values, DAGNodes, and DAGNodes nested within Python argument values) and options (Ray API .options() used for function, class or class method) argskwargsoptionsother_args_to_resolvec&|pg|_|pi|_|pi|_|pi|_g|_t jj|_d|_ | |_ i|_ t|_d|_d|_dS)a7 args: args (Tuple[Any]): Bound node arguments. ex: func_or_class.bind(1) kwargs (Dict[str, Any]): Bound node keyword arguments. ex: func_or_class.bind(a=1) options (Dict[str, Any]): Bound node options arguments. ex: func_or_class.options(num_cpus=2) other_args_to_resolve (Dict[str, Any]): Bound kwargs to resolve that's specific to subclass implementation without exposing as args in base class, example: ClassMethodNode FN) _bound_args _bound_kwargs_bound_options_bound_other_args_to_resolve_downstream_nodesuuiduuid4hex _stable_uuid_args_contain_nested_dag_node_collect_upstream_nodes_upstream_nodescache_from_last_executer _type_hint_original_type_hintis_cgraph_output_node)selfrrrrs d/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/dag/dag_node.py__init__zDAGNode.__init__)s&(,zr-3\r.5m ! 'R ) 35!JLL, .3*150L0L0N0N(*$->-@-@ AE &+"""returncXg}t|jdsJ|jD]}t|tr||-t }||}|||t|dk|_ t }||j |j g}||||D]}|j ||S)a Retrieve upstream nodes and update their downstream dependencies. Currently, the DAG assumes that all DAGNodes in `args`, `kwargs`, and `other_args_to_resolve` are upstream nodes. However, Ray Compiled Graphs builds the upstream/downstream relationship based only on args. Be cautious when persisting DAGNodes in `other_args_to_resolve` and kwargs in the future. TODO (kevin85421): Currently, the upstream nodes and downstream nodes have circular references. Therefore, it relies on the garbage collector to clean them up instead of reference counting. We should consider using weak references to avoid circular references. __iter__r)hasattrr isinstancerappendr find_nodesextendclearlenr'rr!r")r.upstream_nodesargscanner dag_nodesother_upstream_nodes upstream_nodes r/r(zDAGNode._collect_upstream_nodes^s8+-t'44444# H HC#w'' H%%c****'//#..s33 %%i000 58^^a5G22//070B0B"1 1 1  2333 + 9 9M  + 2 24 8 8 8 8r1autodefaultF transportdevice)rCcpugpucuda _static_shape_direct_returnc, t|}nH#t$r;ddtD}td|d|dwxYw|dkrt||||_n|dkrt d ||| |_n}|d krt d ||| |_n^|d krt ||||_n@t |tstd |d t |||| |_|S)a Configure the torch tensor transport for this node. Args: transport: Specifies the tensor transport mechanism. - "accelerator": Tensors are communicated using accelerator-specific backends (e.g., NCCL, XLA, or vendor-provided transport). This is the recommended option for most use cases, as it supports extensibility and future hardware backends. - "nccl": Tensors are passed explicitly via NCCL. This option is kept for backwards compatibility and may be removed in the future. Use "accelerator" instead unless you have legacy requirements. - "shm": Tensors are passed via host shared memory and gRPC. Typically used when accelerator-based transport is unavailable or not suitable. - "auto" (default): The system automatically selects the appropriate transport mechanism based on the sender and receiver, usually preferring accelerator-based transport when available. device: The target device to use for the tensor transport. "default": The tensor will maintain its original device placement from the sender "cpu": The tensor will be explicitly moved to CPU device in the receiver "gpu" or "cuda": The tensor will be explicitly moved to GPU device in the receiver _static_shape: A hint indicating whether the shape(s) and dtype(s) of tensor(s) contained in this value always remain the same across different executions of the DAG. If this is True, the transport will be more efficient. _direct_return: Whether the tensor is sent directly or inside of other data. If a "nccl" transport is used, this allows the sender and receiver to eliminate performance overhead from an additional data transfer. z, c3,K|]}d|jdVdS)'N)value).0ds r/ z0DAGNode.with_tensor_transport..s,%E%En!'nnn%E%E%E%E%E%Er1zInvalid device 'z'. Valid options are: .rB)rErIrJnccl accelerator)rDrErIrJshmzInvalid transport type: zd. Transport must be one of 'auto', 'nccl', 'shm', 'accelerator' or an instance of Communicator type.)r ValueErrorjoinrr+rr6r)r.rDrErIrJ valid_devicess r/with_tensor_transportzDAGNode.with_tensor_transportsH F^^FF    II%E%Ef%E%E%EEEMQ6QQQQQ     /+-DOO & -'+- DOO - ' '-'+- DOO %  -+-DOO i66  8y888 .#+- DO  s AAc|jSN)r+r.s r/ type_hintzDAGNode.type_hints r1r]Nc`t|jtr |j|_||_dSr[)r6r+rr,)r.r]s r/r]zDAGNode.type_hints, do'8 9 9 7'+D $#r1c|jS)z,Return the tuple of arguments for this node.)rr\s r/get_argszDAGNode.get_argss r1c4|jS)z3Return the dict of keyword arguments for this node.)rcopyr\s r/ get_kwargszDAGNode.get_kwargss!&&(((r1c4|jS)z3Return the dict of options arguments for this node.)r rbr\s r/ get_optionszDAGNode.get_optionss"'')))r1c4|jS)zAReturn the dict of other args to resolve arguments for this node.)r!rbr\s r/get_other_args_to_resolvez!DAGNode.get_other_args_to_resolves055777r1c|jS)zReturn stable uuid for this node. 1) Generated only once at first instance creation 2) Stable across pickling, replacement and JSON serialization. )r&r\s r/get_stable_uuidzDAGNode.get_stable_uuids   r1cKi}|jD]0\}}t|tjr |d{V||<+|||<1|S)zGets cached object refs from the last call to execute(). After this DAG is executed through execute(), retrieves a map between node UUID to a reference to the return value of the default executor on that node. N)r*itemsr6asyncioTask)r.cache node_uuidrNs r/!get_object_refs_from_last_executez)DAGNode.get_object_refs_from_last_executeso  $ < B B D D ) ) Iu%.. )).;;;;;;i  #(i   r1ci|_dSr[)r*r\s r/ clear_cachezDAGNode.clear_caches')$$$r1create_submit_timeout_buffer_size_bytesenable_asyncio_max_inflight_executions_max_buffered_results_overlap_gpu_communication_default_communicatorzray.dag.CompiledDAGc ddlm}|} || j}|jrt dd|_t ||||||||S)a Compile an accelerated execution path for this DAG. Args: _submit_timeout: The maximum time in seconds to wait for execute() calls. None means using default timeout, 0 means immediate timeout (immediate success or timeout without blocking), -1 means infinite timeout (block indefinitely). _buffer_size_bytes: The initial buffer size in bytes for messages that can be passed between tasks in the DAG. The buffers will be automatically resized if larger messages are written to the channel. enable_asyncio: Whether to enable asyncio for this DAG. _max_inflight_executions: The maximum number of in-flight executions that can be submitted via `execute` or `execute_async` before consuming the output using `ray.get()`. If the caller submits more executions, `RayCgraphCapacityExceeded` is raised. _max_buffered_results: The maximum number of results that can be buffered at the driver. If more than this number of results are buffered, `RayCgraphCapacityExceeded` is raised. Note that when result corresponding to an execution is retrieved (by calling `ray.get()` on a `CompiledDAGRef` or `CompiledDAGRef` or await on a `CompiledDAGFuture`), results corresponding to earlier executions that have not been retrieved yet are buffered. _overlap_gpu_communication: (experimental) Whether to overlap GPU communication with computation during DAG execution. If True, the communication and computation can be overlapped, which can improve the performance of the DAG execution. If None, the default value will be used. _default_communicator: The default communicator to use to transfer tensors. Three types of values are valid. (1) Communicator: For p2p operations, this is the default communicator to use for nodes annotated with `with_tensor_transport()` and when shared memory is not the desired option (e.g., when transport="nccl", or when transport="auto" for communication between two different GPUs). For collective operations, this is the default communicator to use when a custom communicator is not specified. (2) "create": for each collective operation without a custom communicator specified, a communicator is created and initialized on its involved actors, or an already created communicator is reused if the set of actors is the same. For all p2p operations without a custom communicator specified, it reuses an already created collective communicator if the p2p actors are a subset. Otherwise, a new communicator is created. (3) None: a ValueError will be thrown if a custom communicator is not specified. Returns: A compiled DAG. r) DAGContextNzIt is not allowed to call `experimental_compile` on the same DAG object multiple times no matter whether `teardown` is called or not. Please reuse the existing compiled DAG or create a new one.T)ray.dagr| get_currentbuffer_size_bytesr-rVr) r.rtrurvrwrxryrzr|ctxs r/experimental_compilezDAGNode.experimental_compilest '&&&&&$$&&  %!$!6   % N &*".     $ ! & !   r1)_ray_cache_refsrzray.actor.ActorHandlec\fd}||}|r |j|_|S)aExecute this DAG using the Ray default executor _execute_impl(). Args: _ray_cache_refs: If true, stores the default executor's return values on each node in this DAG in a cache. These should be a mix of: - ray.ObjectRefs pointing to the outputs of method and function nodes - Serve handles for class nodes - resolved values representing user input at runtime c|jiSr[) _execute_impl)noderrs r/executorz!DAGNode.execute..executorzs%4%t6v66 6r1)apply_recursivernr*)r.rrrrresults `` r/executezDAGNode.executemsN 7 7 7 7 7 7%%h//  :+3>D ( r1cg}|D]0}t|tr||vr||1|D]0}t|tr||vr||1|D]0}t|tr||vr||1|S)aTReturn the list of nodes specified as top-level args. For example, in `f.remote(a, [b])`, only `a` is a top-level arg. This list of nodes are those that are typically resolved prior to task execution in Ray. This does not include nodes nested within args. For that, use ``_get_all_child_nodes()``. )r`r6rr7rcvaluesrg)r.childrenas r/_get_toplevel_child_nodesz!DAGNode._get_toplevel_child_nodess ' 'A!W%% 'H$$OOA&&&""))++ ' 'A!W%% 'H$$OOA&&&//1188:: ' 'A!W%% 'H$$OOA&&&r1ct}g}||j|j|jgD]}||vr||||S)aReturn the list of nodes referenced by the args, kwargs, and args_to_resolve in current node, even they're deeply nested. Examples: f.remote(a, [b]) -> [a, b] f.remote(a, [b], key={"nested": [c]}) -> [a, b, c] )rr8rrr!r7r:)r.r>rns r/_get_all_child_nodeszDAGNode._get_all_child_nodess| //## "1    # #A  """ r1fnCallable[[DAGNode], T]cHi}t}||j|j|jgD]}||vr||||<||\}}}||||||S)apApply and replace all immediate child nodes using a given function. This is a shallow replacement only. To recursively transform nodes in the DAG, use ``apply_recursive()``. Args: fn: Callable that will be applied once to each child of this node. Returns: New DAGNode after replacing all child nodes. ) rr8rrr! replace_nodesr:_copyre)r.r replace_tabler>rnew_args new_kwargsnew_other_args_to_resolves r/"_apply_and_replace_all_child_nodesz*DAGNode._apply_and_replace_all_child_nodess  //&& "1    / /D=((&(bhh d#:A:O:O ; ; 7*7  zz j$"2"2"4"46O   r1ctjdksGdd}|n |jjvrj|jS|fdS)aApply callable on each node in this DAG in a bottom-up tree walk. Args: fn: Callable that will be applied once to each node in the DAG. It will be applied recursively bottom-up, so nodes can assume the fn has been applied to their args already. Returns: Return type of the fn after application to the tree. _CachingFnceZdZdZddZdS)+DAGNode.apply_recursive.._CachingFncRi|_||_|j|j_d|_dSr[)rnrinput_node_uuid)r.rs r/r0z4DAGNode.apply_recursive.._CachingFn.__init__s)!#DJ DG$(JDGM+/D(((r1rrcddlm}|j|jvr"|||j|j<t ||r3|js |j|_n|j|jkrtd|j|jS)Nr InputNodez/Each DAG should only have one unique InputNode.)ray.dag.input_noderr&rnrr6rAssertionError)r.rrs r/__call__z4DAGNode.apply_recursive.._CachingFn.__call__s<<<<<<( ::8<  4#45!$ 22#3373DD00!1T5FFF"0 Q## :d&788r1N)rr)__name__ __module__ __qualname__r0rr1r/rrs7000 9 9 9 9 9 9r1c.|Sr[)r)rrs r/z)DAGNode.apply_recursive.. sT11"55r1)typerr&rnr)r.rrs ` r/rzDAGNode.apply_recursivesBxx L00 9 9 9 9 9 9 9 9*BBB BH,,x 122r  3 35555     r1ct}|g}d}|r|d}|jr||j||vr}|jr|t d|d||}|||| tj |j |j gD]}||vr| ||dSdS)z Traverse all nodes in the connected component of the DAG that contains the `self` node, and apply the given function to each node. Nrz^The DAG was compiled more than once. The following two nodes call `experimental_compile`: (1) z, (2) ) setpopr'_raise_nested_dag_node_errorrr-rVaddr from_iterabler"r)r7)r.rvisitedqueuecgraph_output_noderneighbors r/traverse_and_applyzDAGNode.traverse_and_applys? %%04' /99Q<r?s r/rz$DAGNode._raise_nested_dag_node_error@s  C#w'' '//#..u55  y>>A%%$BYBBBB"BBB& L   r1cddlm}|}t||sHt|jdkrt d|d|jd}t||H|S)zV Return the root node of the DAG. The root node must be an InputNode. rrzWNo InputNode found in the DAG: when traversing upwards, no upstream node was found for rR)rrr6r;r)rV)r.rrs r/ _find_rootzDAGNode._find_root[s 100000T9-- +4'((A-- >6:>>>'*D T9-- + r1source_input_list predicate_fnapply_fnci}t}||D]}||r||vr||||< ||}||S)aa Apply a given function to DAGNodes in source_input_list, and return the replaced inputs without mutating or coping any DAGNode. Args: source_input_list: Source inputs to extract and apply function on all children DAGNode instances. predicate_fn: Applied on each DAGNode instance found and determine if we should apply function to it. Can be used to filter node types. apply_fn: Function to apply on the node on bound attributes. Example:: apply_fn = lambda node: node._get_serve_deployment_handle( node._deployment, node._bound_other_args_to_resolve ) Returns: replaced_inputs: Outputs of apply_fn on DAGNodes in source_input_list that passes predicate_fn. )rr8rr:)r.rrrrr>rreplaced_inputss r/apply_functionalzDAGNode.apply_functionalks4 //&&'899 5 5D|D!! 5d-&?&?&.htnn d#!// >> r1ct)z?Execute this node, assuming args have been transformed already.NotImplementedError)r.rrs r/rzDAGNode._execute_impls "!r1rr new_optionsrctz3Return a copy of this node with the given new args.r)r.rrrrs r/ _copy_implzDAGNode._copy_impls "!r1c|||||}|j|_tj|j|_tj|j|_|Sr)rr&rbdeepcopyr+r,)r.rrrrinstances r/rz DAGNode._copys\?? j+/H  !% 1"mDO<<'+}T5M'N'N$r1c|jS)=Required due to overriding `__getattr__` else pickling fails.)__dict__r\s r/ __getstate__zDAGNode.__getstate__s }r1rPc:|j|dS)rN)rupdate)r.rPs r/ __setstate__zDAGNode.__setstate__s Qr1attrc|dkr tdt|d|dkr tdt|d||S)Nbindz .bind() cannot be used again on  remotez.remote() cannot be used on z:. To execute the task graph for this node, use .execute().)AttributeErrorr__getattribute__)r.rs r/ __getattr__zDAGNode.__getattr__sz 6>> !QDJJ!Q!Q!QRR R X   7tDzz777  ((.. .r1)rBrCFF)NNFNNNrs)rrr2r)rr)r2r)2rrr__doc__r rrstrr0rr(r r rrboolrYpropertyrr]setterr`rcrergrirprrfloatintrray ObjectRefrrrrrrrrrrrrrrrrrrr1r/rrsF3+Cj3+S#X3+c3h 3+ $CH~ 3+3+3+3+j-i----b9?;D#$ RRE#|"345R78R R  RRRRh,X$#4$$$$$  %*    )DcN)))) *T#s(^**** 84S>8888!!!!! c3h    *** ,0,0$26/359DLT T !%T %SMT  T #+3- T (} T %-TNT  (lC.?(@AT  T T T T n.3&* s}55 6*4 ?:d9o2& *& & & & & P, ":, q, , , , \0/0/0/0/d   6 ### ####J" s}55 6"""" "s)"cN"#s(^ " $(S> "  """"s)cN#s(^  $(S>     d38n     / / / / / / /r1r)$rlrbr# itertoolsrtypingrrrrrr r r r r ray.dag.baser ray.dag.compiled_dag_noderray.dag.py_obj_scannerrray.experimental.channelr,ray.experimental.channel.auto_transport_typer%ray.experimental.channel.communicatorr*ray.experimental.channel.torch_tensor_typerray.experimental.util.typesrray.util.annotationsrrrrr1r/rs                        $$$$$$EEEEEE000000666666JJJJJJ>>>>>>FFFFFF......------ GCLLa /a /a /a /a /ka /a /a /a /a /r1