> ## Documentation Index
> Fetch the complete documentation index at: https://docs.reducto.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Pipeline

> Run pre-configured pipelines using the Python SDK

The `pipeline.run()` method executes a pipeline you've created in [Reducto Studio](https://studio.reducto.ai). Pipelines bundle parse, extract, split, and edit operations into a single reusable workflow. You configure the pipeline in Studio, deploy it to get a `pipeline_id`, then call it from your code.

***

## Basic Usage

```python theme={null}
from pathlib import Path
from reducto import Reducto

client = Reducto()

# Upload document
upload = client.upload(file=Path("invoice.pdf"))

# Run a deployed pipeline
result = client.pipeline.run(
    input=upload.file_id,
    pipeline_id="your_pipeline_id"  # From Studio deployment
)

# Access results based on pipeline type
if result.result.extract:
    print(result.result.extract)
elif result.result.parse:
    for chunk in result.result.parse.result.chunks:
        print(chunk.content)
```

***

## Method Signature

```python theme={null}
def pipeline.run(
    input: str | list[str],
    pipeline_id: str,
    settings: dict | None = None
) -> PipelineResponse
```

### Parameters

| Parameter     | Type               | Required | Description                                                                      |
| ------------- | ------------------ | -------- | -------------------------------------------------------------------------------- |
| `input`       | `str \| list[str]` | Yes      | File ID (`reducto://...`), URL, or list of file IDs for multi-document pipelines |
| `pipeline_id` | `str`              | Yes      | The pipeline ID from Studio deployment                                           |
| `settings`    | `dict \| None`     | No       | Settings that override pipeline defaults                                         |

***

## Creating Pipelines

Pipelines are created and configured in [Reducto Studio](https://studio.reducto.ai):

1. Go to Studio and create a new pipeline
2. Add and configure steps (Parse, Extract, Split, Edit)
3. Test with sample documents
4. Click **Deploy** and select **Pipeline**
5. Copy the generated `pipeline_id`

See [Pipeline Basics](/workflows/pipeline-basics) for detailed instructions.

***

## Response Structure

The response structure depends on what steps your pipeline includes:

```python theme={null}
result = client.pipeline.run(
    input=upload.file_id,
    pipeline_id="your_pipeline_id"
)

# Common fields
print(result.job_id)           # str: Job identifier
print(result.usage.num_pages)  # int: Pages processed
print(result.usage.credits)    # float: Credits used

# Results - populated based on pipeline type
print(result.result.parse)     # Parse results (if pipeline includes Parse)
print(result.result.extract)   # Extract results (if pipeline includes Extract)
print(result.result.split)     # Split results (if pipeline includes Split)
print(result.result.edit)      # Edit results (if pipeline includes Edit)
```

### Parse Results

```python theme={null}
if result.result.parse:
    parse_result = result.result.parse
    for chunk in parse_result.result.chunks:
        print(chunk.content)
```

### Extract Results

For Parse + Extract pipelines, extract is a single object:

```python theme={null}
if result.result.extract and not isinstance(result.result.extract, list):
    print(result.result.extract.result)
```

For Parse + Split + Extract pipelines, extract is a list (one per section):

```python theme={null}
if result.result.extract and isinstance(result.result.extract, list):
    for section in result.result.extract:
        print(f"{section.split_name}: {section.result}")
```

***

## Multi-Document Pipelines

Pass multiple file IDs to process related documents together:

```python theme={null}
upload1 = client.upload(file=Path("doc1.pdf"))
upload2 = client.upload(file=Path("doc2.pdf"))

result = client.pipeline.run(
    input=[upload1.file_id, upload2.file_id],
    pipeline_id="your_pipeline_id"
)
```

Multi-document pipelines parse each document separately but run extraction across combined content.

<Warning>
  Multi-document pipelines require a pipeline without Split. Split is not supported with multiple input documents.
</Warning>

***

## Error Handling

```python theme={null}
from reducto import Reducto
import reducto

try:
    result = client.pipeline.run(
        input=upload.file_id,
        pipeline_id="your_pipeline_id"
    )
except reducto.APIConnectionError as e:
    print(f"Connection failed: {e}")
except reducto.APIStatusError as e:
    print(f"Pipeline failed: {e.status_code} - {e.response}")
```

***

## Complete Example

```python theme={null}
from pathlib import Path
from reducto import Reducto

client = Reducto()

# Upload
upload = client.upload(file=Path("fidelity-example.pdf"))

# Run pipeline
result = client.pipeline.run(
    input=upload.file_id,
    pipeline_id="k9798h9mwt0wmq5qz5e45qxbfx7yj4bq"  # Example ID
)

print(f"Job ID: {result.job_id}")
print(f"Credits used: {result.usage.credits}")

# Handle different pipeline types
if result.result.extract:
    if isinstance(result.result.extract, list):
        # Split + Extract pipeline
        for section in result.result.extract:
            print(f"\n{section.split_name}:")
            print(section.result)
    else:
        # Parse + Extract pipeline
        print("Extracted data:", result.result.extract.result)
elif result.result.parse:
    # Parse-only pipeline
    for chunk in result.result.parse.result.chunks:
        print(chunk.content[:200])
```

***

## Best Practices

<CardGroup cols={2}>
  <Card title="Create Pipelines in Studio" icon="wand-magic-sparkles">
    Use Studio to configure and test pipelines before deploying to production.
  </Card>

  <Card title="Version Your Pipelines" icon="code-branch">
    Use version names when deploying to track configuration changes.
  </Card>
</CardGroup>

***

## Next Steps

* Learn how to [create pipelines in Studio](/workflows/pipeline-basics)
* Explore [multi-document pipelines](/workflows/multi-document-pipelines)
* Check out the [async client](/sdk/python/async) for concurrent processing