# Steps

[**`gllm-pipeline`**](https://github.com/GDP-ADMIN/gl-sdk/tree/main/libs/gllm-pipeline/gllm_pipeline/steps) | **Tutorial**: [steps](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/tutorials/orchestration/steps "mention")| **Use Case**: [build-end-to-end-rag-pipeline](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/how-to-guides/build-end-to-end-rag-pipeline "mention")| [API Reference](https://api.python.docs.gdplabs.id/gen-ai/library/gllm_pipeline/api/steps.html)

## 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](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/tutorials/pipeline#the-pipe-operator "mention").

Learn in greater detail about input maps in [input-mapping](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/tutorials/orchestration/steps/input-mapping "mention").

<details>

<summary>Prerequisites</summary>

This example specifically requires completion of all setup steps listed on the [prerequisites](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/gen-ai-sdk/prerequisites "mention") page.

You should be familiar with these concepts:

1. [state](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/tutorials/orchestration/state "mention")

</details>

## Installation

{% tabs %}
{% tab title="Linux, macOS, or Windows WSL" %}

```bash
# 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"
```

{% endtab %}

{% tab title="Windows Powershell" %}

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

{% endtab %}

{% tab title="Windows Command Prompt" %}

```bash
# you can use a Conda environment
FOR /F "tokens=*" %T IN ('gcloud auth print-access-token') DO pip install --extra-index-url "gllm-pipeline"
```

{% endtab %}
{% endtabs %}

## 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.

{% code title="echo.py" %}

```python
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"])
```

{% endcode %}

### step

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

```python
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",
)
```

{% hint style="info" %}
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](https://gdplabs.gitbook.io/sdk/~/revisions/w6A7tUKJGDYFXuci5HcW/tutorials/orchestration/steps/input-mapping "mention").
{% endhint %}

### transform

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

```python
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.

```python
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.

```python
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` .

```python
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.

```python
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](#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.

```python
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.

```python
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.

```python
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](#toggle).

```python
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](#guard) step.

```python
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.

```python
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.

```python
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.

```python
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"},
)

```
