Input Mapping

Input mapping in gllm_pipeline allows you to control how data flows from the pipeline state and runtime configuration into your components.

Input sources

When you create some steps, you need to specify how data flows from the pipeline into your component.

There are three main sources of data:

1. Pipeline State: Data that flows through the pipeline from step to step. This is mutable.

2. Runtime Configuration: Settings that can only be changed at runtime. This is immutable.

3. Fixed Values: Constants that never change. This is also immutable.

Using input_map

New in gllm-pipeline v0.4.16 !

If you are using gllm-pipeline v0.4.15 and below, refer to The Three Input Mapping Parameters. Note that these parameters are slated for removal in v0.5.0.

The input_map is an argument that can be passed into some steps to define where they get their inputs from. It is the dictionary, where:

• Keys are component argument names.

• Values can be:

  • Strings: State/config keys (tries state first, then config).

  • Val objects: Fixed/literal value.

Let's see how input_map works when you actually invoke a pipeline. This example shows the complete flow from pipeline creation to execution:

from gllm_pipeline.steps import ComponentStep
from gllm_pipeline.types import Val
from gllm_core.schema.component import Component
from typing import Any

class AdvancedProcessor(Component):
    """A component that processes text with various parameters."""
    
    async def _run(self, **kwargs: Any) -> Any:
        text = kwargs["text"]
        max_length = kwargs["max_length"]
        prefix = kwargs["prefix"]
        suffix = kwargs["suffix"]
        
        processed = f"{prefix}{text[:max_length]}{suffix}"
        return {"processed_text": processed, "original_length": len(text)}

# Create step with mixed input_map
processor_step = step(
    AdvancedProcessor(),
    input_map={
        "text": "user_input",           # From state
        "max_length": "config_max_len", # From config
        "prefix": Val(">>> "),          # Fixed value
        "suffix": Val(" <<<")           # Fixed value
    },
    output_state=["processed_text", "original_length"]  # Multiple outputs
)

# Create pipeline
pipeline = Pipeline(steps=[processor_step])

# Invoke with state and config
result = await pipeline.invoke(
    initial_state={"user_input": "This is a very long text that will be truncated"},
    config={"configurable": {"config_max_len": 20}}
)

print(result)
# Output: {
#   "user_input": "This is a very long text that will be truncated",
#   "processed_text": ">>> This is a very long <<<",
#   "original_length": 50
# }

For better ergonomics, input_map also accepts a list format:

# List form with mixed mappings
echo_step = ComponentStep(
    name="echo_step",
    component=Echo(),
    input_map=[
        "x",                           # Identity mapping: {"x": "x"}
        {"y": "user_input"},           # State mapping: {"y": "user_input"}
        {"z": Val("Fixed")}            # Fixed value: {"z": Val("Fixed")}
    ],
    output_state="echoed_value"

The Three Input Mapping Parameters

These parameters are now deprecated and will be removed in gllm-pipeline v0.5.0. Consider using input_map instead.

1. input_state_map

Maps component input arguments to pipeline state keys. The component receives values from the pipeline state.

from gllm_pipeline.steps import ComponentStep
from gllm_core.schema.component import Component
from typing import Any

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

    def identity(self, x: Any) -> Any:
        """Return the input 'x' unchanged."""
        return x

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

# Using input_state_map
echo_step = ComponentStep(
    name="echo_step",
    component=Echo(),
    input_state_map={"x": "user_input"},  # Maps component arg 'x' to state key 'user_input'
    output_state="echoed_value"
)

# When executed with state={"user_input": "Hello World"}
# The Echo component receives: x="Hello World"

2. runtime_config_map

Maps component input arguments to runtime configuration keys. The component receives values from the runtime configuration.

# Using runtime_config_map
echo_step = ComponentStep(
    name="echo_step",
    component=Echo(),
    runtime_config_map={"x": "config_value"},  # Maps component arg 'x' to config key 'config_value'
    output_state="echoed_value"
)

# When executed with config={"configurable": {"config_value": "From Config"}}
# The Echo component receives: x="From Config"

3. fixed_args

Provides fixed values directly to component arguments. These values don't come from state or config.

# Using fixed_args
echo_step = ComponentStep(
    name="echo_step",
    component=Echo(),
    fixed_args={"x": "Fixed Value"},  # Always passes "Fixed Value" to component arg 'x'
    output_state="echoed_value"
)

# The Echo component always receives: x="Fixed Value" 

Combining the Three Parameters

You could combine all three approaches, with fixed_args taking precedence over runtime_config_map, which takes precedence over input_state_map.

# Combining all three approaches
echo_step = ComponentStep(
    name="echo_step",
    component=Echo(),
    input_state_map={"x": "user_input"},           
    runtime_config_map={"y": "config_value"},       
    fixed_args={"z": "Fixed Value"},              
    output_state="echoed_value"
)