o i+ @sdZddlmZddlZddlmZmZmZmZddl m Z ddl m Z m Z mZmZddlmZmZmZddlmZdd lmZdd lmZmZmZdd lmZdd lmZdd l m!Z!e rfddl"m#Z#m$Z$edddZ% edddZ& edddZ' edddZ( e ddGdddee%e&e'fZ)edddZ* e Gdddee*Z+edddZ, edddZ- e ddGdd d ee%e&e,e*e'fZ.dS)!a:Decision node implementation for conditional branching in graph execution. This module provides the Decision node type and related classes for implementing conditional branching logic in parallel control flow graphs. Decision nodes allow the graph to choose different execution paths based on runtime conditions. ) annotationsN) AsyncIterableCallableIterableSequence) dataclass) TYPE_CHECKINGAnyGeneric get_origin)NeverSelfTypeVar)BaseNode)NodeID)Path PathBuilderTransformFunction)NodeStep)TypeOrTypeExpression)GraphBuildingError)AnyDestinationNodeDestinationNodeStateTT)infer_varianceDepsTHandledTT)kw_onlyc@sDeZdZUdZded< ded< ded< dd d ZdddZdS)DecisionzDecision node for conditional branching in graph execution. A Decision node evaluates conditions and routes execution to different branches based on the input data type or custom matching logic. ridzlist[DecisionBranch[Any]]branches str | NonenotebranchDecisionBranch[T]return%Decision[StateT, DepsT, HandledT | T]cCst|j|j|g|jdS)zAdd a new branch to this decision. Args: branch: The branch to add to this decision. Returns: A new Decision with the additional branch. )r r!r#)rr r!r#)selfr$r)b/var/www/html/karishye-ai-python/venv/lib/python3.10/site-packages/pydantic_graph/beta/decision.pyr$9s zDecision.branchinputsrr cCstd)aqForces this type to be contravariant in the HandledT type variable. This is an implementation detail of how we can type-check that all possible input types have been exhaustively covered. Args: inputs: Input data of handled types. Raises: RuntimeError: Always, as this method should never be executed. zKThis method should never be called, it is just defined for typing purposes.) RuntimeError)r(r+r)r)r*_force_handled_contravariantDs z%Decision._force_handled_contravariantN)r$r%r&r')r+rr&r )__name__ __module__ __qualname____doc____annotations__r$r-r)r)r)r*r(s   rSourceTc@s8eZdZUdZded< ded< ded< ded <d S) DecisionBranchaRepresents a single branch within a decision node. Each branch defines the conditions under which it should be taken and the path to follow when those conditions are met. Note: with the current design, it is actually _critical_ that this class is invariant in SourceT for the sake of type-checking that inputs to a Decision are actually handled. See the `# type: ignore` comment in `tests.graph.beta.test_graph_edge_cases.test_decision_no_matching_branch` for an example of how this works. TypeOrTypeExpression[SourceT]sourceCallable[[Any], bool] | Nonematchesrpathzlist[AnyDestinationNode] destinationsN)r.r/r0r1r2r)r)r)r*r4Ws   r4OutputT NewOutputTF)initc@seZdZUdZded< ded< ded< ded < d0ddZddd1ddZddd2ddZd3d"d#Zddd$d4d)d*Z d5d.d/Z dS)6DecisionBranchBuilderaYBuilder for constructing decision branches with fluent API. This builder provides methods to configure branches with destinations, forks, and transformations in a type-safe manner. Instances of this class should be created using [`GraphBuilder.match`][pydantic_graph.beta.graph_builder.GraphBuilder], not created directly. !Decision[StateT, DepsT, HandledT] _decisionr5_sourcer7_matches#PathBuilder[StateT, DepsT, OutputT] _path_builderdecisionr6r8 path_buildercCs||_||_||_||_dSN)r@rArBrD)r(rEr6r8rFr)r)r*__init__s  zDecisionBranchBuilder.__init__Nfork_id destinationLDestinationNode[StateT, DepsT, OutputT] | type[BaseNode[StateT, DepsT, Any]]extra_destinationsrJr"r&DecisionBranch[SourceT]cGsVt|p|}tdd|D}dd|g|RD}t|j|j|jj|d|i|dS)aSet the destination(s) for this branch. Args: destination: The primary destination node. *extra_destinations: Additional destination nodes. fork_id: Optional node ID to use for the resulting broadcast fork if multiple destinations are provided. Returns: A completed DecisionBranch with the specified destinations. css|] }t|p |VqdSrG)r .0dr)r)r* sz+DecisionBranchBuilder.to..cSs"g|] }t|r t|n|qSr))inspectisclassrrOr)r)r* s"z,DecisionBranchBuilder.to..rJr6r8r9r:)r tupler4rArBrDto)r(rKrJrMr:r)r)r*rXs zDecisionBranchBuilder.to get_forks3Callable[[Self], Sequence[DecisionBranch[SourceT]]]cCs\||}dd|D}|std|d|jj||d}dd|D}t|j|j||dS)aaBroadcast this decision branch into multiple destinations. Args: get_forks: The callback that will return a sequence of decision branches to broadcast to. fork_id: Optional node ID to use for the resulting broadcast fork. Returns: A completed DecisionBranch with the specified destinations. cSsg|]}|jqSr))r9)rPbr)r)r*rUsz3DecisionBranchBuilder.broadcast..z The call to z4 returned no branches, but must return at least one.rIcSsg|] }|jD]}|qqSr))r:)rPfdprQr)r)r*rUsrV)rrD broadcastr4rArB)r(rYrJfork_decision_branches new_pathsr9r:r)r)r*r]s zDecisionBranchBuilder.broadcastfunc5TransformFunction[StateT, DepsT, OutputT, NewOutputT]CDecisionBranchBuilder[StateT, DepsT, NewOutputT, SourceT, HandledT]cCt|j|j|j|j|dS)aApply a transformation to the branch's output. Args: func: Transformation function to apply. Returns: A new DecisionBranchBuilder where the provided transform is applied prior to generating the final output. rEr6r8rF)r>r@rArBrD transform)r(r`r)r)r*re   zDecisionBranchBuilder.transformrJdownstream_join_idr(DecisionBranchBuilder[StateT, DepsT, Iterable[T], SourceT, HandledT] | DecisionBranchBuilder[StateT, DepsT, AsyncIterable[T], SourceT, HandledT]rh:DecisionBranchBuilder[StateT, DepsT, T, SourceT, HandledT]cCs"t|j|j|j|jj||ddS)a3Spread the branch's output. To do this, the current output must be iterable, and any subsequent steps in the path being built for this branch will be applied to each item of the current output in parallel. Args: fork_id: Optional ID for the fork, defaults to a generated value downstream_join_id: Optional ID of a downstream join node which is involved when mapping empty iterables Returns: A new DecisionBranchBuilder where mapping is performed prior to generating the final output. rgrd)r>r@rArBrDmap)r(rJrhr)r)r*rks zDecisionBranchBuilder.maplabelstr@DecisionBranchBuilder[StateT, DepsT, OutputT, SourceT, HandledT]cCrc)aRApply a label to the branch at the current point in the path being built. These labels are only used in generated mermaid diagrams. Args: label: The label to apply. Returns: A new DecisionBranchBuilder where the label has been applied at the end of the current path being built. rd)r>r@rArBrDrl)r(rlr)r)r*rlrfzDecisionBranchBuilder.label)rEr?r6r5r8r7rFrC)rKrLrMrLrJr"r&rN)rYrZrJr"r&rN)r`rar&rb)r(rirJr"rhr"r&rj)rlrmr&rn) r.r/r0r1r2rHrXr]rerkrlr)r)r)r*r>s(    r>)/r1 __future__rrScollections.abcrrrr dataclassesrtypingrr r r typing_extensionsr r rpydantic_graphrpydantic_graph.beta.id_typesrpydantic_graph.beta.pathsrrrpydantic_graph.beta.steprpydantic_graph.beta.utilrpydantic_graph.exceptionsrpydantic_graph.beta.node_typesrrrrrrrr3r4r;r<r>r)r)r)r*sF            * ' $