Pipelines and Graphs

Two building blocks for work that takes more than one call. A pipeline runs steps in a straight line: fill a prompt, ask the agent, clean the answer. A graph runs nodes that can loop, branch, stop for a human and continue later. Both use the agent you already built, here called $claw.

Block	Use it when	Entry point
Pipeline	the steps are always the same, in the same order	`Pipeline::make()`
Graph	the next step depends on a result, a human must approve, or work splits into branches	`new Graph('name')`

Pipelines

Build and run

1
use PhpClaw\Pipeline\Pipeline;
2
use PhpClaw\Pipeline\Steps\ClawStep;
3
use PhpClaw\Pipeline\Steps\PromptStep;
4

5
$capital = Pipeline::make()
6
    ->pipe(new PromptStep('Reply with one word only: the capital city of {country}.'))
7
    ->pipe(new ClawStep($claw))
8
    ->pipe(fn (string $reply): string => trim($reply, " .\n"));
9

10
$capital->invoke(['country' => 'France']);
11
// "Paris"
12

13
$capital->batch(['jp' => ['country' => 'Japan'], 'it' => ['country' => 'Italy']]);
14
// ['jp' => 'Tokyo', 'it' => 'Rome']

Each step’s output is the next step’s input. pipe() returns a new pipeline and leaves the one you called it on unchanged, so a base pipeline can be extended safely. A plain callable becomes a TransformStep.

Step	Input	Output
`PromptStep($template)`	an array of values	the template with every `{name}` filled
`ClawStep($claw)`	text	the agent’s reply text
`StructuredStep($claw, $schema)`	text	the reply as an array, validated against the JSON Schema
`TransformStep($callable)` or a callable	anything	what the callable returns

batch() runs every input in turn and keeps the keys; the first input that fails stops the batch.

Stream

stream() runs every step but the last, then streams the last step’s reply. The last step must be a ClawStep; anything else throws PipelineException.

1
$sentence = Pipeline::make()
2
    ->pipe(new PromptStep('Write one sentence about {topic}.'))
3
    ->pipe(new ClawStep($claw))
4
    ->stream(['topic' => 'PHP'], function (string $token): void {
5
        echo $token;
6
    });

Retry and fallback

Wrap any step:

1
use PhpClaw\Pipeline\Steps\FallbackStep;
2
use PhpClaw\Pipeline\Steps\RetryStep;
3

4
$reliable = new RetryStep(new ClawStep($claw), attempts: 3);
5
$safe = new FallbackStep(new ClawStep($claw), [new ClawStep($backupClaw)]);

Step	Behaviour	Default errors
`RetryStep($step, attempts, retryOn)`	runs the step again on a listed error, up to `attempts` runs in total (default 3), with no wait between them	`ProviderException`, `StructuredOutputException`
`FallbackStep($step, $fallbacks, fallbackOn)`	on a listed error, tries each fallback in order and returns the first output; when all fail, it throws the main step’s error	`ProviderException`, `StructuredOutputException`

Any other error is thrown straight away. The agent retries a failed provider call only when you set maxRetries() (default 0), and RetryStep adds attempts on top of those.

A pipeline as a tool

PipelineTool lets an agent call a whole pipeline as one tool:

1
use PhpClaw\Claw;
2
use PhpClaw\Pipeline\PipelineTool;
3

4
$tool = new PipelineTool(
5
    $capital,
6
    'capital_city',
7
    'Return the capital city of a country.',
8
    ['type' => 'object', 'properties' => ['country' => ['type' => 'string']], 'required' => ['country']],
9
);
10

11
$agent = Claw::builder()->addTool($tool)->build();

The tool input becomes the pipeline input. Text output is returned as is; anything else is encoded as JSON. A PipelineException, PromptTemplateException or StructuredOutputException becomes a tool error the model can react to; any other error stops the run.

Graphs

Nodes and edges

A node gets the current GraphState and returns it. Edges decide what runs next.

1
use PhpClaw\Graph\Graph;
2
use PhpClaw\Graph\GraphState;
3

4
$review = (new Graph('writer-reviewer'))
5
    ->addNode('write', fn (GraphState $s): GraphState => $s
6
        ->set('draft', trim($claw->send('Write one sentence about '.$s->get('topic').'.')->text))
7
        ->set('drafts', $s->get('drafts', 0) + 1))
8
    ->addNode('review', fn (GraphState $s): GraphState => $s
9
        ->set('ok', str_contains(strtoupper($claw->send('Is this one complete English sentence? Answer only YES or NO: '.$s->get('draft'))->text), 'YES')))
10
    ->setEntry('write')
11
    ->addEdge('write', 'review')
12
    ->addConditionalEdge('review', fn (GraphState $s): string => $s->get('ok') || $s->get('drafts') >= 3 ? Graph::END : 'write');
13

14
$final = $review->run(new GraphState(['topic' => 'PHP']));
15
$final->get('draft');

Method	Meaning
`addNode($name, $node)`	`$node` gets `(GraphState $state, GraphRuntime $runtime)` and returns the state, a `GraphCommand`, or nothing
`addEdge($from, $to)`	always go from one node to another
`addConditionalEdge($from, $router)`	`$router` gets the state and returns the next node’s name or `Graph::END`
`setEntry($name)`	the node the walk starts from
`run($state, maxSteps, context, threadId)`	walk from the entry node until a route reaches `Graph::END`, and return the final state
`validate()`	check that the entry, every edge and every pause name an added node; `run()` calls it first

A conditional edge wins over a plain edge from the same node. A node with neither ends the walk.

Loop guard: a walk that runs more than maxSteps nodes (default 50) stops with GraphException, “Graph stopped after N steps without reaching Graph::END.”

Update state and jump in one step

Return a GraphCommand to set values and pick the next node at once:

1
use PhpClaw\Graph\GraphCommand;
2

3
->addNode('send', fn (GraphState $s): GraphCommand => new GraphCommand(
4
    goto: Graph::END,
5
    update: ['sent' => $s->get('approved') === 'yes'],
6
))

Retries and error handlers

1
->addNode('fetch', $fetch, attempts: 3, onError: fn (GraphState $s, \Exception $e, string $node): GraphState => $s->set('failed', $node))
2
->setNodeDefaults(attempts: 2)

attempts counts the first run. retryOn lists the errors that cause another attempt (by default ProviderException and StructuredOutputException). When the last attempt fails, onError gets the state, the error and the node name, and its return value is used like the node’s own. Without an onError, the error is thrown. setNodeDefaults() sets these for every node that does not set its own.

Context and store

1
$graph->setStore($memory)->run($state, context: ['user_id' => 7]);

Every node gets a GraphRuntime as its second argument: $runtime->context holds the run’s context, $runtime->store the memory driver passed to setStore() (or null), and $runtime->runId the graph run’s id.

Pause for a human and resume later

Give the graph a checkpointer (any memory driver) and run it under a thread id. The state is saved after every node. A node calls $runtime->interrupt($value) to stop and wait:

1
use PhpClaw\Exceptions\GraphInterruptedException;
2
use PhpClaw\Graph\GraphRuntime;
3

4
$refund = (new Graph('refund'))
5
    ->setCheckpointer($memory)
6
    ->addNode('draft', fn (GraphState $s): GraphState => $s
7
        ->set('draft', trim($claw->send('Reply with one short sentence offering a refund for order '.$s->get('order').'.')->text)))
8
    ->addNode('approve', fn (GraphState $s, GraphRuntime $rt): GraphState => $s
9
        ->set('approved', $rt->interrupt(['question' => 'Approve this refund?', 'draft' => $s->get('draft')])))
10
    ->addNode('send', fn (GraphState $s): GraphCommand => new GraphCommand(goto: Graph::END, update: ['sent' => $s->get('approved') === 'yes']))
11
    ->setEntry('draft')
12
    ->addEdge('draft', 'approve')
13
    ->addEdge('approve', 'send');
14

15
try {
16
    $refund->run(new GraphState(['order' => 'A-1042']), threadId: 'refund-A-1042');
17
} catch (GraphInterruptedException $e) {
18
    // $e->node is 'approve'; $e->value is the array passed to interrupt()
19
}
20

21
// Later, in any process:
22
$final = $refund->resume('refund-A-1042', 'yes');
23
$final->get('sent'); // true

On resume the paused node runs again, and this time interrupt() returns the answer given to resume().

Method	Meaning
`setCheckpointer($memory)`	save a snapshot after every node; any `MemoryInterface` driver
`interruptBefore(...$nodes)`, `interruptAfter(...$nodes)`	pause before or after the named nodes without changing them
`resume($threadId, $answer)`	continue a paused thread from its saved snapshot
`getState($threadId)`	the newest `GraphSnapshot`, or `null`
`getStateHistory($threadId)`	every snapshot, oldest first
`updateState($threadId, $values)`	write values into the newest snapshot, for example a human’s edit before `resume()`

Saved state must be plain data: null, scalars, and arrays of those. Anything else throws GraphException when it is saved. Snapshots are kept under the graph_threads memory namespace.

Fan out to branches

addFanOut() splits the state into branches, runs a branch graph once for each, and hands every result to a join node:

1
$translateOne = (new Graph('translate-one'))
2
    ->addNode('translate', fn (GraphState $s): GraphState => $s
3
        ->set('text', trim($claw->send('Translate to '.$s->get('lang').', reply with only the translation: '.$s->get('source'))->text)))
4
    ->setEntry('translate');
5

6
$translateAll = (new Graph('translate-all'))
7
    ->addFanOut(
8
        'split',
9
        fn (GraphState $s): array => array_map(
10
            fn (string $lang): GraphState => new GraphState(['lang' => $lang, 'source' => $s->get('source')]),
11
            ['French', 'German', 'Spanish'],
12
        ),
13
        $translateOne,
14
        join: 'collect',
15
    )
16
    ->addNode('collect', fn (GraphState $s): GraphState => $s->set('summary', implode(' | ', array_map(
17
        fn (array $result): string => $result['ok'] ? $result['values']['text'] : 'FAILED',
18
        $s->get('split'),
19
    ))))
20
    ->setEntry('split');
21

22
$translateAll->run(new GraphState(['source' => 'Thank you for your order']))->get('summary');

The join node finds the results under the fan-out node’s name, in declared order. Each result is ['ok' => true, 'values' => [...]] or ['ok' => false, 'error' => <class>, 'message' => <text>], so one failed branch does not stop the others. A branch graph may not fan out again, and at most 32 branches are allowed (maxBranches).

Without a thread id the branches run one after another in the same request. With a checkpointer and a thread id, run() saves each branch as a queued thread and stops with GraphInterruptedException; workers then finish the job:

1
$report = $translateAll->drain(limit: 5);
2
// $report->completed, $report->suspended, $report->failed, $report->skipped

drain() runs up to limit queued threads (default 5) and stops starting new ones after timeBudgetSeconds (default 20). It runs branches first, then the join node once every branch of a parent has finished. Call it from cron, a queue worker or a CLI command. Before running a thread it marks the thread as running and skips any thread that is no longer queued; that mark is a read and a write through the memory driver, not a lock, so run one worker at a time per graph to be sure each branch runs once. The graph needs a name for drain() to find its threads, and a branch cannot pause for a human: an interrupt() call inside a queued branch fails that branch.

Events

A graph run fires graph.start (with entry_node) and, when it reaches Graph::END, graph.end (with steps and last_node). A fan-out fires graph.fan_out and graph.join. An agent called from a node carries the graph’s run id as its parent_run_id, so traces nest. The events carry node names and counts, never the graph state. See Hooks and Cloud.

Errors

Exception	When
`PipelineException`	a step gets input it cannot use, or `stream()` ends in a step that cannot stream
`GraphException`	no entry node, an unknown node, the step limit, a checkpointer, thread id or graph name missing, a thread that is unknown or not paused, a value that cannot be saved, a nested fan-out or too many branches
`GraphInterruptedException`	a saved run paused; `$e->threadId`, `$e->node` and `$e->value` say where and why