Pipeline

🧬 Genesis Pipeline: Self-Improving AI System

Quick Start with Genesis

# Generate a new AI pipeline with one command
mix pipeline.generate.live "Create a sentiment analysis pipeline"
# The system will create a complete, executable pipeline in evolved_pipelines/
# Run your generated pipeline immediately:
mix pipeline.run evolved_pipelines/sentiment_analyzer_*.yaml

📦 Library Usage

Add to Your Project

# mix.exs
defp deps do
 [
 # From Hex.pm (recommended)
 {:pipeline_ex, "~> 0.1.0"}
 # Or from GitHub
 # {:pipeline_ex, git: "https://github.com/nshkrdotcom/pipeline_ex.git", tag: "v0.1.0"}
 ]
end

Simple API

# Load and execute a pipeline
{:ok, config} = Pipeline.load_workflow("my_analysis.yaml")
{:ok, results} = Pipeline.execute(config)
# Execute with custom configuration
{:ok, results} = Pipeline.execute(config,
 workspace_dir: "/app/ai_workspace",
 output_dir: "/app/pipeline_outputs"
)
# Convenience function
{:ok, results} = Pipeline.run("my_analysis.yaml", debug: true)
# Health check
case Pipeline.health_check() do
 :ok -> IO.puts("Pipeline system ready")
 {:error, issues} -> IO.puts("Issues: #{inspect(issues)}")
end

Configuration Options

Integration Examples

# Phoenix controller
defmodule MyAppWeb.AIController do
 def analyze(conn, %{"code" => code}) do
 case Pipeline.run("pipelines/code_analysis.yaml",
 workspace_dir: "/tmp/ai_workspace") do
 {:ok, %{"analysis" => result}} -> 
 json(conn, %{analysis: result})
 {:error, reason} -> 
 put_status(conn, 500) |> json(%{error: reason})
 end
 end
end
# Background job with Oban
defmodule MyApp.AnalysisWorker do
 use Oban.Worker, queue: :ai_analysis
 
 def perform(%Oban.Job{args: %{"project_id" => project_id}}) do
 case Pipeline.execute(get_analysis_config(), 
 workspace_dir: "/tmp/analysis_#{project_id}",
 output_dir: "/app/results/#{project_id}") do
 {:ok, results} -> 
 MyApp.Projects.update_analysis(project, results)
 :ok
 {:error, reason} -> 
 {:error, reason}
 end
 end
end

Testing Integration

# Enable mock mode for development/testing
Application.put_env(:pipeline, :test_mode, true)
# All AI calls will be mocked
{:ok, results} = Pipeline.execute(config)

Features

📦 Library-Ready

🤖 AI Integration

⚡ Advanced Features

Quick Start

1. Installation

As a Library Dependency (Recommended)

# mix.exs
defp deps do
 [
 # From Hex.pm (recommended)
 {:pipeline_ex, "~> 0.0.1"}
 
 # Or from GitHub
 # {:pipeline_ex, git: "https://github.com/nshkrdotcom/pipeline_ex.git", tag: "v0.0.1"}
 ]
end

Standalone Development

git clone <repository>
cd pipeline_ex
mix deps.get

2. Try the Showcase (Recommended)

# Mock mode (safe, free, fast)
mix showcase # Complete demo with mocks
# Live mode (real API calls, costs money)
mix showcase --live # Complete demo with live APIs

🎭 Mock vs Live Mode

Running Tests

Mock Mode (Recommended)

mix test # All tests with mocked APIs (fast, free)

Live Mode (Real API Calls)

# Setup authentication first:
export GEMINI_API_KEY="your_api_key"
claude login
# Run tests with real APIs:
mix pipeline.test.live # Only integration tests with live APIs (costs money)

Execution Modes

Mode Examples

# Mock mode - fast, no API costs
mix pipeline.run examples/comprehensive_config_example.yaml
# Live mode - real AI responses 
mix pipeline.run.live examples/comprehensive_config_example.yaml
# Mixed mode - context-dependent
TEST_MODE=mixed mix test

Configuration

API Keys

Claude

claude login

Gemini

export GEMINI_API_KEY="your_gemini_api_key_here"

config :pipeline, gemini_api_key: "your_api_key"

Workflow Configuration

workflow:
 name: "simple_test_workflow"
 description: "Test basic claude functionality"
 
 steps:
 - name: "analyze_code"
 type: "claude"
 prompt: 
 - type: "static"
 content: |
 Analyze this simple Python function and provide feedback:

 def add(a, b):
 return a + b

 Please provide your analysis in JSON format.
 
 - name: "plan_improvements"
 type: "gemini" 
 prompt:
 - type: "static"
 content: |
 Based on the previous analysis, create a plan to improve the function.
 Consider error handling, type hints, and documentation.

Example Usage

Library Usage in Your Application

defmodule MyApp.AIProcessor do
 @doc "Analyze code using the pipeline library"
 def analyze_code(code_content) do
 # Load your pipeline configuration
 case Pipeline.load_workflow("pipelines/code_analysis.yaml") do
 {:ok, config} ->
 # Execute with custom workspace
 Pipeline.execute(config,
 workspace_dir: "/tmp/ai_workspace",
 debug: true
 )
 {:error, reason} ->
 {:error, "Failed to load workflow: #{reason}"}
 end
 end
 
 @doc "Health check for the AI system"
 def system_ready? do
 case Pipeline.health_check() do
 :ok -> true
 {:error, _issues} -> false
 end
 end
end
# Usage in your application
{:ok, analysis} = MyApp.AIProcessor.analyze_code(user_code)
IO.inspect(analysis["analysis_step"])

Simple Script Example

#!/usr/bin/env elixir
Mix.install([
 {:pipeline_ex, git: "https://github.com/nshkrdotcom/pipeline_ex.git"}
])
# Load and execute pipeline
case Pipeline.run("test_simple_workflow.yaml", output_dir: "outputs") do
 {:ok, results} ->
 IO.puts("✅ Pipeline completed!")
 IO.inspect(results)
 
 {:error, reason} ->
 IO.puts("❌ Pipeline failed: #{reason}")
end

Testing Different Scenarios

# Test with mock responses (fast)
TEST_MODE=mock elixir run_example.exs
# Test with real Claude + mock Gemini
# (useful when you have Claude access but no Gemini API key)
TEST_MODE=mixed elixir -e "
System.put_env(\"FORCE_MOCK_GEMINI\", \"true\")
Code.compile_file(\"run_example.exs\")
"
# Full live test (requires both API keys)
elixir run_example.exs

Running the Comprehensive Example

Mock Mode (Safe, No API Keys Required)

# Run the comprehensive example with mocked AI responses
mix pipeline.run examples/comprehensive_config_example.yaml
# Run with debug output to see detailed execution
PIPELINE_DEBUG=true mix pipeline.run examples/comprehensive_config_example.yaml

Live Mode (Real AI Providers)

1. Set Up API Keys

# Set your Gemini API key (get from https://aistudio.google.com/)
export GEMINI_API_KEY="your_gemini_api_key_here"
# Authenticate Claude CLI (pre-authenticated, no API key needed)
claude auth

2. Run in Live Mode

# Run with real AI providers
mix pipeline.run.live examples/comprehensive_config_example.yaml
# Run with full debug logging
PIPELINE_DEBUG=true PIPELINE_LOG_LEVEL=debug mix pipeline.run.live examples/comprehensive_config_example.yaml

What the Comprehensive Example Demonstrates

Enhanced Claude Step Types

🎯 Claude Smart (claude_smart)

- name: "smart_analysis" 
 type: "claude_smart"
 preset: "analysis" # Uses analysis-optimized settings
 prompt: [...]

🗣️ Claude Session (claude_session)

- name: "session_start"
 type: "claude_session" 
 session_name: "math_tutor"
 session_config:
 persist: true
 max_turns: 50

📄 Claude Extract (claude_extract)

- name: "extract_json"
 type: "claude_extract"
 preset: "analysis"
 extraction_config:
 format: "json"
 post_processing: ["extract_code_blocks", "extract_recommendations"]
 include_metadata: true

⚡ Claude Batch (claude_batch)

- name: "batch_analysis"
 type: "claude_batch"
 batch_config:
 max_parallel: 3
 tasks:
 - id: "task1"
 prompt: "Analyze JavaScript code..."
 - id: "task2" 
 prompt: "Analyze Python code..."

🛡️ Claude Robust (claude_robust)

- name: "robust_analysis"
 type: "claude_robust"
 retry_config:
 max_retries: 3
 backoff_strategy: "exponential"
 fallback_action: "simplified_prompt"

Testing Enhanced Step Types

# Test individual enhanced step types
mix pipeline.run.live examples/claude_smart_example.yaml
mix pipeline.run.live examples/claude_session_example.yaml 
mix pipeline.run.live examples/claude_extract_example.yaml
mix pipeline.run.live examples/claude_batch_example.yaml
mix pipeline.run.live examples/claude_robust_example.yaml
# Or run all enhanced examples in mock mode (free)
mix pipeline.run examples/claude_smart_example.yaml
mix pipeline.run examples/claude_session_example.yaml
# ... etc

🚀 Async Streaming (NEW)

Why Use Async Streaming?

Basic Streaming Example

- name: "streaming_claude"
 type: "claude"
 claude_options:
 # Enable async streaming
 async_streaming: true
 stream_handler: "console" # Real-time console output
 stream_buffer_size: 100 # Message buffer size
 
 # Standard options
 max_turns: 10
 allowed_tools: ["Write", "Read", "Edit"]

Available Stream Handlers

Streaming Examples

# Simple streaming example
mix pipeline.run examples/clean_streaming_numbers.yaml
# Multi-message streaming with file operations
mix pipeline.run examples/streaming_file_operations.yaml
# Run streaming tests
mix test test/integration/async_streaming_test.exs

Works with All Claude Step Types

Environment Configuration

# API Configuration
export GEMINI_API_KEY="your_gemini_api_key"
# Note: Claude uses CLI authentication (claude auth), no API key needed
# Pipeline Directories
export PIPELINE_WORKSPACE_DIR="./workspace" # Claude's sandbox
export PIPELINE_OUTPUT_DIR="./outputs" # Result storage
export PIPELINE_CHECKPOINT_DIR="./checkpoints" # State management
# Logging and Debug
export PIPELINE_LOG_LEVEL="debug" # debug, info, warn, error
export PIPELINE_DEBUG="true" # Detailed execution logs
# Execution Mode
export TEST_MODE="live" # live, mock, mixed

Creating Your Own Workflows

Development

Running the Example

# Quick test with mocks
TEST_MODE=mock elixir run_example.exs
# Full test with APIs
elixir run_example.exs

Test Structure

test/
├── unit/ # Fast unit tests (mocked)
├── integration/ # Integration tests (live APIs) 
├── fixtures/ # Test data and workflows
└── support/ # Test helpers

Adding New Tests

defmodule MyTest do
 use ExUnit.Case
 use Pipeline.TestCase # Provides test mode helpers
 
 test "my feature works in mock mode" do
 # Test automatically uses mocks
 assert Pipeline.execute_something() == expected_result
 end
end

Debugging

# See detailed execution logs
DEBUG=true elixir run_example.exs
# See API request/response details
VERBOSE=true elixir run_example.exs

Common Issues

Claude CLI Not Authenticated

# Error: Claude CLI not found or not authenticated
claude login

Missing Gemini API Key

# Error: GEMINI_API_KEY environment variable not set
export GEMINI_API_KEY="your_key_here"

Mixed Results (Some Success, Some Failure)

# Check which providers are mocked vs live
TEST_MODE=mock elixir run_example.exs # All mocked
TEST_MODE=live elixir run_example.exs # All live

License