Steps

gllm-pipeline | Tutorial: Steps| Use Case: Build End-to-End RAG Pipeline| API Reference

What's a Step?

A step is a unit of work (a node) that:

1. Reads specific keys from the pipeline state (and optionally runtime config).

2. Performs an operation (component call, function, branching, etc.).

3. Writes outputs back to the state under specific keys.

You can learn more about composing steps into a pipeline using The Pipe Operator.

Learn in greater detail about input maps in Input Mapping.

Prerequisites

This example specifically requires completion of all setup steps listed on the Prerequisites page.

You should be familiar with these concepts:

1. State

Installation

Linux, macOS, or Windows WSL

# you can use a Conda environment
pip install --extra-index-url https://oauth2accesstoken:$(gcloud auth print-access-token)@glsdk.gdplabs.id/gen-ai-internal/simple/ "gllm-pipeline"

Windows Powershell

Windows Command Prompt

Basic Steps

We will first learn the basic steps that can be used to create a Pipeline.

Throughout this tutorial page, we will be using the Echo component below as a replacement for our SDK components.

echo.py

from typing import Any

from gllm_core.schema.component import Component


class Echo(Component):
    """A component that returns the value of 'x' unchanged."""

    def identity(self, x: Any) -> Any:
        """Return the input 'x' unchanged.
        
        Args:
            x (Any): Any input to be passed to the function.
        
        Returns:
            Any: The same value provided via 'x'.
        """
        return x

    async def _run(self, **kwargs: Any) -> Any:
        """Returns the value passed under 'x' unchanged.

        Args:
            **kwargs (Any): Must contain key 'x' with the value to return.

        Returns:
            Any: The same value provided via 'x'.

        Raises:
            KeyError: If 'x' is not provided.
        """
        return self.identity(kwargs["x"])

step

A step wraps and runs a Component of our SDK with mapped inputs and stores outputs under a state key.

from gllm_pipeline.steps import step

echo_step = step(
    component=Echo(),
    input_map={"x": "value"}, # Maps argument `x` to the state `value`.
    output_state="answer",
)

When a step calls a Component, use input_map to map the component’s parameter names to keys in your pipeline state. This tells the step where to read each argument.

Learn more about input maps: Input Mapping.

transform

A transform step applies a callable to selected state keys and writes the result.

from gllm_pipeline.steps import transform

def uppercase(data: dict) -> str:
    """Makes a string `UPPERCASE`
    
    Args:
        data(dict): The state. Must contain `text`.
    
    Returns:
        str: The uppercase text.
    """
    return data["text"].upper()

uppercase_step = transform(
    operation=uppercase,
    input_map=["text"],
    output_state="upper_text",
)

bundle

A bundle step collects multiple state keys into a dictionary without changes to their values.

from gllm_pipeline.steps import bundle

make_payload = bundle(
    input_states=["user", "query"],
    output_state="payload",  # state payload is now {"user": ..., "query": ...}
)

Branching

if_else

An if_else step chooses between two branches based on a condition (a callable or a Component). The condition result can be saved to state.

from gllm_pipeline.steps import if_else, log

feature_flag_step = if_else(
    condition=lambda s: s["flag"],           # truthy -> if_branch
    if_branch=log("Feature ON", is_template=False),
    else_branch=log("Feature OFF", is_template=False),
    output_state="feature_status",           # stores "true"/"false"
)

The condition could also be a Component. In this case, you need to pass an input_map .

from gllm_pipeline.steps import if_else, step
from gllm_pipeline.pipeline import Pipeline

# Branches: produce constant messages via fixed_args (Echo returns 'x' unchanged)
grant_access = step(Echo(), {}, "decision", fixed_args={"x": "Access granted"})
deny_access = step(Echo(), {}, "decision", fixed_args={"x": "Access denied"})

# Condition: a Component that returns "true" or "false"
# Echo will return the state value at key "is_adult" (must be "true" or "false")
auth_gate = if_else(
    condition=Echo(),                      # Component condition
    if_branch=grant_access,                # runs when condition returns "true"
    else_branch=deny_access,               # runs when condition returns "false"
    input_map={"x": "is_adult"},           # map state -> component input
    output_state="condition_result",       # optional: persist "true"/"false"
)

switch

A switch step selects a branch from a dict of options based on a condition output (string). Supports default branch.

from gllm_pipeline.steps import switch, log

dispatch_step = switch(
    condition=lambda s: s["command"],        # e.g., "search", "filter"
    branches={
        "search": log("Searching...", is_template=False),
        "filter": log("Filtering...", is_template=False),
    },
    default=log("Unknown command", is_template=False),
    output_state="command_type",
)

toggle

A toggle step runs its if_branch if condition is true; otherwise behaves like no_op. Condition can be callable, Component, or a string key looked up in merged state/config.

This step is useful to enable "optional" steps that are decided at runtime.

from gllm_pipeline.steps import toggle, log

feature_flag_step = toggle(
    condition="feature_enabled",             # truthy in state/config?
    if_branch=log("Feature executed", is_template=False),
    output_state="feature_status",
)

Concurrency

parallel

A parallel step runs multiple branches in parallel and merges results. Each branch can be a step or a list of steps.

from gllm_pipeline.steps import parallel, log

parallel_step = parallel(
    branches=[
        log("A", is_template=False),
        log("B", is_template=False),
    ],
)

map_reduce

A map_reduce step maps a function (or Component) over multiple items, then reduces results.

from gllm_pipeline.steps import map_reduce

sum_numbers_step = map_reduce(
    input_map={"n": "numbers"},              # numbers is a list in state
    output_state="total",
    map_func=lambda item: item["n"],         # returns each number
    reduce_func=sum,                         # sums the list of results
)

Flow Control

no_op

A no_op step is a step that does nothing. This is useful as a placeholder step, or as part of a toggle.

from gllm_pipeline.steps import no_op

skip_step = no_op()

terminate

A terminate step explicitly terminates the current Pipeline. This is useful as part of a guard step.

from gllm_pipeline.steps import terminate

stop_now_step = terminate()

guard

A guard step evaluates a condition. On success runs success_branch. On failure runs failure_branch, then terminates the current Pipeline.

from gllm_pipeline.steps import guard, log, terminate

enforce_auth_step = guard(
    condition=lambda s: s.get("is_authenticated", False),
    success_branch=log("Welcome!", is_template=False),
    output_state="auth_result",
)

Observability

log

A log step emits a message through an event emitter. Can be a plain string or a template with state placeholders.

from gllm_pipeline.steps import log

plain = log("Processing...", is_template=False)
templated = log("User: {user_id}, Query: {query}")  # pulls from state keys

Composition

subgraph

A subgraph executes another Pipeline as a step, with flexible input/output mapping.

from gllm_pipeline.steps import subgraph

# sub_pipeline is a Pipeline you built elsewhere
sub_pipeline = Pipeline(
    steps = [step_a, step_b, step_b]
)

use_subgraph = subgraph(
    subgraph=sub_pipeline,
    input_map={"query": "user_query"},
    output_state_map={"result": "subgraph_result"},
)