Flowneer is a tiny (~3 kB gzipped), zero-dependency TypeScript flow builder that gives you full control over deterministic, stateful LLM agents and workflows.

Why Flowneer?

• Ultra-lightweight — ~3 kB gzipped core, zero dependencies
• Fluent & composable — Chain steps with shared mutable state
• Full control flow primitives — .startWith(), .then(), .branch(), .loop(), .parallel(), .batch(), .anchor() jumps
• Streaming-first — Real-time .stream() with event/chunk yielding
• Precise extensibility — Subclass with .extend([plugins]) and scope hooks/plugins exactly where needed (via StepFilter globs/predicates)
• Production-ready patterns — Built-in presets for ReAct, sequential crews, supervisor-workers, round-robin debate, refinement loops

Plugins unlock what you actually need

• Tool calling & registries
• ReAct / reasoning loops
• Memory (buffer, summary, KV)
• Human-in-the-loop interrupts
• Structured output parsing
• Rate limiting, retries, timeouts, tracing, eval, graph export/import

No forced abstractions. No monolith. Just a fast, deterministic builder that stays out of your way while giving you structured concurrency, cancellation, observability, and agentic power.

Flowneer is currently under heavy development with ongoing pattern exploration and architectural refinement. Breaking changes are expected frequently, potentially on a daily basis, as the core design is actively evolving.

Install

bun add flowneer

For LLM Agents

llms.txt llms-full.txt

Quick start

import { FlowBuilder } from "flowneer";

interface State {
  count: number;
}

await new FlowBuilder<State>()
  .startWith(async (s) => {
    s.count = 0;
  })
  .then(async (s) => {
    s.count += 1;
  })
  .then(async (s) => {
    console.log(s.count);
  }) // 1
  .run({ count: 0 });

Every step receives a shared state object (s) that you mutate directly. That's the whole data model.

API

startWith(fn, options?)

Set the first step, resetting any prior chain.

then(fn, options?)

Append a sequential step.

branch(router, branches, options?)

Route to a named branch based on the return value of router.

await new FlowBuilder<{ role: string; message: string }>()
  .startWith(async (s) => {
    s.role = "admin";
  })
  .branch((s) => s.role, {
    admin: async (s) => {
      s.message = "Welcome, admin!";
    },
    guest: async (s) => {
      s.message = "Limited access.";
    },
  })
  .then(async (s) => console.log(s.message))
  .run({ role: "", message: "" });
// -> Welcome, admin!

loop(condition, body)

Repeat a sub-flow while condition returns true.

await new FlowBuilder<{ ticks: number }>()
  .startWith(async (s) => {
    s.ticks = 0;
  })
  .loop(
    (s) => s.ticks < 3,
    (b) =>
      b.startWith(async (s) => {
        s.ticks += 1;
      }),
  )
  .then(async (s) => console.log("done, ticks =", s.ticks))
  .run({ ticks: 0 });
// -> done, ticks = 3

batch(items, processor, options?)

Run a sub-flow once per item. The current item is written to shared.__batchItem by default. Pass a { key } option to name the item slot — required for nested batches.

await new FlowBuilder<{
  numbers: number[];
  results: number[];
  __batchItem?: number;
}>()
  .startWith(async (s) => {
    s.results = [];
  })
  .batch(
    (s) => s.numbers,
    (b) =>
      b.startWith(async (s) => {
        s.results.push((s.__batchItem ?? 0) * 2);
      }),
  )
  .then(async (s) => console.log(s.results))
  .run({ numbers: [1, 2, 3], results: [] });
// -> [2, 4, 6]

parallel(fns, options?, reducer?)

Run multiple functions concurrently against the same shared state. When a reducer is provided, each fn receives its own shallow clone and the reducer merges results back.

await new FlowBuilder<{ posts?: any[]; users?: any[] }>()
  .parallel([
    async (s) => {
      s.posts = await fetch("/posts").then((r) => r.json());
    },
    async (s) => {
      s.users = await fetch("/users").then((r) => r.json());
    },
  ])
  .then(async (s) => console.log(s.posts?.length, s.users?.length))
  .run({});

anchor(name)

Insert a named marker in the step chain. Any NodeFn can return "#anchorName" to jump to that anchor, enabling iterative refinement loops without nesting.

await new FlowBuilder<{ draft: string; quality: number; passes: number }>()
  .startWith(async (s) => {
    s.draft = await generateDraft(s);
  })
  .anchor("refine")
  .then(async (s) => {
    s.quality = await scoreDraft(s.draft);
    if (s.quality < 0.8) {
      s.draft = await improveDraft(s.draft);
      s.passes++;
      return "#refine";
    }
  })
  .then(async (s) => console.log("Final draft after", s.passes, "passes"))
  .run({ draft: "", quality: 0, passes: 0 });

Pair with withCycles to cap the maximum number of jumps.

fragment() and .add(fragment)

Fragments are reusable partial flows that can be spliced into any FlowBuilder.

import { FlowBuilder, fragment } from "flowneer";

const enrich = fragment<State>()
  .then(async (s) => {
    s.enriched = true;
  })
  .then(async (s) => {
    s.input = s.input.trim();
  });

await new FlowBuilder<State>()
  .startWith(async (s) => {
    s.input = "  hello  ";
  })
  .add(enrich)
  .then(async (s) => console.log(s.input))
  .run({ input: "", enriched: false, summary: "" });

Fragments support all step types. They cannot be run directly — calling .run() on a fragment throws.

run(shared, params?, options?)

Execute the flow. Optionally pass a params object that every step receives as a second argument, and an AbortSignal to cancel between steps.

await flow.run(shared);
await flow.run(shared, { userId: "123" });

const controller = new AbortController();
await flow.run(shared, undefined, { signal: controller.signal });

stream(shared, params?, options?)

An async-generator alternative to run() that yields StreamEvent values as the flow executes.

for await (const event of flow.stream(shared)) {
  if (event.type === "step:before") console.log("->", event.meta.index);
  if (event.type === "chunk") process.stdout.write(event.chunk as string);
  if (event.type === "done") break;
}

Steps emit chunks by assigning to shared.__stream:

.then(async (s) => {
  for await (const token of llmStream()) {
    s.__stream = token; // -> yields { type: "chunk", chunk: token }
  }
})

| Event type | Extra fields | When emitted | | ------------- | ---------------- | --------------------------------------- | | step:before | meta | Before each step | | step:after | meta, shared | After each step completes | | chunk | meta, chunk | When a step writes to shared.__stream | | error | meta, error | When a step throws | | done | shared | After the flow finishes |

Step options

Any step that accepts options supports:

| Option | Default | Description | | ----------- | ------- | ------------------------------------------------------ | | retries | 1 | Number of attempts before throwing | | delaySec | 0 | Seconds to wait between retries | | timeoutMs | 0 | Milliseconds before the step is aborted (0 = no limit) |

Error handling

When a step throws, the error is wrapped in a FlowError with the step index and type:

import { FlowBuilder, FlowError } from "flowneer";

try {
  await new FlowBuilder()
    .startWith(async () => {})
    .then(async () => {
      throw new Error("boom");
    })
    .run({});
} catch (err) {
  if (err instanceof FlowError) {
    console.log(err.step); // "step 1"
    console.log(err.cause); // Error: boom
  }
}

InterruptError is a special error that bypasses FlowError wrapping and propagates directly to the caller. Use it for human-in-the-loop patterns via withInterrupts or withHumanNode.

Plugins

The core is intentionally small. Use FlowBuilder.extend([...plugins]) to create a subclass with plugins mixed in. Unlike the removed use(), extend() never mutates the base class — each call returns an isolated subclass.

Using a plugin

import { FlowBuilder } from "flowneer";
import { withTiming } from "flowneer/plugins/observability";
import { withRateLimit } from "flowneer/plugins/llm";

const AppFlow = FlowBuilder.extend([withTiming, withRateLimit]);

const flow = new AppFlow<State>()
  .withTiming()
  .withRateLimit({ intervalMs: 500 })
  .startWith(step1)
  .then(step2);

Chain extend() calls to layer plugins on top of a base subclass:

const BaseFlow = FlowBuilder.extend([withTiming]);
const TracedFlow = BaseFlow.extend([withTrace]); // has both plugins

Writing a plugin

A plugin is an object of functions that get mixed onto FlowBuilder.prototype. Each function receives the builder as this and should return this for chaining.

import type {
  FlowBuilder,
  FlowneerPlugin,
  StepFilter,
  StepMeta,
} from "flowneer";

declare module "flowneer" {
  interface FlowBuilder<S, P> {
    withTracing(
      fn: (meta: StepMeta, event: string) => void,
      filter?: StepFilter,
    ): this;
  }
}

export const tracingPlugin: FlowneerPlugin = {
  withTracing(this: FlowBuilder<any, any>, fn, filter?: StepFilter) {
    (this as any)._setHooks(
      {
        beforeStep: (meta: StepMeta) => fn(meta, "before"),
        afterStep: (meta: StepMeta) => fn(meta, "after"),
        onError: (meta: StepMeta) => fn(meta, "error"),
      },
      filter,
    );
    return this;
  },
};

Lifecycle hooks

Plugins register hooks via _setHooks(). Multiple registrations of the same hook compose — the first registered is the outermost.

| Hook | Called | Arguments | | ---------------- | --------------------------------------------------------- | --------------------------------------- | | beforeFlow | Once before the first step | (shared, params) | | beforeStep | Before each step executes | (meta, shared, params) | | wrapStep | Wraps step execution — call next() to run the step body | (meta, next, shared, params) | | afterStep | After each step completes | (meta, shared, params) | | wrapParallelFn | Wraps each individual fn inside a parallel() step | (meta, fnIndex, next, shared, params) | | onError | When a step throws (before re-throwing) | (meta, error, shared, params) | | afterFlow | After the flow finishes (success or failure) | (shared, params) |

Step-scoped hooks (beforeStep, afterStep, onError, wrapStep, wrapParallelFn) accept an optional StepFilter as the second argument to _setHooks(). beforeFlow / afterFlow are unaffected. Unmatched wrapStep/wrapParallelFn hooks always call next() automatically so the middleware chain is never broken.

StepFilter

type StepFilter = string[] | ((meta: StepMeta) => boolean);

• String array — matches steps by label. Supports * as a glob wildcard ("llm:*" matches "llm:summarise", "llm:embed", …). Steps without a label are never matched.
• Predicate — return true to match. Use this for runtime conditions or multi-criteria logic.

// Array form with glob
flow.addHooks({ beforeStep: log }, ["llm:*", "embed:*"]);

// Predicate form
flow.addHooks(
  { beforeStep: log },
  (meta) => meta.label?.startsWith("llm:") ?? false,
);

addHooks(hooks, filter?) returns a dispose() function to remove the hooks.

Available plugins

All plugins are imported from flowneer/plugins or their individual subpath (e.g. flowneer/plugins/resilience).

Observability

| Plugin | Method | Description | | ---------------- | -------------------------- | -------------------------------------------------------------------------------------------------- | | withHistory | .withHistory() | Appends a shallow state snapshot after each step to shared.__history | | withTiming | .withTiming() | Records wall-clock duration (ms) of each step in shared.__timings[index] | | withVerbose | .withVerbose() | Prints the full shared object to stdout after each step | | withInterrupts | .interruptIf(condition) | Throws an InterruptError (with a deep-clone of shared) when condition is true | | withCallbacks | .withCallbacks(handlers) | LangChain-style lifecycle callbacks dispatched by step label prefix (llm:*, tool:*, agent:*) |

Persistence

| Plugin | Method | Description | | ------------------------- | --------------------------------- | --------------------------------------------------------------------------------------------------- | | withCheckpoint | .withCheckpoint(store) | Saves shared to a store after each successful step | | withAuditLog | .withAuditLog(store) | Writes an immutable deep-clone audit entry to a store after every step (success and error) | | withReplay | .withReplay(fromStep) | Skips all steps before fromStep; combine with .withCheckpoint() to resume a failed flow | | withVersionedCheckpoint | .withVersionedCheckpoint(store) | Diff-based versioned checkpoints with parent pointers; use .resumeFrom(version, store) to restore |

Resilience

| Plugin | Method | Description | | -------------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------- | | withCircuitBreaker | .withCircuitBreaker(opts?) | Opens the circuit after maxFailures consecutive failures and rejects all steps until resetMs elapses | | withFallback | .withFallback(fn) | Catches any step error and calls fn instead of propagating | | withTimeout | .withTimeout(ms) | Aborts any step that exceeds ms milliseconds | | withCycles | .withCycles(n, anchor?) | Throws after n anchor jumps globally, or after n visits to a named anchor — guards against infinite loops |

Messaging

| Plugin | Method | Description | | -------------- | ----------------- | ------------------------------------------------------------------------------------------------------- | | withChannels | .withChannels() | Map-based message-channel system on shared.__channels; use sendTo / receiveFrom / peekChannel | | withStream | .withStream() | Enables real-time chunk streaming via shared.__stream |

LLM

| Plugin | Method | Description | | ---------------------- | -------------------------------- | ------------------------------------------------------------------------ | | withCostTracker | .withCostTracker() | Accumulates per-step shared.__stepCost values into shared.__cost | | withRateLimit | .withRateLimit({ intervalMs }) | Enforces a minimum gap of intervalMs ms between steps | | withTokenBudget | .withTokenBudget(limit) | Aborts the flow before any step where shared.tokensUsed >= limit | | withStructuredOutput | .withStructuredOutput(opts) | Parses and validates shared.__llmOutput via a Zod-compatible validator |

Tools

| Plugin | Method | Description | | ----------- | ---------------------- | ----------------------------------------------------------------------------------------- | | withTools | .withTools(registry) | Attaches a ToolRegistry to shared.__tools; use executeTool / executeTools helpers |

Agent

| Plugin | Method | Description | | --------------- | ---------------------- | --------------------------------------------------------------------------------------------------------- | | withReActLoop | .withReActLoop(opts) | Built-in ReAct loop: think -> tool-call -> observe, with configurable maxIterations and onObservation | | withHumanNode | .humanNode(opts?) | Inserts a human-in-the-loop pause; pair with resumeFlow() to continue after receiving input |

Memory

| Plugin | Method | Description | | ------------ | ----------------------- | -------------------------------------------------------------------------------------------------------------- | | withMemory | .withMemory(instance) | Attaches a Memory instance to shared.__memory; choose BufferWindowMemory, SummaryMemory, or KVMemory |

Output

Pure parsing helpers — no plugin registration needed. Import from flowneer/plugins/output.

| Function | Description | | -------------------- | ------------------------------------------------------------- | | parseJsonOutput | Parse raw JSON, fenced, or embedded JSON from LLM text | | parseListOutput | Parse dash, *, bullet, numbered, or newline-separated lists | | parseMarkdownTable | Parse GFM tables to Record<string, string>[] | | parseRegexOutput | Extract named or positional regex capture groups |

Eval

| Export | Description | | ---------------------------------------- | ----------------------------------------------------------------- | | runEvalSuite | Run a flow against a labelled dataset and collect per-item scores | | exactMatch | Scorer: exact string equality | | containsMatch | Scorer: substring containment | | f1Score | Scorer: token-level F1 | | retrievalPrecision / retrievalRecall | Scorer: retrieval quality metrics | | answerRelevance | Scorer: relevance signal |

Graph

| Plugin | Method | Description | | ----------- | -------------- | ------------------------------------------------------------------------------------------------------------ | | withGraph | .withGraph() | Describe a flow as a DAG with .addNode() / .addEdge(), then .compile() to a ready-to-run FlowBuilder |

Telemetry

| Plugin | Method | Description | | --------------- | ----------------------- | ---------------------------------------------------------------------------------------------------------------- | | withTelemetry | .withTelemetry(opts?) | Structured span telemetry via TelemetryDaemon; accepts consoleExporter, otlpExporter, or a custom exporter |

Dev

| Plugin | Method | Description | | ------------------- | ----------------------------------------- | ------------------------------------------------------------------------------------------- | | withDryRun | .withDryRun() | Skips all step bodies while still firing hooks — useful for validating observability wiring | | withMocks | .withMocks(map) | Replaces step bodies at specified indices with mock functions | | withStepLimit | .withStepLimit(max?) | Throws after max total step executions (default 1000) | | withAtomicUpdates | .parallelAtomic(fns, reducer, options?) | Sugar over parallel() with a reducer — each fn runs on an isolated draft |

Presets

Presets are ready-made FlowBuilder factories for common patterns. Import from flowneer/presets or their individual subpath.

Agent presets (flowneer/presets/agent)

| Preset | Description | | -------------------- | ---------------------------------------------------------------------------------- | | createAgent | LangChain-style factory — wire up tools and an LLM adapter to get a runnable agent | | withReActLoop | ReAct think -> tool-call -> observe loop with configurable max iterations | | supervisorCrew | Supervisor dispatches to parallel worker agents, with an optional aggregator step | | sequentialCrew | Agents run in sequence, each receiving the output of the previous | | hierarchicalCrew | Tree-structured multi-agent delegation | | roundRobinDebate | Agents take turns responding for N rounds | | planAndExecute | Planner LLM produces a step-by-step plan; executor LLM carries out each step | | reflexionAgent | Generate -> critique -> revise loop (Reflexion paper) | | critiqueAndRevise | Two-agent generate -> critique -> revise loop | | evaluatorOptimizer | DSPy-style generate -> evaluate -> improve loop | | selfConsistency | Parallel sampling + majority-vote aggregation | | tool | Minimal tool-calling agent helper |

Pipeline presets (flowneer/presets/pipeline)

| Preset | Description | | -------------------- | ------------------------------------------------------------------------ | | generateUntilValid | Generate -> validate -> retry with error context until output passes | | mapReduceLlm | Batch LLM calls across N items, then reduce results into a single output |

RAG presets (flowneer/presets/rag)

| Preset | Description | | -------------- | --------------------------------------------------------- | | ragPipeline | Standard retrieve -> augment -> generate pipeline | | iterativeRag | RAG with follow-up retrieval loop for multi-hop questions |

Config presets (flowneer/presets/config)

| Preset | Description | | ------- | ---------------------------------------------------------------- | | build | Compile a FlowConfig JSON/object into a runnable FlowBuilder |

License

MIT