# Pipelines

The Pipeline system provides composable RAG workflows with parallel execution, caching, and error handling.

## Overview

Pipelines consist of:
- **Steps** - Individual processing units
- **Context** - Shared state between steps
- **Executor** - Runs steps with caching/retry/telemetry

## Creating a Pipeline

```elixir
alias Rag.Pipeline
alias Rag.Pipeline.Step

pipeline = Pipeline.new(:rag_pipeline, description: "Complete RAG workflow")
```

### Adding Steps

```elixir
pipeline = Pipeline.add_step(pipeline,
  name: :embed_query,
  module: Steps,
  function: :embed_query,
  args: [model: "gemini"],
  timeout: 10_000,
  on_error: {:retry, 2},
  cache: true
)
```

### Step Options

| Option | Default | Description |
|--------|---------|-------------|
| `name` | required | Step identifier (atom) |
| `module` | required | Module containing function |
| `function` | required | Function to call |
| `args` | `[]` | Arguments passed to function |
| `inputs` | `nil` | Dependencies on previous steps |
| `parallel` | `false` | Run concurrently |
| `on_error` | `:halt` | Error handling strategy |
| `cache` | `false` | Cache results with ETS |
| `timeout` | `nil` | Timeout in milliseconds |

## Step Functions

Every step function must have this signature:

```elixir
def step_name(input, context, opts) do
  # input: Result from previous step(s)
  # context: Pipeline.Context struct
  # opts: Keyword list from step's :args

  # Return one of:
  {:ok, result}
  {:ok, result, updated_context}
  {:error, reason}
end
```

### Example Step

```elixir
defmodule MySteps do
  alias Rag.Pipeline.Context

  def embed_query(query, context, opts) do
    router = opts[:router]
    case Router.execute(router, :embeddings, [query], []) do
      {:ok, [embedding], _} ->
        updated_context = %{context | query_embedding: embedding}
        {:ok, embedding, updated_context}
      {:error, reason} ->
        {:error, reason}
    end
  end
end
```

## Context

The context holds state throughout pipeline execution:

```elixir
alias Rag.Pipeline.Context

# Create context
context = Context.new("What is Elixir?")

# Context structure
%Context{
  input: any(),                    # Original input
  query: String.t() | nil,
  query_embedding: [float()] | nil,
  retrieval_results: list(),
  reranked_results: list(),
  context_text: String.t() | nil,
  response: String.t() | nil,
  metadata: %{step_results: %{}},
  errors: []
}
```

### Context API

```elixir
# Store step result
context = Context.put_step_result(context, :embed, embedding)

# Get step result
embedding = Context.get_step_result(context, :embed)

# Store metadata
context = Context.put_metadata(context, :user_id, 123)

# Add error (for :continue strategy)
context = Context.add_error(context, {:step_failed, :rerank})
```

## Input Dependencies

### No Dependencies (Default)

Step receives output from previous step:

```elixir
Pipeline.add_step(pipeline, name: :step2, ...)
# step2 receives output from step1
```

### Single Dependency

```elixir
Pipeline.add_step(pipeline,
  name: :generate,
  inputs: [:retrieve],
  ...
)
# generate receives the :retrieve step's result
```

### Multiple Dependencies

```elixir
Pipeline.add_step(pipeline,
  name: :combine,
  inputs: [:semantic_search, :fulltext_search],
  ...
)
# combine receives: %{semantic_search: [...], fulltext_search: [...]}
```

## Error Handling

### `:halt` (Default)

Stop pipeline immediately on error:

```elixir
Pipeline.add_step(pipeline,
  name: :critical,
  on_error: :halt  # Pipeline stops if this fails
)
```

### `:continue`

Log error but continue execution:

```elixir
Pipeline.add_step(pipeline,
  name: :optional_rerank,
  on_error: :continue  # Skip if fails, continue pipeline
)
```

### `{:retry, n}`

Retry up to n times:

```elixir
Pipeline.add_step(pipeline,
  name: :api_call,
  on_error: {:retry, 3}  # Retry up to 3 times
)
```

## Parallel Execution

Independent steps can run concurrently:

```elixir
Pipeline.new(:hybrid_search)
|> Pipeline.add_step(name: :embed, ...)
|> Pipeline.add_step(
  name: :semantic_search,
  inputs: [:embed],
  parallel: true  # Runs in parallel
)
|> Pipeline.add_step(
  name: :fulltext_search,
  inputs: [:embed],
  parallel: true  # Runs in parallel
)
|> Pipeline.add_step(
  name: :combine,
  inputs: [:semantic_search, :fulltext_search]  # Waits for both
)
```

## Caching

Cache expensive operations with ETS:

```elixir
Pipeline.add_step(pipeline,
  name: :embed,
  cache: true  # Results cached in ETS
)
```

**Benefits:**
- Same input skips execution, uses cache
- Persists across pipeline runs
- Useful for embeddings, expensive computations

**Performance:**
```
First run:  3500ms (embedding computed)
Second run: 2100ms (embedding cached)
Speedup:    ~40%
```

## Timeouts

Prevent hanging on slow steps:

```elixir
Pipeline.add_step(pipeline,
  name: :llm_generate,
  timeout: 30_000  # 30 second timeout
)
```

## Telemetry

Pipeline emits telemetry events:

```elixir
# Start event
[:rag, :pipeline, :step, :start]
# Metadata: %{pipeline: :name, step: :step_name, attempt: 0}

# Stop event
[:rag, :pipeline, :step, :stop]
# Measurements: %{duration: microseconds}

# Exception event
[:rag, :pipeline, :step, :exception]
# Metadata: %{pipeline: :name, step: :step_name, error: reason}
```

### Attaching Handlers

```elixir
:telemetry.attach(
  "pipeline-logger",
  [:rag, :pipeline, :step, :stop],
  fn _event, measurements, metadata, _config ->
    IO.puts("Step #{metadata.step} completed in #{measurements.duration}μs")
  end,
  nil
)
```

## Complete Example

```elixir
defmodule MyApp.RAGPipeline do
  alias Rag.Pipeline
  alias Rag.Pipeline.Context

  def build(router, retriever) do
    Pipeline.new(:rag_pipeline, description: "Complete RAG")
    |> Pipeline.add_step(
      name: :embed_query,
      module: __MODULE__,
      function: :embed_query,
      args: [router: router],
      timeout: 10_000,
      on_error: {:retry, 2},
      cache: true
    )
    |> Pipeline.add_step(
      name: :retrieve,
      module: __MODULE__,
      function: :retrieve,
      args: [retriever: retriever],
      inputs: [:embed_query],
      timeout: 5_000
    )
    |> Pipeline.add_step(
      name: :rerank,
      module: __MODULE__,
      function: :rerank,
      args: [router: router],
      inputs: [:retrieve],
      on_error: :continue  # Optional step
    )
    |> Pipeline.add_step(
      name: :generate,
      module: __MODULE__,
      function: :generate,
      args: [router: router],
      inputs: [:rerank],
      timeout: 30_000
    )
  end

  def embed_query(query, context, opts) do
    router = opts[:router]
    case Router.execute(router, :embeddings, [query], []) do
      {:ok, [embedding], _} ->
        {:ok, embedding, %{context | query: query, query_embedding: embedding}}
      {:error, reason} ->
        {:error, reason}
    end
  end

  def retrieve(embedding, context, opts) do
    retriever = opts[:retriever]
    case Retriever.retrieve(retriever, {embedding, context.query}, limit: 10) do
      {:ok, results} ->
        {:ok, results, %{context | retrieval_results: results}}
      {:error, reason} ->
        {:error, reason}
    end
  end

  def rerank(results, context, opts) do
    router = opts[:router]
    reranker = Rag.Reranker.LLM.new(router: router)
    case Rag.Reranker.rerank(reranker, context.query, results, top_k: 5) do
      {:ok, reranked} ->
        {:ok, reranked, %{context | reranked_results: reranked}}
      {:error, _} ->
        # Return original results if reranking fails
        {:ok, results}
    end
  end

  def generate(results, context, opts) do
    router = opts[:router]
    context_text = Enum.map(results, & &1.content) |> Enum.join("\n\n")

    prompt = """
    Answer based on context:
    #{context_text}

    Question: #{context.query}
    """

    case Router.execute(router, :text, prompt, []) do
      {:ok, response, _} ->
        {:ok, response, %{context | response: response, context_text: context_text}}
      {:error, reason} ->
        {:error, reason}
    end
  end
end

# Usage
{:ok, router} = Router.new(providers: [:gemini])
retriever = %Rag.Retriever.Hybrid{repo: Repo}

pipeline = MyApp.RAGPipeline.build(router, retriever)
context = Context.new("What is GenServer?")

case Pipeline.execute(pipeline, context.input) do
  {:ok, response, final_context} ->
    IO.puts("Answer: #{response}")
    IO.puts("Used #{length(final_context.retrieval_results)} documents")

  {:error, reason} ->
    IO.puts("Pipeline failed: #{inspect(reason)}")
end
```

## Configuration Best Practices

### Timeouts

| Step Type | Suggested Timeout |
|-----------|-------------------|
| Embedding | 10-15 seconds |
| Database query | 5 seconds |
| LLM generation | 30-60 seconds |
| Overall pipeline | Sum + buffer |

### Caching

Enable for:
- Embeddings (deterministic)
- Expensive computations
- Repeated queries

Disable for:
- User-specific results
- Time-sensitive data

### Error Handling

| Step Type | Strategy |
|-----------|----------|
| Critical (embedding, retrieval) | `:halt` or `{:retry, 2}` |
| Enhancement (reranking) | `:continue` |
| External APIs | `{:retry, 3}` |

## Next Steps

- [Retrievers](retrievers.md) - Retrieval strategies for pipelines
- [Rerankers](rerankers.md) - Improve retrieval quality
- [Agent Framework](agent_framework.md) - Integrate agents in pipelines