Skip to content
Kindgi

Run steps in parallel

View as Markdown

This guide runs several tools on the same input at the same time, with a fanout step, and chooses how their results come back.

A fanout step runs several tools at once, on the same input, and collects their outputs into one.

Before you start

Section titled “Before you start”
  • A pack with the tools from Write a flow, such as acme.get-order, on a runtime such as kindgi dev, with Kindgi 0.1's CLI.

Fan out to several tools

Section titled “Fan out to several tools”

Ask your coding agent: "I want a flow that scores an order's risk and quotes its shipping at the same time." Kindgi's skills in your pack tell it how (How it knows Kindgi). For reference, the tools and the flow look like this. This guide calls the flow acme.assess-order, and it uses two new read-only tools:

tools/score-risk/index.ts
import { defineTool } from '@kindgi/sdk/define';
import type { ToolId } from '@kindgi/sdk/types';
import { z } from 'zod';


const defined = defineTool({
  id: 'acme.score-risk' as ToolId,
  description: 'Scores the fraud risk of an order.',
  version: '0.1.0',
  input: z.object({ total: z.number() }),
  output: z.object({ level: z.enum(['low', 'high']) }),
  effects: [],
  mutating: false,
  handler: async ({ total }) => ({ level: total > 1000 ? ('high' as const) : ('low' as const) }),
});


if (defined.kind === 'err') throw new Error(defined.error.message);


export default defined.value;
tools/quote-shipping/index.ts
import { defineTool } from '@kindgi/sdk/define';
import type { ToolId } from '@kindgi/sdk/types';
import { z } from 'zod';


const defined = defineTool({
  id: 'acme.quote-shipping' as ToolId,
  description: 'Quotes how many days an order takes to ship.',
  version: '0.1.0',
  input: z.object({ items: z.array(z.object({ sku: z.string(), quantity: z.number() })) }),
  output: z.object({ days: z.number().int() }),
  effects: [],
  mutating: false,
  handler: async ({ items }) => ({ days: 2 + items.length }),
});


if (defined.kind === 'err') throw new Error(defined.error.message);


export default defined.value;
flows/assess-order/index.ts
import { defineFlow } from '@kindgi/sdk/define';


const defined = defineFlow({
  id: 'acme.assess-order',
  version: '0.1.0',
  name: 'Assess an order',
  description: 'Scores the risk and quotes shipping for an order, at the same time.',
  nodes: [
    {
      id: 'order',
      kind: 'tool',
      ref: 'acme.get-order',
      inputMapping: { orderId: { path: 'runInput.orderId' } },
    },
    {
      id: 'assess',
      kind: 'fanout',
      convergence: 'all-succeed',
      branches: [
        {
          branchId: 'risk',
          handler: 'acme.score-risk',
          outputSchema: { type: 'object', properties: { level: { type: 'string' } }, required: ['level'] },
        },
        {
          branchId: 'shipping',
          handler: 'acme.quote-shipping',
          outputSchema: { type: 'object', properties: { days: { type: 'integer' } }, required: ['days'] },
        },
      ],
    },
  ],
  edges: [
    { id: 'e1', from: '$start', to: 'order' },
    { id: 'e2', from: 'order', to: 'assess' },
    { id: 'e3', from: 'assess', to: '$end' },
  ],
  output: {
    mapping: {
      risk: { path: 'nodeOutputs.assess.risk.level' },
      shippingDays: { path: 'nodeOutputs.assess.shipping.days' },
    },
  },
});


if (defined.kind === 'err') throw new Error(defined.error.message);


export default defined.value;
  • Each branch has a branchId, the tool it runs (handler), and the outputSchema its output must match. A branch's outputSchema that doesn't compile is refused when the flow is loaded or deployed (/nodes/<i>/branches/<k>/outputSchema), not at each run.
  • Every branch gets the fanout step's input: the output of the step before it, here the whole order. Each tool takes the fields its input declares, such as total and items, and drops the rest.
  • convergence decides when the step is done and what it returns, as Convergence shows.

Run it:

Terminal window
pnpm exec kindgi runs start --flow=acme.assess-order --input='{"orderId":"A-200"}'

--flow is the flow, and --input its input. It prints the run, with both tools' results:

{
  …
  "status": "completed",
  …
  "output": {
    "risk": "high",
    "shippingDays": 4
  },
  …
}
Excerpt

In the journal, which kindgi runs journal <run-id> prints, both branches are dispatched before either completes:

{"sequence": 6, "kind": "fanout.dispatched", "nodeId": "assess", "payload": {…, "handler": "acme.score-risk", "branchId": "risk", "fanoutNodeId": "assess"}, …}
{"sequence": 7, "kind": "fanout.dispatched", "nodeId": "assess", "payload": {…, "handler": "acme.quote-shipping", "branchId": "shipping", "fanoutNodeId": "assess"}, …}
{"sequence": 8, "kind": "fanout.branch-completed", "nodeId": "assess", "payload": {"output": {"days": 4}, "branchId": "shipping", "fanoutNodeId": "assess"}, …}
{"sequence": 9, "kind": "fanout.branch-completed", "nodeId": "assess", "payload": {"output": {"level": "high"}, "branchId": "risk", "fanoutNodeId": "assess"}, …}
{"sequence": 10, "kind": "fanout.converged", "nodeId": "assess", "payload": {"output": {"risk": {"level": "high"}, "shipping": {"days": 4}}, "convergence": "all-succeed", "fanoutNodeId": "assess"}, …}
Excerpt

Convergence

Section titled “Convergence”
convergence Done when The step's output
all-succeed every branch succeeded. The first failure fails the step and cancels the others { "<branchId>": <output>, … }
any-succeed the first branch succeeds, and the others are cancelled. It fails only if every branch fails { "winnerBranchId": "<branchId>", "output": <output> }
settle-all every branch finished, succeeded or failed. The step itself succeeds { "<branchId>": { "status": "succeeded", "output": … } | { "status": "failed", "error": "…" }, … }

With settle-all, a failed branch is data for the next step, not a failed run. Here a settle-all fanout's stock branch runs acme.check-stock with an input that doesn't fit it, next to risk. The step's output:

{
  "risk": { "output": { "level": "high" }, "status": "succeeded" },
  "stock": {
    "error": "input-validation-failed: Input for tool \"acme.check-stock\" failed validation",
    "status": "failed"
  }
}

With all-succeed, the same branch fails the run:

[branch-failure] Fanout "assess" all-succeed mode: branch "stock" failed — input-validation-failed: Input for tool "acme.check-stock" failed validation

Fanout or loop

Section titled “Fanout or loop”

A fanout runs different tools on the same input. To run the same steps on each element of a list, several at a time, use a foreach loop with concurrency, as Repeat steps in a loop shows.

Next

Section titled “Next”