@oni-bot/core

The graph execution engine for agent swarms — build single agents or orchestrate multi-agent systems in TypeScript. Zero dependencies. Full TypeScript generics. Pregel-based superstep engine. 7 swarm templates.

Part of the ONI Platform (Open Neural Infrastructure).

What It Is

@oni-bot/core is a graph execution framework in TypeScript with production-ready swarm orchestration. Build single agents or multi-agent swarms with:

• State management — typed channels with pluggable reducers (lastValue, appendList, mergeObject, ephemeralValue)
• Graph execution — Pregel superstep model with parallel node execution, fan-out/fan-in, map-reduce
• 5 stream modes — values, updates, debug, messages (token-level), and custom events
• Checkpointing — Memory, SQLite, PostgreSQL, or custom backends with time travel and fork
• Human-in-the-loop — compile-time interrupts + runtime interrupt() with resume, typed input, approval, selection
• Messages — smart reducer with deduplication, RemoveMessage, UpdateMessage, filtering, trimming
• Cross-thread Store — namespaced KV store with semantic search for agent long-term memory
• Runtime context — getConfig(), getStore(), getStreamWriter() via AsyncLocalStorage
• Swarm orchestration — 7 swarm templates: hierarchical, fan-out, pipeline, peer-network, map-reduce, debate, hierarchical-mesh
• Functional API — task(), entrypoint(), pipe(), branch() as alternatives to the builder pattern
• Prebuilt agents — createReactAgent() with bring-your-own LLM
• Injected tools — tools that auto-receive state + store from runtime context
• Graph inspection — topology descriptors, Mermaid diagrams, cycle detection

Installation

npm install @oni-bot/core

Optional peer dependencies:

npm install better-sqlite3          # for SqliteCheckpointer
npm install pg                      # for PostgresCheckpointer

Quick Start

Minimal graph

import { StateGraph, START, END, lastValue, appendList } from "@oni-bot/core";

type MyState = { query: string; answer: string; log: string[] };

const graph = new StateGraph<MyState>({
  channels: {
    query:  lastValue(() => ""),
    answer: lastValue(() => ""),
    log:    appendList(() => []),
  },
});

graph.addNode("agent", async (state) => {
  return { answer: `Response to: ${state.query}`, log: ["agent ran"] };
});

graph.addEdge(START, "agent");
graph.addEdge("agent", END);

const app = graph.compile();
const result = await app.invoke({ query: "What is ONI?" });

Streaming

for await (const event of app.stream({ query: "hello" }, { streamMode: "updates" })) {
  console.log(event.node, event.data);
}

// Multiple modes simultaneously
for await (const evt of app.stream(input, { streamMode: ["values", "messages"] })) {
  if (evt.mode === "messages") process.stdout.write(evt.data.chunk);
}

Functional API

import { entrypoint, lastValue } from "@oni-bot/core";

const app = entrypoint(
  { channels: { query: lastValue(() => ""), answer: lastValue(() => "") } },
  async (state) => ({ answer: await myLLM(state.query) })
);

const result = await app.invoke({ query: "hello" });

Swarm Templates

Build multi-agent systems with pre-wired templates. Each template configures routing, coordination, and error handling automatically.

Hierarchical (Supervisor -> Workers)

import { SwarmGraph } from "@oni-bot/core/swarm";

const swarm = SwarmGraph.hierarchical<MyState>({
  supervisor: { model: myModel, strategy: "llm", maxRounds: 10 },
  agents: [researcher, writer, critic],
  onError: "fallback",
});

const app = swarm.compile();
const result = await app.invoke({ task: "Write a blog post about AI" });

All 7 Templates

| Template | Use case | Pattern | |---|---|---| | SwarmGraph.hierarchical() | Supervisor routes to workers | Supervisor -> Agent -> Supervisor -> END | | SwarmGraph.fanOut() | Parallel execution + aggregation | Send to all -> Collect -> Reduce | | SwarmGraph.pipeline() | Sequential processing chain | A -> B -> C -> END | | SwarmGraph.peerNetwork() | Agents hand off to each other | Dynamic peer-to-peer routing | | SwarmGraph.mapReduce() | Distribute N items across pool | Split -> AgentPool -> Collect -> Reduce | | SwarmGraph.debate() | Multi-round argumentation | Judge -> Debaters -> Judge -> consensus? | | SwarmGraph.hierarchicalMesh() | Nested supervisor teams | Coordinator -> Team subgraphs -> Coordinator |

See examples/swarm/ for complete runnable examples of each template.

Single Agent — No Swarm Required

The swarm layer is purely additive. If you just need one agent:

import { defineAgent } from "@oni-bot/core/agents";
import { anthropic } from "@oni-bot/core/models";

const agent = defineAgent({
  name: "assistant",
  model: anthropic("claude-sonnet-4-6"),
  tools: [webSearch],
  systemPrompt: "You are a helpful assistant.",
});

const result = await agent.invoke({ messages: [{ role: "user", content: "Hello!" }] });

Entry points at every level: ONIModel -> defineAgent() -> StateGraph -> SwarmGraph.

Features at a Glance

| Feature | Description | Example | Guide | |---|---|---|---| | Channels | Typed state with pluggable reducers | examples/ephemeral-channels.ts | Section 2 | | Command routing | State update + routing in one return | examples/command-routing.ts | Section 3 | | Token streaming | messages mode + getStreamWriter() | examples/messages-stream.ts | Section 4 | | Multi-stream | Multiple stream modes simultaneously | examples/multi-stream.ts | Section 4 | | Custom events | Emit named events via StreamWriter | examples/custom-stream.ts | Section 4 | | Messages | Smart reducer, helpers, RemoveMessage | examples/messages-reducer.ts | Section 5 | | Checkpointing | Memory, SQLite, PostgreSQL backends | examples/time-travel.ts | Section 6 | | HITL interrupts | interrupt(), getUserApproval() | examples/hitl/ | Section 7 | | Dynamic interrupts | Runtime breakpoint conditions | examples/dynamic-interrupt.ts | Section 7 | | Cross-thread Store | Namespaced KV with semantic search | examples/store-memory.ts | Section 8 | | Subgraphs | Nested skeletons + Command.PARENT | examples/subgraph.ts | Section 9 | | Parallel fan-out | Static + dynamic (Send API) | examples/parallel-fanout.ts | Section 10 | | Map-reduce | Send + fan-in barrier | examples/map-reduce.ts | Section 10 | | Runtime context | getConfig(), getStore(), etc. | examples/runtime-context.ts | Section 11 | | Retry + cache | Per-node retry policy with backoff | examples/retry-policy.ts | Section 12 | | Functional API | task, entrypoint, pipe, branch | examples/functional-api.ts | Section 13 | | ReAct agent | Prebuilt agent loop with tools | examples/react-agent.ts | Section 14 | | Graph inspection | Topology, Mermaid, cycle detection | examples/graph-inspection.ts | Section 16 | | Time travel | getHistory, getStateAt, forkFrom | examples/time-travel.ts | Section 17 | | Swarm | 7 templates + Supervisor, Handoff, retry-fallback, coordination | examples/swarm/ | Section 18 |

Core Concepts

Channels

Every state field has a channel — a reducer + default factory that controls how concurrent updates merge.

import { lastValue, appendList, mergeObject, ephemeralValue } from "@oni-bot/core";

const channels = {
  query:    lastValue(() => ""),          // last write wins
  messages: appendList(() => []),         // arrays concatenate
  context:  mergeObject(() => ({})),      // shallow merge
  scratch:  ephemeralValue(() => null),   // resets each superstep
};

Nodes, edges, compile

graph.addNode("agent", async (state, config?) => {
  return { answer: "partial update" };  // or Command, or void
});

graph.addEdge(START, "agent");
graph.addConditionalEdges("agent", (state) => state.done ? END : "tools");

const app = graph.compile({ checkpointer, store });

Invoke / Stream / Batch

const result  = await app.invoke(input, { threadId: "t1" });
const stream  = app.stream(input, { streamMode: "updates" });
const results = await app.batch([input1, input2]);

For the full progressive tutorial, see docs/GUIDE.md.

Sub-module Exports

Tree-shakeable sub-module imports for bundle optimization:

| Import path | Contents | |---|---| | @oni-bot/core | Everything (116+ exports) | | @oni-bot/core/prebuilt | createReactAgent, createToolNode, toolsCondition, types | | @oni-bot/core/swarm | SwarmGraph (7 templates), AgentRegistry, AgentPool, Handoff, Supervisor, Mailbox, coordination | | @oni-bot/core/hitl | interrupt, getUserInput, getUserApproval, getUserSelection, session store | | @oni-bot/core/store | BaseStore, InMemoryStore, NamespacedStore, AgentMemoryStore | | @oni-bot/core/messages | messagesChannel, messagesReducer, helpers, RemoveMessage, UpdateMessage | | @oni-bot/core/checkpointers | SqliteCheckpointer, PostgresCheckpointer, NamespacedCheckpointer | | @oni-bot/core/functional | task, entrypoint, pipe, branch | | @oni-bot/core/inspect | buildGraphDescriptor, toMermaidDetailed | | @oni-bot/core/streaming | emitToken, TokenStreamWriter, StreamWriterImpl | | @oni-bot/core/models | ONIModel, anthropic(), openai() model factories | | @oni-bot/core/agents | defineAgent, agent(), AgentContext, types | | @oni-bot/core/coordination | RequestReplyBroker, PubSub coordination primitives | | @oni-bot/core/guardrails | InputGuardrail, OutputGuardrail, guardrail types | | @oni-bot/core/tools | defineTool, createInjectedTool, tool types |

Architecture

@oni-bot/core v0.6.0
├── StateGraph / MessageGraph     ← fluent graph builder
│   ├── addNode / addSubgraph     ← nodes are async functions or compiled skeletons
│   ├── addEdge / addConditional  ← static + dynamic routing
│   └── compile()
│       └── ONIPregelRunner       ← superstep execution engine
│           ├── parallel node execution (Promise.all per superstep)
│           ├── channel reducers (state merging)
│           ├── edge resolution + Command routing
│           ├── Send API (dynamic fan-out)
│           ├── checkpointing (Memory / SQLite / Postgres / custom)
│           ├── interrupt handling (boundary + in-node)
│           ├── retry engine (exponential backoff)
│           ├── runtime context (AsyncLocalStorage)
│           └── stream writer (tokens + custom events + messages)
│
├── Messages         ← smart reducer, dedup, RemoveMessage, UpdateMessage
├── HITL             ← interrupt(), getUserInput, sessions, resume
├── Store            ← BaseStore, InMemoryStore, NamespacedStore, AgentMemoryStore
├── Swarm            ← SwarmGraph (7 templates), Supervisor, Handoff, AgentPool, coordination
│   ├── Templates    ← hierarchical, fanOut, pipeline, peerNetwork, mapReduce, debate, hierarchicalMesh
│   ├── Coordination ← RequestReplyBroker, PubSub (lazy auto-wired)
│   └── Retry        ← retry-then-fallback error recovery
├── Functional       ← task, entrypoint, pipe, branch
├── Prebuilt         ← createReactAgent, createToolNode, toolsCondition
├── Inspect          ← graph descriptors, Mermaid, cycle detection
├── Injected Tools   ← createInjectedTool (state + store auto-injected)
├── Stream Events    ← streamEvents v2 protocol
└── Errors           ← ONIError hierarchy, ONIInterrupt

ONI Platform

@oni-bot/core is the foundation layer. Other ONI packages build on top:

| Package | Built on | |---|---| | @oni/agentOS | AgentOS ADE — multi-agent orchestration | | @oni/vectorforge | VectorForge — knowledge base & SOP retrieval | | @oni/cic | CIC Agent Assist — call center intelligence | | @oni/oats | OATS — Five9 AI Agent Assist integration |

Documentation

• Developer Guide — Progressive tutorial from zero to advanced (19 sections)
• API Reference — Complete reference for all 116+ public exports
• Examples — 30+ runnable example files covering every feature

License

MIT — ONI Platform