Guard step

A guard step that can terminate pipeline execution based on a condition.

Author

Dimitrij Ray (dimitrij.ray@gdplabs.id) Kadek Denaya (kadek.d.r.diana@gdplabs.id)

References

NONE

`GuardStep(name, condition, success_branch, failure_branch=None, input_state_map=None, output_state=None, runtime_config_map=None, fixed_args=None, input_map=None, retry_config=None, error_handler=None, cache_store=None, cache_config=None)`

Bases: ConditionalStep

A conditional step that can terminate pipeline execution if a condition is not met.

This step evaluates a condition and either: 1. Continues execution through the success_branch if the condition is True 2. Executes the failure_branch and terminates if the condition is False

Example

pipeline = (
    step_a
    | GuardStep(
        name="auth_check",
        condition=lambda x: x["is_authenticated"],
        success_branch=step_b,
        failure_branch=error_handling_step,
    )
    | step_c
)

Attributes:

Name	Type	Description
`name`	`str`	A unique identifier for this pipeline step.
`condition`	`ConditionType`	The condition to evaluate.
`input_map`	`dict[str, str \| Any] \| None`	Unified input map.
`success_branch`	`BasePipelineStep \| list[BasePipelineStep]`	Steps to execute if condition is True.
`failure_branch`	`BasePipelineStep \| list[BasePipelineStep] \| None`	Steps to execute if condition is False. If None, pipeline terminates immediately on False condition.
`retry_policy`	`RetryPolicy \| None`	Configuration for retry behavior using LangGraph's RetryPolicy.

Initializes a new GuardStep.

Parameters:

Name	Type	Description	Default
`name`	`str`	A unique identifier for this pipeline step.	required
`condition`	`ConditionType`	The condition to evaluate.	required
`success_branch`	`BasePipelineStep \| list[BasePipelineStep]`	Steps to execute if condition is True.	required
`failure_branch`	`BasePipelineStep \| list[BasePipelineStep] \| None`	Steps to execute if condition is False. If None, pipeline terminates immediately. Defaults to None.	`None`
`input_state_map`	`dict[str, str] \| None`	Mapping of condition input arguments to pipeline state keys. Defaults to None.	`None`
`output_state`	`str \| None`	Key to store the condition result in the pipeline state. Defaults to None.	`None`
`runtime_config_map`	`dict[str, str] \| None`	Mapping of condition input arguments to runtime configuration keys. Defaults to None.	`None`
`fixed_args`	`dict[str, Any] \| None`	Fixed arguments to be passed to the condition. Defaults to None.	`None`
`input_map`	`InputMapSpec \| None`	Unified input map. Can be a dict (arg -> str\|Val) or a list with elements: 1. str for identity mapping 2. dict[str, str] for state/config mapping 3. dict[str, Val] for fixed args. Defaults to None.	`None`
`error_handler`	`BaseStepErrorHandler \| None`	Strategy to handle errors during execution. Defaults to None, in which case the RaiseStepErrorHandler is used.	`None`
`retry_config`	`RetryConfig \| None`	Configuration for retry behavior using GLLM Core's RetryConfig. Defaults to None, in which case no retry config is applied.	`None`
`cache_store`	`'BaseCache' \| None`	Cache store to be used for caching. Defaults to None, in which case no cache store is used.	`None`
`cache_config`	`dict[str, Any] \| None`	Cache configuration to be used for caching. Defaults to None, in which case no cache configuration is used.	`None`