Flows¶

pkg/agent/flow/ provides five reusable processor-composition patterns — SequentialFlow, ParallelFlow, ConditionalFlow, LoopFlow, PipelineFlow. They are engine.Processor implementations, so they plug into Helix routes anywhere a processor is expected and can be nested freely. This package is distinct from the agent-type sub-agent orchestration patterns under pkg/agent/adk/ (the names overlap deliberately): adk's runSequential / runParallel / runLoop compose *Agents on llm.Messages; flows compose engine.Processors on *exchange.Exchanges.

What it does¶

Every flow type implements engine.Processor:

type Processor interface {
    Process(ctx context.Context, ex *exchange.Exchange) error
}

Because flows ARE processors, you can nest them — a ParallelFlow inside one branch of a ConditionalFlow inside a step of a SequentialFlow is fine. The package builds on pkg/engine and pkg/exchange, so flows operate on the same data envelope as the rest of Helix's processor pipeline.

Use this when you're composing route processors. For LLM-driven sub-agent orchestration use the matching agent type (sequential, parallel, loop) instead.

Implementations¶

SequentialFlow¶

Runs processors one after another against the SAME exchange. Mutations accumulate (each processor sees the previous one's changes). Stops at the first error.

type SequentialFlow struct {
    Steps []engine.Processor
}

Behavior: - Before each step: ctx.Err() check and ex.IsFailed() check. - Step error returns sequential flow: step <i> failed: <err>. - A failed exchange short-circuits with ex.Exception().

ParallelFlow¶

Runs processors concurrently. Each branch gets a COPY of the exchange (ex.Copy()) so branches don't interfere. Results are combined via the Aggregator or, by default, by joining branch bodies with newlines.

type ParallelFlow struct {
    Branches   []engine.Processor
    Aggregator func(results []*exchange.Exchange, target *exchange.Exchange)
}

Behavior: - Spawns one goroutine per branch, waits for all (no concurrency cap). - Collects results even on partial failure. The aggregator runs regardless; the first branch error is then returned. - Default aggregator: combined := r1.body + "\n" + r2.body + ..., written back to the original exchange's In().Body().

ConditionalFlow¶

Routes to one of two processors based on a predicate.

type ConditionalFlow struct {
    Predicate PredicateFunc          // (ctx, ex) -> (bool, error)
    Then      engine.Processor
    Else      engine.Processor       // optional
}

Behavior: - Predicate error → conditional flow: predicate error: <err>. - Predicate true → run Then (no-op when Then is nil). - Predicate false → run Else if set, else no-op.

LoopFlow¶

Repeats a body processor until a Go-level condition returns false or MaxIterations is reached.

type LoopFlow struct {
    Body          engine.Processor
    Condition     LoopConditionFunc   // (ctx, ex, i) -> (continue, error)
    MaxIterations int                  // 0 = unlimited
}

Behavior: - Per-iteration ctx.Err() check. - Runs Body, then evaluates Condition (post-condition loop — body always runs at least once). - Condition returning false is the clean exit (returns nil). - MaxIterations exhaustion returns loop flow: max iterations (N) reached. - MaxIterations <= 0 means unlimited (use with caution; a runaway condition will spin forever).

This is the processor-side counterpart to the loop agent type. The loop agent uses an LLM-emitted DONE sentinel as its condition; LoopFlow uses any Go function — useful when the condition is exchange-state-driven ("loop until body.score > 0.9").

PipelineFlow¶

Chains processors, passing the OUTPUT of each as the INPUT to the next. Unlike SequentialFlow (which shares one exchange), PipelineFlow creates a fresh exchange between stages, seeded with the previous stage's body and headers.

type PipelineFlow struct {
    Stages []engine.Processor
}

Behavior: - First stage runs on the input exchange. - Subsequent stages run on a new exchange whose body and headers are copied from the previous stage's exchange. - After the last stage, the final body + headers are written back to the original exchange so the caller sees the result. - A stage error returns pipeline flow: stage <i> failed: <err>. - ctx.Err() is checked before each stage.

Use PipelineFlow when stages should NOT see each other's state — when each stage is a pure transformation of bytes. Use SequentialFlow when stages should share headers / properties / the exchange itself.

Configuration¶

Flows are Go-level constructs. They are not directly addressable from YAML route definitions — route YAML uses the processor types under pkg/processor/ (which have their own choice / multicast / split / aggregate / pipeline analogues). Flows are most useful when embedding agents/processors in custom Go code or building higher-order processors.

Examples¶

A parallel-then-sequential composition in Go:

import "github.com/arvenza/helix/pkg/agent/flow"

route := flow.NewSequentialFlow(
    enrichWithMetadata,
    flow.NewParallelFlow(
        searchVectorStore,
        searchSQL,
        searchAPI,
    ),
    rankAndDedupe,
    formatResponse,
)

A retry-until-good with bounded attempts:

loop := flow.NewLoopFlow(
    tryCall,
    func(ctx context.Context, ex *exchange.Exchange, i int) (bool, error) {
        // Continue if body is empty (need to retry).
        return ex.In().BodyAsString() == "", nil
    },
    5,
)

Source¶

pkg/agent/flow/flow.go — All five flow types and their Process implementations.