15 — Pipelines and agents, each calling the other

Generated from a notebook

This page is notebooks/15_pipelines_and_agents.ipynb, executed, with its outputs. Run it yourself with jupyter lab notebooks/ after pip install 'somatize[viz]'. Interactive Plotly figures show here as the PNG the notebook also stores.

The claim behind Soma’s agentic layer is symmetry: an agent is a node a pipeline can contain, and a pipeline is a tool an agent can run. Notebook 13 tuned an agentic flow; this one exercises the seam itself, in both directions, plus the two pieces that make the seam safe — schema checks on the edges and the effect journal underneath.

Everything runs against an embedded mock provider, so there is no key, no GPU and no network.

import json
import os
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

import soma

WORK = Path(tempfile.mkdtemp(prefix="soma_nb15_"))
os.environ["SOMA_CACHE_DIR"] = str(WORK / "cache")
os.chdir(WORK)

HITS = {"n": 0}


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        HITS["n"] += 1
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        asked = body["messages"][-1]["content"] or ""
        label = "positive" if "good" in asked.lower() else "negative"
        payload = json.dumps({
            "choices": [{"message": {"content": label}, "finish_reason": "stop"}],
            "usage": {"prompt_tokens": 5, "completion_tokens": 1},
        }).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, *a):
        pass


server = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()

catalog = WORK / "providers.toml"
catalog.write_text(
    f'[providers.mock]\n'
    f'base_url = "http://127.0.0.1:{server.server_port}/v1"\n'
    f'auth = {{ type = "none" }}\n'
)
os.environ["SOMA_PROVIDERS"] = str(catalog)
print(soma.__version__, "· mock model on port", server.server_port)

0.4.0 · mock model on port 40275

An agent inside a pipeline

The first direction: a computational pipeline whose data path runs through a model. clean is an ordinary filter, labeller is an agent, and score is soma.library.Eval — a real metric, because here there is a reference to compare against, and a metric answers exactly, for free, reproducibly. The graph neither knows nor cares which nodes call a model.

from soma.library import Eval


class Clean(soma.Filter):
    _kind = "stateless"
    _cache_version = "1"
    _output_schema = "text"

    def forward(self, x, state):
        return x.strip().lower()


class Truth(soma.Filter):
    """The reference label — in a real pipeline, your labelled data."""

    _kind = "stateless"
    _cache_version = "1"

    def forward(self, x, state):
        return "positive" if "good" in x.lower() else "negative"


g = soma.Graph(cache="memory")
g.node("clean", Clean())
g.node("labeller", soma.Agent(model="mock/any",
                              system="You label sentiment: positive or negative."))
g.node("truth", Truth())
g.node("score", Eval(metrics=["accuracy"], prediction="labeller", reference="truth"))
g.connect("clean", "labeller")
g.connect("labeller", "score")
g.connect("truth", "score")

g.forward("  The food was GOOD  ")

{'accuracy': 1.0, 'n': 1}

Diagrams tell the two kinds of node apart — the agent renders as a parallelogram because it reaches outside the graph:

g

The edges are checked

Declaring _input_schema / _output_schema on a node makes the edge a contract. A tensor arriving where a conversation is expected has no possible reading, and the compiler refuses to build the graph — before a single token is spent. Across MAST’s 1600+ annotated multi-agent traces, context malformed at a handoff is the largest failure bucket after bad specifications; this is the cheap place to catch it.

class Embedding(soma.Filter):
    _kind = "stateless"
    _cache_version = "1"
    _output_schema = {"dtype": "Float64", "shape": None}

    def forward(self, x, state):
        return [0.1, 0.2, 0.3]


class Chat:
    _cache_version = "1"
    _input_schema = "messages"

    def poll(self, ctx):
        return soma.agentic.Done("never reached")


bad = soma.Graph(cache="memory")
bad.node("embed", Embedding())
bad.node("chat", Chat())
bad.connect("embed", "chat")

try:
    bad.compile("no_cache")
except Exception as e:
    print(type(e).__name__, "—", e)

RuntimeError — compilation error: `embed` outputs f64 but `chat` expects messages, and there is no conversion between them. Insert a node that adapts one to the other

A pipeline as a tool for a step

The second direction. A step hands work to a whole pipeline with soma.agentic.RunGraph and reads the result back like any other effect. Two things to know:

• The effect carries the pipeline’s structure; its implementations are made runnable with register_graph on the outer graph, once.
• A live Graph held by a step is stored underscored (self._pipeline) — it cannot enter the step’s JSON identity, and it does not need to: the journal keys the effect by the graph’s own content.

from soma.agentic import Await, Done, RunGraph


class RunsThePipeline:
    """Hands its input to the labelling pipeline, reads the verdict back."""

    _cache_version = "1"

    def __init__(self, pipeline):
        self._pipeline = pipeline

    def poll(self, ctx):
        if ctx.turn == 0:
            return Await(RunGraph(self._pipeline, input=ctx.input))
        result = ctx.result()
        if result["kind"] == "failed":
            return Done("pipeline failed: " + result["message"])
        return Done(f"the pipeline said: {result['output']}")


pipeline = soma.Graph(cache="memory")
pipeline.node("clean", Clean())
pipeline.node("labeller", soma.Agent(model="mock/any",
                                     system="You label sentiment: positive or negative."))
pipeline.connect("clean", "labeller")

outer = soma.Graph()          # persistent cache: the journal lives on disk
outer.node("driver", RunsThePipeline(pipeline))
outer.register_graph(pipeline)

outer.forward("the service was good", run_id="nb15-demo")

'the pipeline said: positive'

The sub-pipeline contains an agent, so the graph effect is journaled by its site — (run, node, turn, index) — never reused across runs by content, because a model’s answer to the same question twice is genuinely two events. Within one run, though, the journal is the crash-recovery story: replaying the same run_id serves every recorded effect instead of performing it again.

before = HITS["n"]
again = outer.forward("the service was good", run_id="nb15-demo")
again, f"model calls during the replay: {HITS['n'] - before}"

('the pipeline said: positive', 'model calls during the replay: 0')

Fan-out decided by the data

orchestrate is the planner→workers→synthesizer pattern with the width genuinely dynamic: the planner names the tasks, Spawn opens one worker per task, and nothing in the static topology says how many. Steps are plain Python objects with poll, so this one needs no model at all.

from soma.agentic import orchestrate


class Splits:
    _cache_version = "1"

    def poll(self, ctx):
        return Done(list(ctx.input))


class Labels:
    _cache_version = "1"

    def poll(self, ctx):
        return Done("positive" if "good" in ctx.input else "negative")


class Joins:
    _cache_version = "1"

    def poll(self, ctx):
        votes = list(ctx.input)
        return Done({"positive": votes.count("positive"),
                     "negative": votes.count("negative")})


flow = orchestrate(Splits(), Labels(), Joins(), max_workers=8, cache="memory")
flow.forward(["good soup", "slow service", "really good bread"])

{'negative': 1, 'positive': 2}

A human in the loop

Suspend stops the run and persists it; the pause arrives as a typed SomaSuspended carrying everything needed to answer it. Filing the answer and re-running under the same run id replays the journal up to the pause and continues — in another process, days later, it works the same way, because the journal is the checkpoint.

from soma.agentic import Suspend


class NeedsApproval:
    _cache_version = "1"

    def poll(self, ctx):
        if not ctx.results:
            return Suspend("ship this label?")
        answer = ctx.results[0].get("output")
        return Done(f"shipped, approved by: {answer}")


approval = soma.Graph()
approval.node("gate", NeedsApproval())

try:
    approval.forward("the label")
except soma.SomaSuspended as pause:
    print("paused:", pause.reason["prompt"])
    approval.resume(pause.run_id, pause.node_id, pause.turn, pause.reason, "manu")
    print(approval.forward("the label", run_id=pause.run_id))

paused: ship this label?
shipped, approved by: manu

Where this leaves the seam

direction	spelling
agent in a pipeline	g.node("labeller", soma.Agent(...))
pipeline in an agent	Await(RunGraph(sub, ...)) + g.register_graph(sub)
contract on an edge	_input_schema / _output_schema
dynamic width	Spawn / orchestrate
pause for a person	Suspend → SomaSuspended → resume

One graph model, one executor, one journal — which is the point: nothing above had a second engine behind it.