Pipeline (PipelineEx v0.1.1)

View Source

AI pipeline orchestration library for Elixir.

Pipeline provides a robust framework for chaining AI provider calls (Claude, Gemini) with advanced features like fault tolerance, session management, and self-improving Genesis pipelines.

Quick Start

# Load and execute a pipeline
{:ok, config} = Pipeline.load_workflow("my_pipeline.yaml")
{:ok, results} = Pipeline.execute(config)

# Execute with custom options
{:ok, results} = Pipeline.execute(config,
  workspace_dir: "/tmp/pipeline_workspace",
  debug: true
)

Configuration

Pipelines can be configured via:

  • Function options: Pipeline.execute(config, workspace_dir: "/custom/path")
  • Environment variables: PIPELINE_WORKSPACE_DIR, PIPELINE_OUTPUT_DIR, etc.
  • YAML configuration: workspace_dir, output_dir settings in the pipeline file

See Pipeline.Executor and Pipeline.Config for detailed documentation.

Summary

Functions

execute(workflow, opts \\ [])

Execute a pipeline workflow.

execute_step(step, context)

Execute a single pipeline step for testing or debugging.

get_config()

Get the current pipeline configuration.

health_check()

Check if the pipeline system is properly configured.

load_workflow(path)

Load a workflow configuration from a YAML file.

run(workflow_path, opts \\ [])

Load and execute a workflow in one call.

Functions

execute(workflow, opts \\ [])

@spec execute(
  map(),
  keyword()
) :: {:ok, map()} | {:error, String.t()}

Execute a pipeline workflow.

Parameters

  • workflow - Pipeline configuration map (from load_workflow/1)
  • opts - Execution options (optional)

Options

  • :workspace_dir - Directory for AI workspace operations (default: "./workspace")
  • :output_dir - Directory for saving pipeline outputs (default: "./outputs")
  • :checkpoint_dir - Directory for saving execution checkpoints (default: "./checkpoints")
  • :debug - Enable debug logging (default: false)

Examples

# Basic execution
{:ok, results} = Pipeline.execute(config)

# With custom directories
{:ok, results} = Pipeline.execute(config,
  workspace_dir: "/tmp/ai_workspace",
  output_dir: "/app/pipeline_outputs"
)

# With debug logging
{:ok, results} = Pipeline.execute(config, debug: true)

Returns

  • {:ok, results} - Map of step results keyed by step name
  • {:error, reason} - Execution failure with error details

execute_step(step, context)

@spec execute_step(map(), map()) :: {:ok, map()} | {:error, String.t()}

Execute a single pipeline step for testing or debugging.

Examples

step = %{"name" => "analyze", "type" => "claude", "prompt" => "Analyze this code"}
context = %{workspace_dir: "/tmp", results: %{}}
{:ok, result} = Pipeline.execute_step(step, context)

get_config()

@spec get_config() :: %{required(atom()) => any()}

Get the current pipeline configuration.

Returns application-level configuration including default directories, test mode, and debug settings.

Examples

config = Pipeline.get_config()
IO.inspect(config.workspace_dir)

health_check()

@spec health_check() :: :ok | {:error, [String.t()]}

Check if the pipeline system is properly configured.

Validates that required dependencies and configurations are available.

Examples

case Pipeline.health_check() do
  :ok -> IO.puts("Pipeline system ready")
  {:error, _problems} -> IO.puts("Configuration issues found")
end

load_workflow(path)

@spec load_workflow(String.t()) :: {:ok, map()} | {:error, String.t()}

Load a workflow configuration from a YAML file.

Examples

{:ok, config} = Pipeline.load_workflow("examples/simple_workflow.yaml")
{:ok, config} = Pipeline.load_workflow("/path/to/my_pipeline.yaml")

run(workflow_path, opts \\ [])

@spec run(
  String.t(),
  keyword()
) :: {:ok, map()} | {:error, String.t()}

Load and execute a workflow in one call.

Convenience function that combines load_workflow/1 and execute/2.

Examples

{:ok, results} = Pipeline.run("my_pipeline.yaml")
{:ok, results} = Pipeline.run("my_pipeline.yaml", debug: true)