Writing a flow

A flow is a directory whose __init__.py holds a function marked @flow, and that function drives the agents. A weaver writes one when the same agents should be run the same way again and again, rather than typed out afresh each time.

Write the flow

mkdir -p .humanize/flows/twice

# .humanize/flows/twice/__init__.py
"""Two passes: do the work, then read it back and fix what is wrong."""

from hmz.flows import Agent, flow


@flow
def run(agents: tuple[Agent], task: str) -> None:
    (agent,) = agents
    session = agent.new()
    session(task)
    session("Now review what you just did, and fix anything that is wrong.")

Run the flow

hmz exec -f twice -a claude/claude-opus-4-8:high "add a --dry-run flag to calc.py"

/flow offers it in the interface too, beside the flows humanize ships and everything in every flowverse fetched here. A flowverse is a place flows live, and your own two directories are places like any other: .humanize/flows here is local, ~/.humanize/flows is user. ← and → step between the places.

The contract, in three rules

1. A function marked @flow, taking the agents and the task. You can call it whatever you like. The @flow mark is what makes it a flow, not the name.

2. The annotation on agents says how many the flow drives. It is a fixed-length tuple, or a NamedTuple of them.

tuple[Agent]             # one
tuple[Agent, Agent]      # two
tuple[Agent, ...]        # refused — that is not an answer

The command line that starts the flow cannot know its length any other way, so humanize checks it before the first turn:

$ hmz exec -f twice -a claude/claude-opus-5:max -a codex/gpt-5.6-sol:high "…"
hmz exec: error: twice: the flow drives 1 agents, 2 given

3. The annotation must be readable at runtime. Import Agent normally, and not under if TYPE_CHECKING. A count that nothing can read back is not one a command line can be held to.

The most common first mistake

from typing import TYPE_CHECKING

if TYPE_CHECKING:                      
    from hmz.flows import Agent   

hmz exec: error: twice: the flow's agents cannot be read here (name 'Agent' is not
defined) -- import what the annotation names at runtime, so the count it states can be checked

Say what each agent is allowed

Whoever runs your flow names a CLI, an account, a model and an effort. What the agent may do is yours: write an AgentDefaults beside the place, and every agent handed to it runs that way from its first turn, whichever CLI fills it.

from typing import Annotated, NamedTuple

from hmz.flows import Agent, AgentDefaults, flow


class Agents(NamedTuple):
    """One that writes the change, and one that only reads it."""

    builder: Agent
    reviewer: Annotated[
        Agent, AgentDefaults(permission="read-only", web_search=False)
    ]


@flow
def run(agents: Agents, task: str) -> None:
    agents.builder(task)
    agents.reviewer(f"review what was just done: {task}")

Written beside the type	What it says	Where
permission=	one rung of the four	Permissions
goals=	whether the backend's own goal feature is available	Goals
web_search=	whether it may read the internet	Agents

What a place declares only ever tightens. bypass, goals on and the web readable are the loosest of each and what a place that writes none of them declares, so a flow that says nothing runs its agents at exactly what they came with -- and a flow you call cannot hand itself more than you were running at. None of the three can be said on the line that runs the flow, and none of them has a row on the sheet an agent is set up on: they are things about the work, and the work is what the flow is.

Choose what the next turn remembers

agent("do the task")          # a session of its own, dropped straight after: nothing carries over
session = agent.new()
session("do the task")        # opens it
session("keep going")         # resumes it, the first turn still in context

A turn is one request to the agent. A session keeps its turns in context, so the second turn knows what the first one did. The flow above holds a session; without one:

@flow
def run(agents: tuple[Agent], task: str) -> None:
    (agent,) = agents
    agent(task)
    agent("Now review what you just did, and fix anything that is wrong.")

The second turn arrives with no idea what "what you just did" refers to. It has to find out from the repository. Sometimes that is exactly what you want, which is what a Ralph loop is.

Make the loop survive a bad turn

A turn that fails raises subprocess.CalledProcessError. This happens whatever it was actually run through, so a flow catches turns rather than transports. In a loop, that raise would end the run on the first hiccup.

suppress=True is || true for a turn:

import time

@flow
def run(agents: tuple[Agent], task: str) -> None:
    (agent,) = agents
    while True:
        agent(task, suppress=True)     # "" if it failed, and the loop goes round again
        time.sleep(5)

It catches a turn that failed and nothing else. It does not catch an agent that has been stopped, and it does not catch a backend with no goal feature. A backend with no goal feature is a flow to correct rather than a turn to retry.

Give the loop a finish line

A while True is only useful if something ends it. The flow is ordinary Python, so read the repository:

import subprocess
from pathlib import Path

def green() -> bool:
    return subprocess.run(["python", "-m", "pytest", "-q"], check=False).returncode == 0

@flow
def run(agents: tuple[Agent], task: str) -> None:
    (agent,) = agents
    for _ in range(20):
        agent(task, suppress=True)
        if green() and "- [ ]" not in Path("TASK.md").read_text():
            return

A flow is just a function, so it may branch, sleep, read files, shell out and give up.

Say what the flow is

The first line of the docstring is what is shown beside the flow's name where flows are listed:

"""Two passes: do the work, then read it back and fix what is wrong."""

Where a flow lives, and what it is called

Lives at	Called
.humanize/flows/twice/__init__.py	twice in this project, or local/twice
~/.humanize/flows/twice/__init__.py	twice in every project, or user/twice
a flowverse	<flowverse>/twice
anywhere else	its path: -f ./flows/twice

A name is looked for nearest first, so a flow of yours may stand in for one of humanize's by taking its name. A file whose name starts with _ is not a flow.

Check your work

Hmz().flows.check reads the flow before anything runs it: a static reading that executes nothing, then the flow loaded in a subprocess held to a clock. Driving it with stubs — including the world where the reviewer never says the work is done — is proved(), a call of your own. See Checking a flow.

from hmz.sdk import Hmz

for one in Hmz().flows.check("local/twice"):
    print(f"{one.where}:{one.line}: {one.severity}: {one.code}: {one.said}")

And to ask a flow what it declares, without reading it for anything else:

from hmz.flows import drives

drives("twice")       # the names of the agents it declares

A flow whose shape is known before it runs

Everything above is a flow: Python that may branch any way it likes, and whose shape is whatever it does. Where the shape is known in advance — a pipeline of phases, a review loop meant to run for a week — an atlas is the stricter bargain. Its body is a declaration rather than a program, compiled into a graph before the first turn, so it is checked whole up front and a run of one is picked up node by node rather than started again.

See also

• An atlas
• Read the run back
• Flowverses
• Loops
• Stopping
• Port a project