Executing Pipelines

Once your pipes are defined in .plx files, you can execute them in multiple ways.

The Simplest Approach: Run a Bundle File

The easiest way to execute a pipeline is to point directly to your .plx bundle file. No library configuration needed.

Using the CLI

# Run the bundle's main_pipe
pipelex run path/to/my_bundle.plx

# Run a specific pipe from the bundle
pipelex run path/to/my_bundle.plx --pipe my_specific_pipe

# Run with inputs
pipelex run path/to/my_bundle.plx --inputs inputs.json

Preparing Inputs

You can generate an input template with pipelex build inputs path/to/my_bundle.plx, which creates a results/inputs.json file with the required input structure.

Using Python

from pipelex.pipelex import Pipelex
from pipelex.pipeline.execute import execute_pipeline

Pipelex.make()

# Run the bundle's main_pipe
pipe_output = await execute_pipeline(
    bundle_uri="path/to/my_bundle.plx",
    inputs={
        "my_input": {
            "concept": "Text",
            "content": "Hello world",
        },
    },
)

# Or run a specific pipe from the bundle
pipe_output = await execute_pipeline(
    bundle_uri="path/to/my_bundle.plx",
    pipe_code="my_specific_pipe",
    inputs={...},
)

How main_pipe Works

When you run a bundle without specifying a pipe_code, Pipelex executes the bundle's main_pipe (declared at the top of the .plx file). If no main_pipe is defined and no pipe_code is provided, an error is raised.

If you provide both bundle_uri and pipe_code, the explicit pipe_code takes priority over main_pipe.

See the Pipelex Bundle Specification for more about the main_pipe property.

---

Advanced: Using Library Directories

For larger projects with multiple bundles spread across directories, you can configure library directories to load all your pipes at once and reference them by code.

Understanding Libraries

When executing pipelines programmatically, Pipelex can load pipe libraries - collections of pipes loaded from one or more directories. This is useful when:

• You have multiple bundles across different directories
• You want to reference pipes by code without specifying the bundle path each time
• You're building an application that needs access to many pipes

Library Parameters

When using execute_pipeline or start_pipeline, you can control library behavior with these parameters:

• library_id: A unique identifier for the library instance. If not specified, it defaults to the pipeline_run_id (a unique ID generated for each pipeline execution).

• library_dirs: A list of directory paths to load pipe definitions from. These directories must contain both your .plx files AND any Python files defining StructuredContent classes (e.g., *_struct.py files). If not specified, Pipelex falls back to the PIPELEXPATH environment variable, then to the current working directory.

• plx_content: When provided, Pipelex will load only this PLX content into the library, bypassing directory scanning. This is useful for dynamic pipeline execution without file-based definitions.

Python Structure Classes

If your concepts use Python StructuredContent classes instead of inline structures, those Python files must be in the directories specified by library_dirs. Pipelex auto-discovers and registers these classes during library loading. Learn more about Python StructuredContent Classes.

Learn More About Libraries

For a comprehensive understanding of libraries, including their structure, uniqueness rules, lifecycle, and best practices, see Libraries.

Library Directory Configuration

For complete configuration options including the PIPELEXPATH environment variable, CLI options, and priority resolution, see Configuring Library Directories.

Running Pipes by Code

This approach loads pipe definitions from directories and executes a specific pipe by its code.

Basic usage (loads from current working directory):

from pipelex.pipelex import Pipelex
from pipelex.pipeline.execute import execute_pipeline

# First, initialize Pipelex (this loads all pipeline definitions)
Pipelex.make()

# Execute the pipeline and wait for the result
pipe_output = await execute_pipeline(
    pipe_code="description_to_tagline",
    inputs={
        "description": {
            "concept": "ProductDescription",
            "content": "EcoClean Pro is a revolutionary biodegradable cleaning solution that removes 99.9% of germs while being completely safe for children and pets. Made from plant-based ingredients.",
        },
    },
)

Loading from specific directories:

# Load pipes from specific directories
pipe_output = await execute_pipeline(
    pipe_code="description_to_tagline",
    library_dirs=["./pipelines", "./shared_pipes"],
    inputs={
        "description": {
            "concept": "ProductDescription",
            "content": "...",
        },
    },
)

Using a custom library ID:

# Use a custom library ID for managing multiple library instances
pipe_output = await execute_pipeline(
    pipe_code="description_to_tagline",
    library_id="my_marketing_library",
    library_dirs=["./marketing_pipes"],
    inputs={
        "description": {
            "concept": "ProductDescription",
            "content": "...",
        },
    },
)

Listing available pipes

Use the pipelex show pipes command to list all the pipes available in your project.

Using PLX Content Directly

You can directly pass PLX content as a string to execute_pipeline, useful for dynamic pipeline execution without file-based definitions.

from pipelex.pipelex import Pipelex
from pipelex.pipeline.execute import execute_pipeline

my_pipe_content = """
domain = "marketing"
description = "Marketing content generation domain"
main_pipe = "generate_tagline"

[concept]
ProductDescription = "A description of a product's features and benefits"
Tagline = "A catchy marketing tagline"

[pipe.generate_tagline]
type = "PipeLLM"
description = "Generate a catchy tagline for a product"
inputs = { description = "ProductDescription" }
output = "Tagline"
prompt = "
Product Description: $description

Generate a catchy tagline based on the above description. The tagline should be memorable, concise, and highlight the key benefit.
"
"""

Pipelex.make()

pipe_output = await execute_pipeline(
    plx_content=my_pipe_content,
    inputs={
        "description": {
            "concept": "ProductDescription",
            "content": "EcoClean Pro is a revolutionary biodegradable cleaning solution that removes 99.9% of germs while being completely safe for children and pets. Made from plant-based ingredients.",
        },
    },
)

Pipe Code Resolution

When using plx_content:

• If the content has a main_pipe property and you don't provide pipe_code, the main_pipe is executed
• If you provide pipe_code, it overrides main_pipe
• If there's no main_pipe and no pipe_code, an error is raised

---

Using the Pipelex API

Pipelex has a REST API for executing pipelines. See more about it here.

---

Background Execution with start_pipeline

For more complex scenarios where you need asynchronous control, use start_pipeline. This function immediately returns a pipeline_run_id and an asyncio.Task, allowing you to run pipelines in the background.

from pipelex.pipelex import Pipelex
from pipelex.pipeline.start import start_pipeline

Pipelex.make()

# Start the pipeline without waiting
pipeline_run_id, task = await start_pipeline(
    bundle_uri="path/to/my_bundle.plx",
    inputs={
        "description": {
            "concept": "ProductDescription",
            "content": "...",
        },
    },
)

print(f"Pipeline started with ID: {pipeline_run_id}")

# Do other work while the pipeline runs...

# Wait for completion when ready
pipe_output = await task