docs/guides/composing-pipelines.md

# Composing Custom Pipelines

ACE is built on a composable pipeline engine. Every runner (`ACE`, `BrowserUse`,
`LangChain`, `ClaudeCode`, `TraceAnalyser`) is a thin wrapper around a `Pipeline`
made of steps. You can compose your own pipelines by mixing and matching these
steps — or writing custom ones.

## Three Levels of ACE

| Level | Pattern | Control |
|-------|---------|---------|
| **Zero-config** | `ACELiteLLM.from_model("gpt-4o-mini")` | Roles + pipeline auto-created |
| **Role customisation** | `ACE.from_roles(agent=..., reflector=..., ...)` | Custom roles, pipeline auto-composed |
| **Pipeline composition** | `Pipeline([AgentStep(...), ...])` | Full control over step ordering |

This guide covers **Level 3** — composing pipelines directly.

## Anatomy of an ACE Pipeline

Every ACE pipeline is a sequence of steps, each with a `requires`/`provides`
contract that declares what context fields it reads and writes:

```
AgentStep ─────> EvaluateStep ─────> ReflectStep ─────> UpdateStep
  provides:        provides:           provides:          provides:
  agent_output     trace                reflections        skill_manager_output
                                                          (also mutates skillbook
                                                           via the SM's tools)
```

The pipeline validates these contracts at construction time — if a step requires
a field that no earlier step provides, you'll get an error immediately.

## Composing from Steps

All pipeline classes and ACE steps are importable from `ace`:

```python
from ace import (
    # Pipeline engine
    Pipeline, Branch, MergeStrategy, StepProtocol, SampleResult,
    # ACE context
    ACEStepContext, SkillbookView,
    # Roles
    Agent, Reflector, SkillManager,
    # Steps
    AgentStep, EvaluateStep, learning_tail,
    # Types
    Sample, Skillbook, SimpleEnvironment,
)

skillbook = Skillbook()

pipe = Pipeline([
    AgentStep(Agent("gpt-4o-mini"), skillbook),
    EvaluateStep(SimpleEnvironment()),
    *learning_tail(Reflector("gpt-4o-mini"), SkillManager("gpt-4o-mini"), skillbook),
])
```

## Using `learning_tail()`

The `learning_tail()` helper returns the standard learning step sequence:

```python
from ace import learning_tail, Reflector, SkillManager, Skillbook

steps = learning_tail(
    Reflector(llm),
    SkillManager(llm),
    Skillbook(),
    dedup_manager=my_dedup_manager,      # optional
    checkpoint_dir="/tmp/checkpoints",    # optional
)
# Returns: [ReflectStep, UpdateStep,
#           DeduplicateStep, CheckpointStep]
```

Use it when building custom integrations that provide their own execute step but
want the standard learning pipeline.

## Inspecting Runner Presets with `build_steps()`

Every runner has a `build_steps()` classmethod that returns the step list it
would use internally. This lets you inspect, modify, and recompose:

```python
from ace import ACE, Pipeline, ACERunner, Skillbook

# Get the default steps
steps = ACE.build_steps(
    agent=my_agent,
    reflector=my_reflector,
    skill_manager=my_skill_manager,
    environment=my_env,
)

# Insert a custom step after EvaluateStep
steps.insert(2, MyLoggingStep())

# Build your own pipeline and runner
skillbook = Skillbook()
pipe = Pipeline(steps)
runner = ACERunner(pipeline=pipe, skillbook=skillbook)
results = runner.run(samples)
```

All runners support `build_steps()`: `ACE`, `BrowserUse`, `ClaudeCode`,
`LangChain`, and `TraceAnalyser`.

## Writing Custom Steps

A step is any object satisfying `StepProtocol` — no base class needed:

```python
from ace import ACEStepContext

class MyLoggingStep:
    requires = frozenset({"agent_output"})
    provides = frozenset()

    def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
        print(f"Agent answered: {ctx.agent_output.final_answer}")
        return ctx
```

Key rules:

- `requires`: frozenset of context field names this step reads
- `provides`: frozenset of context field names this step writes
- `__call__`: receives and returns `ACEStepContext` (use `ctx.replace(...)` for updates)
- Steps should be stateless — no internal counters

## Mixing Integrations

You can compose steps from different integrations into one pipeline. For example,
combining a browser-use execute step with custom learning:

```python
from ace import Pipeline, learning_tail, Reflector, SkillManager, Skillbook
from ace.integrations.browser_use import BrowserExecuteStep, BrowserToTrace

skillbook = Skillbook()
pipe = Pipeline([
    BrowserExecuteStep(browser_llm),
    BrowserToTrace(),
    MyCustomFilterStep(),  # your custom step
    *learning_tail(Reflector(llm), SkillManager(llm), skillbook),
])
```

Integration steps live in `ace.integrations` since they have
framework-specific dependencies.

## Running the Pipeline

### With a runner

The simplest way to run a custom pipeline is through `ACERunner`:

```python
from ace import ACERunner, Sample, Skillbook

runner = ACERunner(pipeline=pipe, skillbook=skillbook)
results = runner.run(
    [Sample(question="What is 2+2?", ground_truth="4")],
    epochs=1,
)
```

### Directly

You can also run the pipeline directly by constructing contexts yourself:

```python
from ace import Pipeline, ACEStepContext, SkillbookView, Sample, Skillbook

ctx = ACEStepContext(
    sample=Sample(question="What is 2+2?", ground_truth="4"),
    skillbook=SkillbookView(skillbook),
)

results = pipe.run([ctx])
pipe.wait_for_background()  # wait for async learning steps
```

## Branching (Parallel Steps)

The pipeline engine supports parallel branches for steps that can run
concurrently:

```python
from ace import Pipeline, Branch, MergeStrategy

pipe = Pipeline([
    AgentStep(agent, skillbook),
    Branch(
        [EvaluateStep(env_a), EvaluateStep(env_b)],
        merge=MergeStrategy.LAST,
    ),
    *learning_tail(reflector, skill_manager, skillbook),
])
```

See the [Pipeline Engine docs](../pipeline/branching.md) for full branching
and merge strategy details.

## Using RRStep (Recursive Reflector)

`RRStep` satisfies both `StepProtocol` and `ReflectorLike`, so it can be used
in two ways:

### As a drop-in reflector replacement

Pass it anywhere a `Reflector` is expected:

```python
from ace import ACELiteLLM
from ace.rr import RRStep, RRConfig

ace = ACELiteLLM.from_model("gpt-4o-mini", reflector=RRStep("gpt-4o-mini", config=RRConfig(max_requests=10)))
```

### As a pipeline step

Place it directly in a pipeline (it provides `reflections`):

```python
from ace import Pipeline, learning_tail, SkillManager, Skillbook
from ace.rr import RRStep, RRConfig

skillbook = Skillbook()
rr = RRStep("gpt-4o-mini", config=RRConfig(max_requests=15))

pipe = Pipeline([
    MyExecuteStep(),
    MyToTrace(),
    rr,  # replaces ReflectStep — provides "reflections"
    *learning_tail(None, SkillManager("gpt-4o-mini"), skillbook)[1:],  # skip ReflectStep
])
```

### With recursion enabled

Allow the RR to decompose large batch inputs via recursive child sessions:

```python
from ace.rr import RRStep, RRConfig

rr = RRStep(
    "gpt-4o",
    config=RRConfig(max_requests=40, max_depth=1),  # depth=1 allows one level of recursion
)
```

## Available Steps

All steps are importable from `ace`:

| Step | Purpose |
|------|---------|
| `AgentStep` | Execute Agent role |
| `EvaluateStep` | Run TaskEnvironment evaluation |
| `ReflectStep` | Run Reflector role (async boundary) |
| `UpdateStep` | Run the agentic SkillManager; its tools mutate the skillbook directly |
| `DeduplicateStep` | Merge near-duplicate skills |
| `CheckpointStep` | Save skillbook to disk |
| `LoadTracesStep` | Load JSONL trace files |
| `ExportSkillbookMarkdownStep` | Export skillbook as markdown |
| `ObservabilityStep` | Generic observability hook |
| `PersistStep` | Persist step output |
| `OpikStep` | Log traces to Opik |
| `RRStep` | Recursive Reflector |

Integration steps (in `ace.integrations`):

| Step | Integration |
|------|-------------|
| `BrowserExecuteStep` / `BrowserToTrace` | browser-use |
| `LangChainExecuteStep` / `LangChainToTrace` | LangChain |
| `ClaudeCodeExecuteStep` / `ClaudeCodeToTrace` | Claude Code |
| `ClaudeSDKExecuteStep` / `ClaudeSDKToTrace` | Anthropic Python SDK |
| `OpenClawToTraceStep` | OpenClaw |