# Multi-Step Pipelines

Multi-step pipelines allow you to chain multiple processing steps together using the pipe operator (`|`). This enables you to create sophisticated processing flows by combining simple, reusable components.

## Basic Concept

The pipe operator (`|`) combines multiple steps into a single pipeline:

```python
full_pipeline = step1 | step2 | step3
```

Each step processes the output of the previous step, creating a chain of transformations.

## Simple Example

Let's create a simple pipeline with three steps: preprocessing, agent processing, and postprocessing.

### Step 1: Define Custom Components

First, create simple components for preprocessing and postprocessing:

{% code lineNumbers="true" %}

```python
from typing import Any
from gllm_core.schema import Component
from gllm_pipeline.steps import step

class InputValidatorComponent(Component):
    """Validates and cleans input text."""

    async def _run(self, message: str, **kwargs: Any) -> str:
        cleaned = " ".join(message.strip().split())
        return f"[VALIDATED] {cleaned}"

class ResponseFormatterComponent(Component):
    """Formats the agent's response."""

    async def _run(self, agent_response: str, **kwargs: Any) -> dict:
        return {
            "response": agent_response,
            "formatted": True,
            "timestamp": "2026-03-06",
        }
```

{% endcode %}

### Step 2: Create the Agent

Create a simple Digital Employee agent:

{% code lineNumbers="true" %}

```python
from digital_employee_core import (
    DigitalEmployeeIdentity,
    DigitalEmployeeJob,
)

identity = DigitalEmployeeIdentity(
    name="Multi-Step Assistant",
    email="multistep@example.com",
    job=DigitalEmployeeJob(
        title="Multi-Step Processing Assistant",
        description="An assistant that demonstrates multi-step pipelines",
        instruction="You are a helpful assistant. Provide clear and concise answers.",
    ),
)
```

{% endcode %}

### Step 3: Build the Pipeline with Pipe Operator

Now, combine all steps using the pipe operator:

{% code lineNumbers="true" %}

```python
from digital_employee_core import (
    DigitalEmployeeAgentBuilder,
    DigitalEmployeeBuilder,
    DigitalEmployeePipelineBuilder,
)
from digital_employee_core.digital_employee.state import DigitalEmployeeState

validator_step = step(
    component=InputValidatorComponent(),
    input_map={"message": "message"},
    output_state="message",
    name="InputValidator",
)

agent_step = DigitalEmployeeAgentBuilder().build_step(
    identity=identity, configs=[]
)

formatter_step = step(
    component=ResponseFormatterComponent(),
    input_map={"agent_response": "agent_response"},
    output_state="agent_response",
    name="ResponseFormatter",
)

full_pipeline = validator_step | agent_step | formatter_step
full_pipeline.state_type = DigitalEmployeeState

digital_employee = (
    DigitalEmployeeBuilder()
    .set_name(identity.name)
    .set_email(identity.email)
    .set_job(
        title=identity.job.title,
        description=identity.job.description,
        instruction=identity.job.instruction,
    )
    .set_pipeline(full_pipeline)
    .build()
)
```

{% endcode %}

### Step 4: Run the Pipeline

{% code lineNumbers="true" %}

```python
import asyncio

result = asyncio.run(
    digital_employee.invoke(message="Hello!  Can you  help   me?")
)

print(result)
```

{% endcode %}

## How It Works

1. **Input Validation Step**: The `InputValidatorComponent` cleans and validates the input message
2. **Agent Step**: The Digital Employee processes the validated input and generates a response
3. **Response Formatting Step**: The `ResponseFormatterComponent` adds formatting to the agent's response

The pipe operator (`|`) ensures that each step receives the output of the previous step as its input.

## Important: Setting `state_type`

When using the pipe operator, the resulting `Pipeline` object has a default `state_type`. For the pipeline to work with DigitalEmployee, you must set it to `DigitalEmployeeState`:

```python
full_pipeline = step1 | step2 | step3
full_pipeline.state_type = DigitalEmployeeState
```

This is necessary because DigitalEmployee creates a `DigitalEmployeeState` object and passes it to the pipeline. If the pipeline's `state_type` doesn't match, the pipeline won't recognize the state keys.

## Related Documentation

* [Extend](https://gdplabs.gitbook.io/catapa/developer-documentation/digital-employee-de-pipeline/guides/advanced-examples/extend) - More advanced examples of extending Digital Employees
* [Digital Employee Architecture](https://gdplabs.gitbook.io/catapa/developer-documentation/digital-employee-de-pipeline/guides/digital-employee-architecture) - Understanding the architecture
* [Getting Started Examples](https://gdplabs.gitbook.io/catapa/developer-documentation/digital-employee-de-pipeline/getting-started/examples) - Basic examples