Subgraph step

A pipeline step that executes another pipeline as a subgraph.

Author

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

References

[1] https://langchain-ai.github.io/langgraph/

SubgraphStep(name, subgraph, input_state_map=None, output_state_map=None, runtime_config_map=None, fixed_args=None)

Bases: BasePipelineStep, HasInputsMixin

A pipeline step that executes another pipeline as a subgraph.

This step allows for encapsulation and reuse of pipeline logic by treating another pipeline as a step. The subgraph can have its own state schema, and this step handles the mapping between the parent and subgraph states.

Attributes:

Name	Type	Description
name	str	A unique identifier for this pipeline step.
subgraph	Pipeline	The pipeline to be executed as a subgraph.
input_state_map	dict[str, str]	Mapping of subgraph input keys to parent pipeline state keys.
output_state_map	dict[str, str]	Mapping of parent pipeline state keys to subgraph output keys.
runtime_config_map	dict[str, str] | None	Mapping of subgraph input keys to runtime configuration keys.
fixed_args	dict[str, Any] | None	Fixed arguments to be passed to the subgraph.

Initializes a new SubgraphStep.

Parameters:

Name	Type	Description	Default
name	str	A unique identifier for this pipeline step.	required
subgraph	Pipeline	The pipeline to be executed as a subgraph.	required
input_state_map	dict[str, str]	Mapping of subgraph input keys to parent pipeline state keys.	None
output_state_map	dict[str, str]	Mapping of parent pipeline state keys to subgraph output keys.	None
runtime_config_map	dict[str, str] | None	Mapping of subgraph input keys to runtime configuration keys. Defaults to None, in which case an empty dictionary is used.	None
fixed_args	dict[str, Any] | None	Fixed arguments to be passed to the subgraph. Defaults to None, in which case an empty dictionary is used.	None

execute(state, config) async

Executes the subgraph and processes its output.

This method prepares data, executes the subgraph, and formats the output for integration into the parent pipeline state. It only uses keys that are actually present in the state, ignoring missing keys to prevent errors.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	The current state of the pipeline, containing all data.	required
config	RunnableConfig	Runtime configuration for this step's execution.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: The update to the pipeline state after this step's operation. This includes new or modified data produced by the subgraph, not the entire state. If a requested output key is not present in the subgraph result, its value will be None.

Raises:

Type	Description
RuntimeError	If an error occurs during subgraph execution, with details about which step caused the error.

get_mermaid_diagram()

Create a Mermaid diagram representation of the subgraph.

Returns:

Name	Type	Description
str	str	The complete Mermaid diagram representation of the subgraph.