Pipeline

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

The Pipeline is the core orchestration component that sequences and manages the execution of the components in our SDK.

Prerequisites

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

You should be familiar with these concepts:

1. Basic Conceptsof orchestration components

2. State

3. Steps

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

Quickstart

We will create a simple general Pipeline to illustrate the general workflow around building Pipelines.

1

Import the Pipeline and the steps

from typing import TypedDict  # Optional, only for custom states
from gllm_pipeline.pipeline.pipeline import Pipeline
from gllm_pipeline.steps._func import step, transform, bundle, log, subgraph

2

Define your state

We will use a simplified state for ths quickstart. Alternatively, if you are dealing with an RAG pipeline, you can use the default RAGState instead.

class MiniState(TypedDict):
    text: str
    text_upper: str
    text_len: int
    summary: dict  # summary bundle

3

Define your steps

Here we use simple transform and bundle steps to illustrate how the Pipeline works. You can always use the other Steps, or follow the How-to guide for a comprehensive guide.

def to_upper(data: dict) -> str:
    return data["text"].upper()

def count_chars(data: dict) -> int:
    return len(data["text_upper"])

pipe = Pipeline(
    steps=[
        transform(to_upper, input_states=["text"], output_state="text_upper"),
        transform(count_chars, input_states=["text_upper"], output_state="text_len"),
        bundle(["text", "text_upper", "text_len"], output_state="summary"),
    ],
    state_type=MiniState,
)

4

Invoke the pipeline

Our pipeline is asynchronous by default. Therefore, to invoke it, you must use asyncio.run.

import asyncio

initial: MiniState = {
    "text": "hello world",
    "text_upper": "",
    "text_len": 0,
    "summary": {},
}
final = asyncio.run(pipe.invoke(initial))

print(final)

After invoking the pipeline, you should get an output similar to this:

{'text': 'hello world', 'text_upper': 'HELLO WORLD', 'text_len': 11,
 'summary': {'text': 'hello world', 'text_upper': 'HELLO WORLD', 'text_len': 11}}

That's it! You have created your first Pipeline! All future Pipelines that you will ever create will follow the above general steps.

The Pipe Operator

You can also utilize the pipe (|) operator to compose your Pipeline.

class MiniState2(TypedDict):
    text: str
    text_upper: str
    text_len: int

def to_upper2(data: dict) -> str:
    return data["text"].upper()

def count_chars2(data: dict) -> int:
    return len(data["text_upper"])

pipe2 = (
    transform(to_upper2, ["text"], "text_upper")
    | transform(count_chars2, ["text_upper"], "text_len")
    | log("Upper2: {text_upper} (len={text_len})")
)
pipe2.state_type = MiniState2  # important: set your TypedDict state

initial2: MiniState2 = {"text": "pipeline!", "text_upper": "", "text_len": 0}
final2 = await pipe2.invoke(initial2)
print(final2)

When composing using | , the Pipeline's state will be RAGState. Make sure that you use the state_type setter to specify the correct state type before invoking.

Appending a Step

You can use the | operator to append a step to a Pipeline.

p1 = Pipeline(
    steps=[transform(to_upper, ["text"], "text_upper")],
    state_type=TextState,
)

# Append step at the end
p1_plus = p1 | transform(count_len, ["text_upper"], "text_len")

final = await p1_plus.invoke({"text": "hello", "text_upper": "", "text_len": 0})
print(final)  # {'text': 'hello', 'text_upper': 'HELLO', 'text_len': 5}

Merge Two Pipelines

You can also use the | operator to merge two pipelines of the same State schema.

p_left = Pipeline(
    steps=[transform(to_upper, ["text"], "text_upper")],
    state_type=TextState,
)
p_right = Pipeline(
    steps=[transform(count_len, ["text_upper"], "text_len"), log("len={text_len}")],
    state_type=TextState,
)

combined = p_left | p_right  # OK: same state type

final = await combined.invoke({"text": "compose", "text_upper": "", "text_len": 0})
print(final)  # {'text': 'compose', 'text_upper': 'COMPOSE', 'text_len': 7}

Merging two pipelines with different State schemata will cause a ValidationError.

Placeholder Pipelines

Finally, you can initialize a Pipeline with an empty step to use as a placeholder, e.g. to set the state_type, then use the | operator to compose the pipeline with the correct state_type.

identity = Pipeline([], state_type=TextState)
p = Pipeline([transform(to_upper, ["text"], "text_upper")], state_type=TextState)

a = p | identity
b = identity | p  # Same as p, but if constructed via step | identity, set state_type explicitly

Visualizing the Pipeline

Our Pipelines come with the get_mermaid_diagram() method, which gives you a Mermaid code. This is useful for docs and reviews.

To obtain the Mermaid diagram of a Pipeline, simply call the method.

diagram = pipe.get_mermaid_diagram()
print(diagram)

The output should look something like this:

graph TD
Transform_to_upper__abc123
Transform_count_chars__def456
Log__ghi789
START --> Transform_to_upper__abc123
Transform_to_upper__abc123 --> Transform_count_chars__def456
Transform_count_chars__def456 --> Log__ghi789
Log__ghi789 --> END

Which you could then copy and paste to any Mermaid renderer.

Runtime Configuration

Some steps support dynamic runtime configuration, which allows us to change the step's behavior at runtime.

To use these runtime configurations, during invocation, supply a dictionary using the config parameter.

pipe.invoke(state, config={"reverse": False, "uppercase": True})

For step, transform, parallel, and map_reduce , the Runtime Config gets merged with the inputs from the State, and can be used as an extra input to the Callable or Component inside these steps. For these steps, the Runtime Config is only available when defined via a runtime_config_map.

import asyncio

from typing import TypedDict
from gllm_pipeline.pipeline.pipeline import Pipeline
from gllm_pipeline.steps._func import transform

# State
class CfgState(TypedDict):
    text: str
    result: str

# Operation uses config-driven switches
def format_text(data: dict) -> str:
    s = data["text"]
    if data["reverse"]:
        s = s[::-1]
    if data["uppercase"]:
        s = s.upper()
    return s

pipe = Pipeline(
    steps=[
        transform(
            format_text,
            input_states=["text"],
            output_state="result",
            # Map operation arg names -> config keys
            runtime_config_map={"reverse": "reverse", "uppercase": "uppercase"},
        )
    ],
    state_type=CfgState,
)

# Provide all required config keys listed in runtime_config_map
state: CfgState = {"text": "Hello", "result": ""}
out = asyncio.run(pipe.invoke(state, config={"reverse": False, "uppercase": True}))
print(out["result"])  # HELLO

out2 = asyncio.run(pipe.invoke({"text": "Hello", "result": ""}, config={"reverse": True, "uppercase": True}))
print(out2["result"])  # OLLEH

For conditionals if_else, switch, toggle, and guard, the Runtime Config is always available and can be accessed using its original keys when using a Callable.

from gllm_pipeline.steps._func import if_else, transform

def is_positive(data: dict) -> bool:
    return data["value"] > 0

# Branches write a message
pos = transform(lambda _data: "value is positive", input_states=[], output_state="msg")
neg = transform(lambda _data: "value is non-positive", input_states=[], output_state="msg")

positive_step = if_else(
    condition=is_positive,
    if_branch=pos,
    else_branch=neg,
    output_state="decision",
)

However, when using a Component, the runtime_config_map must be provided.

import asyncio

from gllm_core.schema import Component
from gllm_pipeline.steps._func import if_else, transform

# Minimal Component that returns "true"/"false"
class GreaterThan(Component):
    def decide(self, value: int, threshold: int) -> str:
        """Return 'true' if value > threshold else 'false'."""
        return "true" if value > threshold else "false"

    async def _run(self, **kwargs: Any) -> str:
        """Implements the core logic; called by Component.run(...)."""
        value: int = kwargs["value"]
        threshold: int = kwargs["threshold"]
        return self.decide(value, threshold)


# Branches write a message
higher = transform(lambda _data: "value > threshold", input_states=[], output_state="msg")
lower_eq = transform(lambda _data: "value <= threshold", input_states=[], output_state="msg")

cond_comp = GreaterThan()

greater_than_step = if_else(
    condition=cond_comp,
    if_branch=higher,
    else_branch=lower_eq,
    input_state_map={"value": "value"},           # map state["value"] -> arg "value"
    runtime_config_map={"threshold": "threshold"},# map config["threshold"] -> arg "threshold"
    output_state="decision",
)

Using the Debug State

Our Pipeline comes with a utility to provide a trace of the Pipeline execution. To do so, pass config={"debug_state": True} . The trace is available as __state_logs__ in the final output.

final_dbg = await pipe.invoke(
    {"text": "debug me", "text_upper": "", "text_len": 0},
    config={"debug_state": True},
)
print(final_dbg["__state_logs__"])

Using a Pipeline as a Subgraph

To use a Pipeline as a Subgraph, one can wrap the Pipeline inside a subgraph step and map the input states and configs as necessary.

# Child pipeline with its own state
class ChildState(TypedDict):
    text: str
    text_upper: str

def to_upper_child(data: dict) -> str:
    return data["text"].upper()

child = Pipeline(
    steps=[transform(to_upper_child, ["text"], "text_upper")],
    state_type=ChildState,
    name="child_pipeline",
)

# Parent pipeline with a different state
class ParentState(TypedDict):
    input_text: str
    result_upper: str
    text_len: int

def count_len_parent(data: dict) -> int:
    return len(data["result_upper"])

parent = Pipeline(
    steps=[
        subgraph(
            child,
            input_state_map={"text": "input_text"},          # parent.input_text -> child.text
            output_state_map={"result_upper": "text_upper"}, # child.text_upper -> parent.result_upper
        ),
        transform(count_len_parent, ["result_upper"], "text_len"),
        log("Parent upper='{result_upper}' (len={text_len})"),
    ],
    state_type=ParentState,
)

initial_parent: ParentState = {"input_text": "SubGraph Rocks!", "result_upper": "", "text_len": 0}
final_parent = await parent.invoke(initial_parent)
print(final_parent)

Using the Leftshift (<<) Operator

Alternatively, you can use the leftshift operator (<<) to embed a Pipeline as a subgraph in another Pipeline. Subgraphs created this way will have overlapping State keys automatically mapped.

class Parent2State(TypedDict):
    text: str          # overlaps with ChildState.text
    text_upper: str    # overlaps with ChildState.text_upper
    note: str

# Reuse 'child' from above (ChildState)
parent2 = Pipeline(
    steps=[log("Before: {text}")],
    state_type=Parent2State,
    name="parent2",
)

# Include child as a subgraph within parent2
combined = parent2 << child   # auto-maps overlapping keys

initial_p2: Parent2State = {"text": "Auto-map!", "text_upper": "", "note": ""}
final_p2 = await combined.invoke(initial_p2)
print(final_p2["text"], final_p2["text_upper"])