Skip to main content

Orchestration Guide

A walkthrough of the three graph builders in @framers/agentos/orchestration: AgentGraph, workflow() and mission(). Each compiles to the same CompiledExecutionGraph IR and runs on the same GraphRuntime. The reference pages cover each one in full: AgentGraph, workflow() DSL, mission() API and Checkpointing.

Every compiled graph exposes its IR (toIR(), and toWorkflow() on a mission), and any IR can run inside another graph as a subgraph node.

How a Run Works​

Executors come from the host​

The runtime calls no model and no tool of its own. compile({ deps }) passes the node executors to it (WorkflowRuntimeDeps), and each node type needs its own:

Node typeNeedsWithout it
gmideps.loopController and deps.providerCallSucceeds with the output 'gmi-placeholder'
tooldeps.toolOrchestratorFails (success: false)
guardraildeps.guardrailEnginePasses
subgraphdeps.subgraphResolver and deps.createSubgraphRuntimeSucceeds with the output 'subgraph-placeholder'
extensiondeps.extensionExecutorSucceeds with the output 'extension-not-configured'
voicedeps.voiceExecutorFails
router, humanNothing

The examples on this page share one set of host bindings:

import { LoopController } from '@framers/agentos/orchestration';
import type { WorkflowRuntimeDeps } from '@framers/agentos/orchestration/builders/WorkflowBuilder';

// Your application's own tool runner and model call.
declare function runMyTool(name: string, args: Record<string, unknown>): Promise<unknown>;
declare function callMyModel(instructions: string, context: unknown): Promise<string>;

const deps: WorkflowRuntimeDeps = {
  toolOrchestrator: {
    async processToolCall({ toolCallRequest }) {
      const output = await runMyTool(toolCallRequest.toolName, toolCallRequest.arguments);
      return { success: true, output };
    },
  },
  loopController: new LoopController(),
  async *providerCall(instructions, state) {
    // state.artifacts holds the outputs of the nodes that ran before this one;
    // state.scratch holds what extension, subgraph and voice nodes wrote there.
    const text = await callMyModel(instructions, {
      input: state.input,
      scratch: state.scratch,
      artifacts: state.artifacts,
    });
    yield { type: 'text_delta', content: text };
    // No tool calls: the node's loop ends after this turn.
    return { responseText: text, toolCalls: [], finishReason: 'stop' };
  },
};

A gmi node runs LoopController's loop for up to maxInternalIterations turns (default 10). Each turn calls providerCall(instructions, state); when the returned toolCalls is not empty, the loop runs those calls through deps.toolOrchestrator and calls providerCall again with the same arguments, so the host keeps the conversation and the tool results itself. The node's output is the text of all its turns.

State and outputs​

  • invoke(input) freezes input into state.input. state.scratch and state.artifacts start empty. The Zod schemas passed to a builder are lowered to JSON Schema and stored in the IR; the runtime neither validates against them nor applies their defaults.
  • A node's output goes into state.artifacts under the node's id (a workflow() step's outputAs names another key). invoke() returns state.artifacts. A gmi node's output is its text, a tool node's is the tool's output.
  • scratch changes only through executors that return a scratch update: a subgraph node's outputMapping, an extension node whose output is an object, and a voice node (its checkpoint). Reducers merge those updates. Every key of an update, __proto__ included (an output parsed from JSON can carry one), becomes an own key of scratch; no update sets the prototype of scratch.
  • A tool node sends the tool the static args from toolNode(name, { args }) and nothing from the graph state.

Scheduling​

  • A node runs when every node with an edge into it has completed or been skipped, and it runs at most once per run. A router marks the targets it did not pick skipped; skipping goes no further, so a node whose incoming edges all come from skipped nodes still runs.
  • Nodes that become ready together run concurrently, each on its own copy of the state. After such a batch the runtime merges the keys each node wrote to scratch and artifacts, in the order the nodes are declared: a reducer registered for the key (scratch.<key> or artifacts.<key>) combines each write with the value the key held before it, as when the nodes run one after another, so the value from before the batch counts once; without one the node declared later wins. A node's copy of a key it did not write is not merged, so it never undoes another node's write. The outputs of every node in the batch reach the nodes after it and the result.
  • A node that fails stops the run: the runtime saves a checkpoint and emits error, interrupt and run_end (see Checkpointing). toolNode(name, { retryPolicy }) and a workflow() step's retryPolicy re-run a failed node first. A node that fails writes nothing to scratch or artifacts and records no result, so a resume runs it again. What a failed attempt did outside the graph, such as a call a tool made before it failed, is not undone, and the node's next run repeats it, so a write or external node should be safe to repeat.
  • The events of a batch are yielded once every node of it has run, in the order the nodes are declared. When a node of the batch interrupts or fails, the run reports every node of the batch that completed, then that node and the stop. When the store refuses a checkpoint a node of the batch saves, or a function on one of its edges throws, the other nodes of the batch finish first, and the run then ends with that error and yields none of the batch's events (see Checkpointing).

AgentGraph​

Use AgentGraph when you lay out the nodes and edges yourself.

Minimal example​

import { AgentGraph, START, END, gmiNode, toolNode } from '@framers/agentos/orchestration';
import { z } from 'zod';

const graph = new AgentGraph({
  input:     z.object({ topic: z.string() }),
  scratch:   z.object({}),
  artifacts: z.object({}),
})
  .addNode('search',    toolNode('web_search', { args: { query: 'quantum computing' } }))
  .addNode('summarize', gmiNode({ instructions: 'Summarize the search results in 3 sentences.' }))
  .addEdge(START, 'search')
  .addEdge('search', 'summarize')
  .addEdge('summarize', END)
  .compile({ deps });

const result = await graph.invoke({ topic: 'quantum computing' });
console.log(result.summarize); // the summary text; result.search holds the tool output

compile() validates the graph and throws on an edge to an unknown node, a missing entry or exit edge, or a node that cannot be reached from START.

Routing on a node's output​

A routerNode returns the id of the node to run next. Give it a static edge to every node it can return; the runtime follows the one it names and marks the others skipped.

import { AgentGraph, START, END, gmiNode, routerNode } from '@framers/agentos/orchestration';
import { z } from 'zod';

const labels = ['factual', 'creative', 'code'];

const graph = new AgentGraph({
  input:     z.object({ query: z.string() }),
  scratch:   z.object({}),
  artifacts: z.object({}),
})
  .addNode('classify', gmiNode({
    instructions: 'Classify the query as "factual", "creative", or "code". Reply with only the label.',
  }))
  .addNode('route', routerNode((state) => {
    const label = String(state.artifacts.classify ?? '').toLowerCase();
    return labels.find((id) => label.includes(id)) ?? 'creative';
  }))
  .addNode('factual',  gmiNode({ instructions: 'Answer with facts and name your sources.' }))
  .addNode('creative', gmiNode({ instructions: 'Write a creative, engaging response.' }))
  .addNode('code',     gmiNode({ instructions: 'Write clean, documented code with an explanation.' }))
  .addEdge(START, 'classify')
  .addEdge('classify', 'route')
  .addEdge('route', 'factual')
  .addEdge('route', 'creative')
  .addEdge('route', 'code')
  .addEdge('factual',  END)
  .addEdge('creative', END)
  .addEdge('code',     END)
  .compile({ deps });

const result = await graph.invoke({ query: 'Write a debounce function in TypeScript' });
// { classify: 'code', code: '...' }

addConditionalEdge(source, fn) stores its target as the placeholder __CONDITIONAL__, and the scheduler counts that placeholder as an edge from the source to every node in the graph, the source included. The source then waits on itself and every other node waits on the source, so no node runs and invoke() returns {}. Route with a routerNode as above.

Loops​

A node runs at most once per run, and a node in a cycle waits for the node before it in the same cycle, so a cycle never starts. compile() accepts cycles, and the run ends when only the nodes of the cycle and the nodes after it are left. Repeated work belongs inside a node, in the gmi node's tool loop (maxInternalIterations), or in a new run.

Subgraphs​

subgraphNode(ir, { inputMapping, outputMapping }) runs another compiled graph as one node. inputMapping maps paths in the parent's scratch to paths in the child's input, and outputMapping maps paths in the child's artifacts to paths in the parent's scratch; the child's artifacts are also the node's output. The node looks the child up with deps.subgraphResolver(graphId) and runs it on the runtime that deps.createSubgraphRuntime() returns.

import {
  AgentGraph, START, END, gmiNode, subgraphNode,
  GraphRuntime, NodeExecutor, InMemoryCheckpointStore,
} from '@framers/agentos/orchestration';
import { z } from 'zod';

const research = new AgentGraph({
  input: z.object({}), scratch: z.object({}), artifacts: z.object({}),
})
  .addNode('notes', gmiNode({ instructions: 'Collect research notes on fusion energy.' }))
  .addEdge(START, 'notes')
  .addEdge('notes', END)
  .compile({ deps });
const researchIR = research.toIR();

const production = new AgentGraph({
  input: z.object({ topic: z.string() }), scratch: z.object({}), artifacts: z.object({}),
})
  .addNode('research', subgraphNode(researchIR, { outputMapping: { notes: 'notes' } }))
  .addNode('write',    gmiNode({ instructions: 'Write a blog post from the research notes.' }))
  .addEdge(START, 'research')
  .addEdge('research', 'write')
  .addEdge('write', END)
  .compile({
    deps: {
      ...deps,
      subgraphResolver: (graphId) => (graphId === researchIR.id ? researchIR : undefined),
      createSubgraphRuntime: () => new GraphRuntime({
        checkpointStore: new InMemoryCheckpointStore(),
        nodeExecutor: new NodeExecutor(deps),
      }),
    },
  });

const result = await production.invoke({ topic: 'fusion energy' });
// result.research = the child's artifacts ({ notes: '...' }); result.write = the post

Node options and policies​

gmiNode(
  {
    instructions: 'Summarize the document.',
    maxInternalIterations: 5, // read: caps the node's loop (default 10)
    parallelTools: false,     // read: runs a turn's tool calls together when true
    executionMode: 'react_bounded', // stored in the IR, not read
    temperature: 0.3,               // stored in the IR, not passed to providerCall
    maxTokens: 2048,                // stored in the IR, not passed to providerCall
  },
  {
    checkpoint: 'after',      // read: save a checkpoint after the node
    effectClass: 'read',      // read: on resume, a 'write', 'external' or 'human' node keeps its recorded output
    memory: { consistency: 'snapshot', read: { types: ['semantic'], maxTraces: 10 } }, // stored, not read
    guardrails: { output: ['pii-redaction'], onViolation: 'sanitize' },               // stored, not read
  },
)

The runtime reads a node's checkpoint flag and its effectClass. It does not read the memory, discovery, persona or guardrails policies; a guardrailNode is the step that checks content during a run, and a humanNode runs pii-redaction and code-safety through deps.guardrailEngine after an auto-accept, a judge's approval or a timeout accept, unless its guardrailOverride is false.

workflow()​

workflow() builds a graph from steps declared in order. compile() throws when .input() or .returns() is missing and when the graph has a cycle, and it always checkpoints after every node.

Quick start​

import { workflow } from '@framers/agentos/orchestration';
import { z } from 'zod';

// Your application's own page fetcher.
declare function fetchPage(url: string): Promise<string>;

const pipeline = workflow('content-pipeline')
  .input(z.object({ url: z.string() }))
  .returns(z.object({ summary: z.string(), tags: z.string() }))
  .step('fetch',     { extension: { extensionId: 'web', method: 'fetchPage' } })
  .step('summarize', { gmi: { instructions: 'Summarize the fetched page in 3 sentences.' }, outputAs: 'summary' })
  .step('tag',       { gmi: { instructions: 'Extract 5 topic tags as a JSON array.' }, outputAs: 'tags' })
  .compile({
    deps: {
      ...deps,
      // An extension step receives the run's input and scratch.
      async extensionExecutor(_extensionId, _method, args) {
        const { input } = args as { input: { url: string } };
        return { success: true, output: { page: await fetchPage(input.url) } };
      },
    },
  });

const result = await pipeline.invoke({ url: 'https://example.com/article' });
console.log(result.summary, result.tags); // result.fetch holds { page }

A tool step sends the tool no arguments: StepConfig has no field for them, so a tool step cannot read the run's input. An extension step can: deps.extensionExecutor receives { input, scratch }, and an object it returns is merged into scratch and kept as the step's artifact, so the summarize step's providerCall finds the page in state.artifacts.fetch and state.scratch.page. A gmi step is recorded as single_turn and runs the same loop as any gmi node, up to 10 turns while providerCall returns tool calls.

Branches​

.branch(condition, routes) takes a function and a map from route key to step config. The compiler adds a router node that runs the function, and a node per route.

workflow('triage')
  .input(z.object({ ticket: z.string() }))
  .returns(z.object({ response: z.string() }))
  .step('classify', { gmi: { instructions: 'Classify as "billing", "technical", or "general".' } })
  .branch(
    (state) => String(state.artifacts.classify ?? '').trim(),
    {
      billing:   { gmi: { instructions: 'Handle the billing issue.' } },
      technical: { gmi: { instructions: 'Diagnose the technical issue.' } },
      general:   { gmi: { instructions: 'Answer the general inquiry.' } },
    },
  )
  .compile({ deps });

The router node returns the route key itself, and the runtime takes a router's return value as the id of the node to run next. The route nodes have generated ids (branch-<key>-<n>), so no node matches: every route node is marked skipped and the run continues with the step declared after the branch. To route on a value, build the graph with AgentGraph and a routerNode that returns node ids.

Parallel steps​

.parallel(steps, join) adds one node per step config, all connected from the previous step, and registers join.merge as reducers. join.strategy, join.quorumCount and join.timeout are stored and not read.

// Your application's own search clients.
declare function searchWeb(query: string): Promise<string[]>;
declare function searchNews(query: string): Promise<string[]>;
declare function searchArxiv(query: string): Promise<string[]>;
const search: Record<string, (query: string) => Promise<string[]>> = {
  web: searchWeb, news: searchNews, arxiv: searchArxiv,
};

workflow('multi-source-research')
  .input(z.object({ query: z.string() }))
  .returns(z.object({ report: z.string() }))
  .parallel(
    [
      { extension: { extensionId: 'search', method: 'web' } },
      { extension: { extensionId: 'search', method: 'news' } },
      { extension: { extensionId: 'search', method: 'arxiv' } },
    ],
    { strategy: 'all', merge: { 'scratch.results': 'concat' } },
  )
  .step('synthesize', { gmi: { instructions: 'Synthesize the sources into a report.' }, outputAs: 'report' })
  .compile({
    deps: {
      ...deps,
      async extensionExecutor(_extensionId, method, args) {
        const { input } = args as { input: { query: string } };
        return { success: true, output: { results: await search[method](input.query) } };
      },
    },
  });

The three extension nodes become ready together and run concurrently. Each returns { results }, which goes into its branch's scratch and, as the node's output, into its artifact. After the batch the concat reducer joins the three arrays into scratch.results in the order the steps are declared, and each node's artifact stays under its id (see Scheduling), so synthesize, which runs after all three, finds the results in state.scratch.results and in state.artifacts. Parallel tool steps reach synthesize through state.artifacts alone: a tool node's output goes only to its artifact.

Human step​

workflow('content-approval')
  .input(z.object({ brief: z.string() }))
  .returns(z.object({ post: z.string() }))
  .step('draft',   { gmi: { instructions: 'Write a blog post draft from the brief.' } })
  .step('approve', { human: { prompt: 'Review the draft. Approve or request changes.' } })
  .step('publish', { tool: 'cms_publish', effectClass: 'external' })
  .compile({ deps });

The human step interrupts the run: the runtime saves a checkpoint, emits an interrupt event with the reason human_approval, and ends the run with run_end. resume(runId) marks the step complete with its recorded output ({ prompt }) and runs the rest; a host that has the reviewer's answer writes it into a fork of the checkpoint and resumes the fork (Checkpointing, Human-in-the-Loop).

A step's memory, discovery and guardrails policies are stored and not read; requiresApproval and onFailure are not read either, and retryPolicy re-runs a failed step whatever onFailure says.

mission()​

mission() builds a linear graph from a goal and a plan template (mission() API).

import { mission } from '@framers/agentos/orchestration';
import { z } from 'zod';

const research = mission('research')
  .input(z.object({ topic: z.string() }))
  .goal('Research quantum error correction and produce a concise 3-paragraph summary with citations.')
  .returns(z.object({ summary: z.string() }))
  .planner({ strategy: 'linear', maxSteps: 8, style: 'research' })
  .compile({ deps });

const artifacts = await research.invoke({ topic: 'quantum error correction' });
// outputs under gather-info, process-info, deliver-result, refine-output
  • planner.style picks the template (research, qa or creative); without it the compiler classifies the goal's wording. strategy and maxSteps are required and not read; a pre-built plan goes in planner.plan.
  • The goal is wrapped in <mission_goal> tags at the start of every step's instructions. {{variable}} placeholders are not filled.
  • .anchor(id, node, { phase, after }) splices a node of your own into the chain.
  • autonomy(), providerStrategy(), costCap(), maxAgents(), branches(), plannerModel() and executionModel() store values the compiler does not read.

Voice Nodes​

voiceNode(id, config) builds a voice node; .on(exitReason, target) maps the reason the voice session ended to the next node, and .build() returns the node.

import { AgentGraph, START, END, voiceNode, gmiNode, toolNode } from '@framers/agentos/orchestration';
import type { GraphNode } from '@framers/agentos/orchestration';
import { VoiceNodeExecutor } from '@framers/agentos/orchestration/runtime/VoiceNodeExecutor';
import { z } from 'zod';

// Your application's own telephony bridge: resolves to the call's voice transport.
declare function openCallTransport(callerId: string): Promise<unknown>;

// An extension node; the object its executor returns is merged into scratch.
const connect: GraphNode = {
  id: 'connect',
  type: 'extension',
  executorConfig: { type: 'extension', extensionId: 'telephony', method: 'connect' },
  executionMode: 'single_turn',
  effectClass: 'external',
  checkpoint: 'none',
};

const callGraph = new AgentGraph({
  input:     z.object({ callerId: z.string() }),
  scratch:   z.object({}),
  artifacts: z.object({}),
})
  .addNode('connect', connect)
  .addNode('listen', voiceNode('listen', { mode: 'conversation', maxTurns: 10 })
    .on('turns-exhausted', 'resolve')
    .on('hangup', 'cleanup')
    .build())
  .addNode('resolve', gmiNode({ instructions: 'Determine the resolution from the call transcript.' }))
  .addNode('cleanup', toolNode('close_ticket'))
  .addEdge(START, 'connect')
  .addEdge('connect', 'listen')
  .addEdge('listen', 'resolve')
  .addEdge('listen', 'cleanup')
  .addEdge('resolve', END)
  .addEdge('cleanup', END)
  .compile({
    deps: {
      ...deps,
      voiceExecutor: new VoiceNodeExecutor((event) => console.log(event.type)),
      // Puts the transport at state.scratch.voiceTransport, where the voice node reads it.
      async extensionExecutor(_extensionId, _method, args) {
        const { input } = args as { input: { callerId: string } };
        return { success: true, output: { voiceTransport: await openCallTransport(input.callerId) } };
      },
    },
  });

A conversation or listen-only node ends with the first of these exit reasons: hangup (the transport emits close or disconnected), turns-exhausted (maxTurns turns, when it is above 0), keyword:<word> (with exitOn: 'keyword', a final transcript containing one of exitKeywords), silence-timeout (with exitOn: 'silence-timeout', 30 seconds without speech) and interrupted (an AbortSignal placed at state.scratch.abortSignal fires; a barge-in emits voice_barge_in and does not end the node). A speak-only node delivers speakText and ends with completed. The executor returns the target mapped to the exit reason, and the runtime runs that node and skips the others. The node still needs a static edge to each target, because the scheduler and the validator read only edges. An exit reason with no mapping follows every static edge, and a mapping back to the voice node itself does not run it again.

The executor needs the voice transport at state.scratch.voiceTransport and fails without it. A run starts with an empty scratch and no builder puts a transport there (workflow().transport('voice', ...) stores its settings and nothing reads them), so the host writes it from an earlier node, as the connect extension node does above. The transport is also kept as the connect artifact, so invoke() returns it. Voice events (voice_session, voice_transcript, voice_barge_in, voice_turn_complete) go to the callback passed to VoiceNodeExecutor, not to the graph's event stream.

OptionTypeRead by the executor
mode'conversation' | 'listen-only' | 'speak-only'Yes (required)
maxTurnsnumberYes: ends the node with turns-exhausted (0 or unset = no limit)
exitOn'hangup' | 'silence-timeout' | 'keyword' | 'turns-exhausted' | 'manual''keyword' and 'silence-timeout' add their exit condition; the other values add none
exitKeywordsstring[]Yes, with exitOn: 'keyword'
speakTextstringYes, on a speak-only node
stt, tts, voice, endpointing, bargeIn, diarization, languageNo: stored in the node's checkpoint and not applied

Checkpointing and Resume​

import { AgentGraph, InMemoryCheckpointStore } from '@framers/agentos/orchestration';

const store = new InMemoryCheckpointStore();

const graph = new AgentGraph(stateSchema, { checkpointPolicy: 'every_node' })
  // ...nodes and edges...
  .compile({ checkpointStore: store, deps });

let runId: string | undefined;
for await (const event of graph.stream({ topic: 'fusion energy' })) {
  if (event.type === 'run_start') runId = event.runId;
}

// After an interrupt or a failure: continue from the run's latest checkpoint.
const artifacts = await graph.resume(runId!);

// Fork a checkpoint with patched state and resume the fork.
const [latest] = await store.list(graph.toIR().id, { runId, limit: 1 });
const forkId = await store.fork(latest.id, { scratch: { approved: true } });
const forked = await graph.resume(forkId);

AgentGraph takes checkpointPolicy in its constructor ('none' by default); workflow() and mission() checkpoint after every node. Under every policy the runtime also saves a checkpoint when a node fails or interrupts the run. InMemoryCheckpointStore keeps checkpoints in process memory and takes no arguments. SqlCheckpointStore keeps them in SQLite or Postgres through @framers/sql-storage-adapter, so a run resumes after a restart; implement ICheckpointStore for any other durable store. Resume, forks and the policies are in Checkpointing.

Graph Events​

stream(input) yields GraphEvent values:

for await (const event of graph.stream({ topic: 'AI safety' })) {
  switch (event.type) {
    case 'node_start':
      console.log(`-> ${event.nodeId}`);
      break;
    case 'node_end':
      console.log(`ok ${event.nodeId} in ${event.durationMs}ms`);
      break;
    case 'interrupt':
      console.log(`paused at ${event.nodeId}: ${event.reason}`);
      break;
    case 'error':
      console.error(event.nodeId, event.error.code, event.error.message);
      break;
    case 'run_end':
      console.log('done', event.finalOutput);
      break;
  }
}
EventPayloadWhen
run_startrunId, graphIdThe run begins
node_startnodeId, state (input and scratch)A node starts, and again before each retry
node_endnodeId, output, durationMs, telemetryA node completes
edge_transitionsourceId, targetId, edgeTypeThe runtime follows an edge to a node
checkpoint_savedcheckpointId, nodeIdA checkpoint is saved
interruptnodeId, reason (human_approval, error, guardrail_violation)A human node pauses the run, or a node fails
node_timeoutnodeId, timeoutMsA node's timeout expired
errornodeId, error.code, error.messageA node failed (NODE_EXECUTION_FAILED or NODE_TIMEOUT)
guardrail:hitl-overridenodeId, guardrailId, reasonA human node's post-approval guardrails rejected the approval
run_endrunId, finalOutput (the artifacts), totalDurationMsThe run ends

A GraphRuntime built with an expansionHandler also emits mission:* events; none of the builders sets one. The checkpoint it saves before an expansion is described in Checkpointing. The other event types the union declares (text_delta, tool_call, tool_result, memory_read, memory_write, discovery_result, guardrail_result) are not emitted by the runtime.

Choosing the Right API​

If you need...Use
Known steps in a fixed orderworkflow()
Routing on a node's outputAgentGraph with a routerNode
A graph built from a goal and a plan templatemission()
Several agents working on one taskagency() (sequential, parallel, debate, review-loop, hierarchical, graph)
One model callgenerateText() / streamText()
A voice turn inside a graphAgentGraph with voiceNode