Parallel step

A pipeline step that executes multiple branches in parallel.

ParallelStep(name, branches, input_states=None, squash=True, input_map=None, retry_config=None, error_handler=None, cache=None, copy_keys=None, max_concurrency=None)

Bases: BranchingStep, HasInputsMixin

A pipeline step that executes multiple branches in parallel.

This step wraps multiple branches and executes them concurrently, then merges their results. Each branch can be either a single step or a list of steps to be executed sequentially.

The step supports two execution modes controlled by the squash parameter: 1. Squashed (default): Uses asyncio.gather() to run branches in parallel within a single LangGraph node. Use for: a. Better raw performance b. Simpler implementation c. Less overhead d. Less transparent for debugging and tracing 2. Expanded (squash=False): Creates a native LangGraph structure with multiple parallel paths. Use for: a. More native LangGraph integration b. More transparent for debugging and tracing

For memory optimization, you can specify input_states to pass only specific keys to branches. This is especially useful when the state is large but branches only need specific parts of it. If input_states is None (default), all state keys will be passed.

The copy_keys parameter controls state isolation between branches. By default, copied keys are inferred independently from each branch's state metadata.

Attributes:

Name	Type	Description
name	str	A unique identifier for this pipeline step.
branches	dict[str, PipelineSteps]	The branches to execute in parallel.
input_map	dict[str, str | Val] | None	Unified input map.
squash	bool	Whether to squash execution into a single node. 1. If True, uses asyncio.gather() to run branches in parallel. This will create a single node. 2. If False, uses native LangGraph structures for parallelism. This will create multiple nodes.
retry_policy	RetryPolicy | None	Configuration for retry behavior using LangGraph's RetryPolicy.
max_concurrency	int | None	Maximum number of concurrent branch executions in squashed mode. None means unbounded.

Initialize a new ParallelStep.

Parameters:

Name	Type	Description	Default
name	str	A unique identifier for this pipeline step.	required
branches	list | dict[str, PipelineSteps]	The branches to execute in parallel. Can be either: List format: Each branch can be: 1. A single step 2. A list of steps to execute sequentially Example: [step1, [step2, step3], step4] Dict format: Keys are branch names, values can be: 1. A single step 2. A list of steps to execute sequentially Example: {"analysis": step1, "validation": [step2, step3], "cleanup": step4} Enables more intuitive step exclusion using branch names.	required
input_states	list[str] | None	Keys from the state to pass to branches. If None, all state keys will be passed. Defaults to None.	None
squash	bool	Whether to squash execution into a single node. 1. If True, uses asyncio.gather() to run branches in parallel. This will create a single node. 2. If False, uses native LangGraph structures for parallelism. This will create multiple nodes. Defaults to True.	True
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
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
error_handler	BaseStepErrorHandler | None	Strategy to handle errors during execution. Defaults to None, in which case the RaiseStepErrorHandler is used.	None
cache	CacheConfig | None	Configuration for the cache store including the cache store instance and its settings. Defaults to None, in which case no caching will be used.	None
copy_keys	Iterable[str] | None	State keys to deep-copy in every branch. Use an empty iterable only for read-only state. Defaults to None, which infers copy keys per branch.	None
max_concurrency	int | None	Maximum number of concurrent branch executions in squashed mode. Has no effect when squash=False. Defaults to None, in which case there is no concurrency limit.	None

Raises:

Type	Description
ValueError	If max_concurrency is less than 1.
TypeError	If branches is neither a list nor a dict.

output_state_keys property

ParallelStep writes the union of all its branches' output keys.

Returns:

Type	Description
set[str] | None	set[str] | None: Union of branch-step output keys, or None if any step is opaque.

add_to_graph(graph, previous_endpoints, retry_policy=None)

Handle both squashed and expanded modes.

For squashed mode: add the parallel step as a single node. For expanded mode: add the parallel step as a single node and add children to graph.

Parameters:

Name	Type	Description	Default
graph	StateGraph	The graph to add this step to.	required
previous_endpoints	list[str]	Endpoints from previous steps to connect to.	required
retry_policy	RetryPolicy | None	Retry policy to propagate to child steps. Defaults to None, in which case the retry policy of the step is used.	None

Returns:

Type	Description
list[str]	list[str]: Exit points after adding all child steps.

execute(state, runtime, config=None) async

Execute all branches in parallel and merge their results.

This method is only used for the squashed approach. For the expanded approach, the execution is handled by the graph structure.

State isolation is controlled by the copy_keys parameter:

Parameters:

Name	Type	Description	Default
state	PipelineState	The current state of the pipeline.	required
runtime	Runtime	Runtime information for this step's execution.	required
config	RunnableConfig | None	The runnable configuration. Defaults to None.	None

Returns:

Type	Description
dict[str, Any] | None	dict[str, Any] | None: The merged results from all parallel branches, or None if no updates were produced.

Raises:

Type	Description
CancelledError	If execution is cancelled, preserved with added context.
BaseInvokerError	If an error occurs during LM invocation.
RuntimeError	Wraps all other exceptions with context information.
TimeoutError	If execution times out, preserved with added context.
ValidationError	If input validation fails.