switch.py

Overview

The switch.py file defines a Switch component within the InfiniFlow system, designed to direct workflow execution based on conditional logic akin to a multi-branch switch statement. It evaluates a set of conditions against inputs obtained from other components and determines the next component(s) to execute. This component is critical in controlling branching logic in process automation, enabling dynamic routing depending on the data or state evaluated at runtime.

The file contains two primary classes:

• SwitchParam: Encapsulates and validates the parameters/configuration of the Switch component.

• Switch: Implements the conditional evaluation logic and determines the next workflow step(s).

Classes and Functions

Class: SwitchParam

class SwitchParam(ComponentParamBase):

Purpose

Defines and validates the parameters that configure the behavior of the Switch component. It inherits from ComponentParamBase and primarily manages the conditions that the Switch evaluates.

Attributes

• conditions (list): A list of condition dictionaries. Each dictionary specifies a logical operator (and / or), a list of items (conditions to evaluate), and a list of target component IDs to route to if the condition is met.

• end_cpn_ids (list): Component IDs representing the default "else" or fallback routes if no conditions match.

• operators (list[str]): Supported operators for condition evaluation, including string containment, equality, emptiness, and numeric comparisons.

Methods

• init(self)

  Initializes the parameter object with empty conditions and default supported operators.

• check(self)

  Validates the parameter data:

  • Ensures conditions is not empty.

  • Each condition must have a non-empty "to" field (destination components).

  • Ensures end_cpn_ids is not empty (providing a fallback route).

  Raises ValueError if validation fails.

• get_input_form(self) -> dict[str, dict]

  Returns a dictionary describing the expected input form for the component. Currently, it specifies one input named "urls" of type "line". This is likely used by the UI or API to generate input forms.

Usage Example

param = SwitchParam()
param.conditions = [
    {
        "logical_operator": "and",
        "items": [
            {"cpn_id": "categorize:0", "operator": "contains", "value": "example"},
            {"cpn_id": "categorize:1", "operator": "=", "value": "test"}
        ],
        "to": ["next_component_id"]
    }
]
param.end_cpn_ids = ["default_component_id"]
param.check()  # Raises error if invalid

Class: Switch

class Switch(ComponentBase, ABC):

Purpose

Implements the Switch component logic for evaluating multiple conditional branches and determining the next step(s) in the workflow. It inherits from ComponentBase and is abstract (ABC).

Class Attributes

• component_name (str): Static name of the component, "Switch".

Methods

• _invoke(self, **kwargs)

  The main execution method triggered by the workflow engine when this component runs.

  Functionality:

  • Iterates over all configured conditions.

  • For each condition, evaluates its items:

    • Retrieves the current value of the referenced component variable (cpn_id).

    • Uses process_operator to evaluate the operator against the value.

    • Depending on the logical operator (and / or), determines if the condition matches.

  • If a condition matches:

    • Sets the output "next" to the names of the next components.

    • Sets the internal output "_next" to the list of next component IDs.

    • Returns immediately.

  • If no condition matches:

    • Sets the output "next" and "_next" to the fallback end_cpn_ids.

  The method is decorated with a timeout (default 3 seconds or environment variable COMPONENT_EXEC_TIMEOUT).

• process_operator(self, input: Any, operator: str, value: Any) -> bool

  Evaluates a single condition operator between input and value. Supports:

  • String operators: contains, not contains, start with, end with, empty, not empty

  • Equality: =, ≠

  • Numeric comparisons: >, <, ≥, ≤ (tries numeric comparison first, falls back to default comparison)

  Raises ValueError if the operator is unsupported.

• thoughts(self) -> str

  Returns a fixed string expressing internal state/thought process. Possibly used for debugging, logging, or UI feedback.

Parameters

• _param (inherited from ComponentBase): An instance of SwitchParam containing the component configuration.

• _canvas (inherited from ComponentBase): Provides access to other components' variables and names via:

  • get_variable_value(cpn_id: str) -> Any

  • get_component_name(cpn_id: str) -> str

Usage Example

switch_component = Switch()
switch_component._param = param  # instance of SwitchParam configured earlier
# _canvas is set by the workflow engine/environment
switch_component._invoke()
# Check outputs via switch_component.get_output("next") etc.

Important Implementation Details

• Logical Operators: The Switch supports two main logical operators for conditions: "and" and "or". For "and", all items within a condition must be true. For "or", any item being true satisfies the condition.

• Operator Evaluation: The process_operator method gracefully handles both string and numeric comparisons. It attempts to cast values to floats for numeric comparisons but falls back to default Python comparison if that fails.

• Timeout Decorator: The invocation method is decorated with a timeout, ensuring the component does not block the workflow indefinitely.

• Dynamic Next Steps: The Switch sets two outputs — "next" (human-readable component names) and "_next" (component IDs) — allowing the workflow engine to route execution accordingly.

• Fallback Route: If none of the configured conditions match, the Switch routes to the components listed in end_cpn_ids.

Interaction with Other System Components

• ComponentBase and ComponentParamBase: The Switch and SwitchParam classes extend base classes that provide common component infrastructure such as input/output management and parameter handling.

• Canvas Object: The Switch uses the _canvas attribute (likely injected by the workflow engine) to query values and names of other components. This enables it to evaluate conditions based on live data from other components in the workflow.

• Timeout Utility: The timeout decorator from api.utils.api_utils is used to limit execution duration, integrating with system-wide timeout policies.

• Agent Component System: The Switch is part of the agent.component module hierarchy, indicating it is a fundamental building block in the agent's workflow execution engine.

Visual Diagram

classDiagram
    class SwitchParam {
        -conditions: list
        -end_cpn_ids: list
        -operators: list[str]
        +__init__()
        +check()
        +get_input_form() dict
    }

    class Switch {
        -component_name: str = "Switch"
        +_invoke(**kwargs)
        +process_operator(input: Any, operator: str, value: Any) bool
        +thoughts() str
    }

    SwitchParam <|-- Switch : uses/configures
    Switch ..> ComponentBase : inherits
    SwitchParam ..> ComponentParamBase : inherits

Summary

The switch.py file implements a configurable conditional branching component for the InfiniFlow system. It allows workflows to dynamically select the next execution path based on composite conditions evaluated against other components' output values. The component is robust, supporting various string and numeric comparisons, logical operators, and fallback routes. It integrates closely with the workflow engine's canvas to fetch live data and route execution accordingly, making it a pivotal element for complex workflow orchestration.

If you need usage examples for integration in the overall workflow or further explanation of related components, please ask!