13 — Tuning an agentic flow

Generated from a notebook

This page is notebooks/13_agentic_flows.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.

Notebooks 10–12 tuned a neural architecture and then used the recorded history to work out which change actually mattered. This one does the same thing to a flow made of agents, and the point is that nothing about the method changes.

An agentic flow in Soma is a graph. Its nodes reach outside the process — they call a model, run a tool — but they sit on the same edges, go through the same compiler, write to the same cache, and land in the same experiment pool. So the prompt an agent uses is a hyperparameter, the model it calls is a hyperparameter, and whether two nodes should be connected at all is a hyperparameter. All three go in one search space and one Study reads it.

What we will build: a refine loop — a writer drafts, a judge grades, the writer sees the critique and tries again — and then search over the writer’s prompt, the judge’s strictness, and the shape.

The model

Everything below runs against a mock model served from this notebook: a small HTTP server that grades by content. That makes this notebook run anywhere with no key, no GPU and no network, and makes its numbers reproducible.

To use a real model instead, delete the mock cell and point the catalog at something real — the rest of the notebook does not change:

providers.toml

[providers.ollama]
base_url = "http://localhost:11434/v1"
auth = { type = "none" }

then soma.Agent(model="ollama/qwen2.5").

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_nb13_"))
os.environ["SOMA_CACHE_DIR"] = str(WORK / "cache")
os.chdir(WORK)
print(soma.__version__, WORK)

0.4.0 /tmp/soma_nb13_35q3dbnl

A model that grades by content

The writer parrots whichever system prompt it was given, and the judge scores that text. A “detailed” writer scores well and a “terse” one badly, so the search has a real gradient to find — which is all we need to watch the machinery work.

QUALITY = {"be detailed": 0.9, "be helpful": 0.65, "be terse": 0.25}


def reply(body):
    """What the mock model answers, given a request. Returns a `message`."""
    messages = body["messages"]
    system = messages[0]["content"] if messages[0]["role"] == "system" else ""

    if system.startswith("You grade"):
        graded = messages[-1]["content"]
        score = next((v for k, v in QUALITY.items() if k in graded), 0.1)
        return {"content": json.dumps({"score": score, "reason": "graded by content"})}

    # A term extractor asked for JSON: answer in the shape it was given.
    if system.startswith("Extract"):
        return {"content": json.dumps({"term": "compiler",
                                       "definition": "translates programs"})}

    # An agent that has tools: consult the glossary once, then answer with it.
    tool_answers = [m["content"] for m in messages if m.get("role") == "tool"]
    if body.get("tools") and not tool_answers:
        return {"content": None, "tool_calls": [{
            "id": "call_1", "type": "function",
            "function": {"name": "lookup",
                         "arguments": json.dumps({"term": "compiler"})},
        }]}
    if tool_answers:
        return {"content": f"Per the glossary — {tool_answers[-1]}"}

    return {"content": system}  # the writer parrots its instructions


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        message = reply(body)
        finish = "tool_calls" if message.get("tool_calls") else "stop"
        payload = json.dumps({
            "choices": [{"message": message, "finish_reason": finish}],
            "usage": {"prompt_tokens": 5, "completion_tokens": 3},
        }).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("mock model on", server.server_port)

mock model on 38323

One agent

soma.Agent is a node like any other: g.node(id, thing) takes it exactly the way it takes a filter. It runs a reason-act loop — ask the model, run whatever tools it asks for, repeat until it answers in prose.

g = soma.Graph(cache="memory")
g.node("writer", soma.Agent(model="mock/any", system="be helpful"))

g.forward("explain compilers")

'be helpful'

Tools

A tool is a Python function with a docstring. The docstring is not decoration: it is what the model reads to decide whether to call it, so a tool without one is refused rather than registered inert.

CALLS = []


@soma.tool
def lookup(term: str) -> str:
    """Look a term up in the glossary. Call this for unfamiliar jargon."""
    CALLS.append(term)
    return f"{term}: a program that translates programs."


lookup.name, lookup.schema["required"]

('lookup', ['term'])

Handing the tool to an agent turns it into a capability: the model decides when to call it, the runtime performs and journals the call like any other effect, and the answer lands back in the conversation. CALLS is the proof that the loop went through Python and back.

g = soma.Graph(cache="memory")
g.node("scholar", soma.Agent(model="mock/any",
                             system="Answer using the glossary.",
                             tools=[lookup]))

answer = g.forward("what is a compiler?")
CALLS, answer

(['compiler'],
 'Per the glossary — compiler: a program that translates programs.')

A refine loop

soma.Judge grades against a rubric and reports done, which is exactly the signal a loop reads to stop. soma.agentic.refine wires the two together:

g.node("worker", worker)
g.node("judge", judge)
g.connect("worker", "judge")
g.loop("refine", body="worker", until="judge", max_iterations=max_rounds)

The loop carries: after each pass, the judge’s verdict becomes what the writer reads next. Without that the loop would redraft the same thing N times, which is the difference between “refine” meaning something and not.

from soma.agentic import refine


def flow(system="be helpful", threshold=0.8, rounds=3):
    return refine(
        worker=soma.Agent(model="mock/any", system=system),
        judge=soma.Judge(model="mock/any", rubric="Is it useful?", threshold=threshold),
        max_rounds=rounds,
        cache="memory",
    )


print(flow().to_mermaid())

graph LR
    revise[revise]
    worker[/worker/]
    judge[/judge/]
    refine((refine (max 3)))
    revise --> worker
    worker --> judge
    refine -.-> revise

Effectful nodes render as parallelograms — they reach outside the graph. Everything else about the diagram is an ordinary Soma graph.

verdict = flow(system="be detailed").forward("explain compilers")
verdict["score"], verdict["passed"], verdict["reason"]

(0.9, True, 'graded by content')

The verdict carries value — what it judged — as well as the score. That is what lets the next round improve something it can still see.

terse = flow(system="be terse").forward("explain compilers")
terse["score"], terse["passed"]

(0.25, False)

The prompt is a hyperparameter

An agent’s constructor arguments are its hyperparameters, so the space is declared where the value goes. A filter declares its space as a class attribute; both end up in the same search_space(), and a Study cannot tell an agent from a filter.

g = soma.Graph(cache="memory")
g.node("writer", soma.Agent(
    model="mock/any",
    system=soma.search(choices=["be terse", "be helpful", "be detailed"]),
))
g.node("judge", soma.Judge(
    model="mock/any",
    rubric="Is it useful?",
    threshold=soma.search(0.5, 0.95),
))
g.connect("writer", "judge")
g.loop("refine", body="writer", until="judge", max_iterations=3)

for dim in g.search_space():
    print(dim)

{'high': 0.95, 'low': 0.5, 'name': 'judge.threshold', 'scale': 'linear', 'type': 'float'}
{'choices': ['be terse', 'be helpful', 'be detailed'], 'name': 'writer.system', 'type': 'categorical'}

A searchable argument still resolves to a concrete value — the first choice, or the lower bound — because the graph has to be runnable before any study samples it.

study = g.study(
    "prompt-search",
    strategy="grid",
    n_trials=3,
    objectives=[("score", "maximize")],
    tracking=False,
)


def trial(t):
    g.apply_params(t.params)
    with g.track_run("agentic-refine", tags=["nb13"]):
        verdict = g.forward("explain compilers")
    return {"score": verdict["score"]}


study.run(trial)
study.best_trial["params"], study.best_trial["metrics"]

({'judge.threshold': 0.5, 'writer.system': 'be detailed'}, {'score': 0.9})

A Step is immutable once built, so a sampled prompt cannot be written into it. The study writes to the live Agent instead, and the graph rebuilds its steps from those before every run. That is the only reason the sampled prompt reaches the model at all.

The shape is a hyperparameter too

Whether two nodes should be connected is a design question, and design questions are better answered by a search than by an argument. optional puts an edge in the space:

g.optional("writer", "judge")
[d["name"] for d in g.search_space()]

['judge.threshold', 'writer.system', 'edge:writer->judge']

Cutting an edge sets it aside whole, so restoring it restores the graph byte-identically — a trial that changes the topology has to leave the next trial starting from the same place. (Control edges are not eligible: cutting one would change what the loop owns, not just what flows.)

before = g.to_mermaid()
g.apply_params({"edge:writer->judge": False})
cut = g.to_mermaid()
g.apply_params({"edge:writer->judge": True})

print("cut differs: ", cut != before)
print("restored exactly:", g.to_mermaid() == before)

cut differs:  True
restored exactly: True

Structured output, validated

schema= asks the model for a shape; Validate checks the claim. The check is deliberately structural — root type, required, property types — because an invented violation would spend a real model call to “fix” a correct answer. The verdict carries a branch, so a graph can route the invalid case instead of crashing on it.

from soma.agentic import Validate

SHAPE = {
    "type": "object",
    "required": ["term", "definition"],
    "properties": {"term": {"type": "string"},
                   "definition": {"type": "string"}},
}

g = soma.Graph(cache="memory")
g.node("extract", soma.Agent(model="mock/any",
                             system="Extract the term being defined.",
                             schema=SHAPE))
g.node("check", Validate(SHAPE))
g.connect("extract", "check")

verdict = g.forward("a compiler translates programs")
verdict["ok"], verdict["value"]

(True, {'definition': 'translates programs', 'term': 'compiler'})

What was recorded

Every tracked run went into the experiment pool — the same .soma/experiments.jsonl a purely computational run writes to. There is no separate agent memory: an agent remembers what it ran because running it recorded it.

records = [
    json.loads(line)
    for line in (WORK / ".soma" / "experiments.jsonl").read_text().splitlines()
    if line
]
len(records), [r["name"] for r in records]

(3, ['agentic-refine', 'agentic-refine', 'agentic-refine'])

for r in records:
    print(f"{r['id'][:18]:20} parent={str(r.get('parent'))[:18]:20} "
          f"arch={bool(r.get('architecture'))}")

run_20260804T08332   parent=None                 arch=True
run_20260804T08332   parent=run_20260804T08332   arch=True
run_20260804T08332   parent=run_20260804T08332   arch=True

Each run names its parent, so the sequence is a lineage rather than a pile — and each carries an architecture fingerprint, which is what makes “were these two flows the same shape?” a question with an answer.

The run directory holds the topology as it was:

run_dir = sorted((WORK / ".soma" / "runs").glob("*"))[-1]
graph = json.loads((run_dir / "graph.json").read_text())

for node in graph["nodes"]:
    print(f"{node['id']:10} {node['kind']}")

writer     {'type': 'Step', 'step_name': 'Agent'}
judge      {'type': 'Step', 'step_name': 'Judge'}
refine     {'type': 'Loop', 'max_iterations': 3, 'until': {'type': 'WhenSignaled', 'node': 'judge'}}

That JSON is the contract. A node says what kind it is and names its filter or step; what it does lives in the registry. Control edges carry the structure — the loop’s ownership, the branch arm’s label. Anything outside this process (a visual editor, another language) reads and edits a Soma graph through exactly this, without depending on Soma’s internals.

What this notebook was about

Nothing here was agent-specific machinery. The search space, the sampler, the pruning, the cache, the run directory, the lineage — all of it already existed for computational pipelines and applies to a flow made of models because a flow made of models is the same kind of object.

That is the whole bet: not another agent framework, but the first runtime where an agentic flow is tuned the way a model is tuned, and where the answer to “which version was better” is recorded rather than remembered.