A node that determines the next step in the graph's execution flow based on the presence and content of a specified key in the graph's state. It extends the BaseNode by adding condition-based logic to the execution process. This node type is used to implement branching logic within
| 10 | |
| 11 | |
| 12 | class ConditionalNode(BaseNode): |
| 13 | """ |
| 14 | A node that determines the next step in the graph's execution flow based on |
| 15 | the presence and content of a specified key in the graph's state. It extends |
| 16 | the BaseNode by adding condition-based logic to the execution process. |
| 17 | |
| 18 | This node type is used to implement branching logic within the graph, allowing |
| 19 | for dynamic paths based on the data available in the current state. |
| 20 | |
| 21 | It is expected that exactly two edges are created out of this node. |
| 22 | The first node is chosen for execution if the key exists and has a non-empty value, |
| 23 | and the second node is chosen if the key does not exist or is empty. |
| 24 | |
| 25 | Attributes: |
| 26 | key_name (str): The name of the key in the state to check for its presence. |
| 27 | |
| 28 | Args: |
| 29 | key_name (str): The name of the key to check in the graph's state. This is |
| 30 | used to determine the path the graph's execution should take. |
| 31 | node_name (str, optional): The unique identifier name for the node. Defaults |
| 32 | to "ConditionalNode". |
| 33 | |
| 34 | """ |
| 35 | |
| 36 | def __init__( |
| 37 | self, |
| 38 | input: str, |
| 39 | output: List[str], |
| 40 | node_config: Optional[dict] = None, |
| 41 | node_name: str = "Cond", |
| 42 | ): |
| 43 | """ |
| 44 | Initializes an empty ConditionalNode. |
| 45 | """ |
| 46 | super().__init__(node_name, "conditional_node", input, output, 2, node_config) |
| 47 | |
| 48 | try: |
| 49 | self.key_name = self.node_config["key_name"] |
| 50 | except (KeyError, TypeError) as e: |
| 51 | raise NotImplementedError( |
| 52 | "You need to provide key_name inside the node config" |
| 53 | ) from e |
| 54 | |
| 55 | self.true_node_name = None |
| 56 | self.false_node_name = None |
| 57 | self.condition = self.node_config.get("condition", None) |
| 58 | self.eval_instance = EvalWithCompoundTypes() |
| 59 | self.eval_instance.functions = {"len": len} |
| 60 | |
| 61 | def execute(self, state: dict) -> dict: |
| 62 | """ |
| 63 | Checks if the specified key is present in the state and decides the next node accordingly. |
| 64 | |
| 65 | Args: |
| 66 | state (dict): The current state of the graph. |
| 67 | |
| 68 | Returns: |
| 69 | str: The name of the next node to execute based on the presence of the key. |
no outgoing calls
no test coverage detected