Workflow#

Workflow is OpenRath’s composition layer. Normal transforming workflows express runtime logic as Session -> Session; Selector is the explicit routing exception that returns the next workflow choice.

This page explains the structure from single-agent to multi-agent workflows, module-tree registration, Provider placement, Selector-driven routing, static compilation, and how session, sandbox, and tool traces move through the call chain.

The diagram below shows the intended mental model: a workflow is a callable module over Session, and its internal agents can fork, compose, and compress state explicitly.

Overview#

OpenRath workflows follow a pattern close to PyTorch modules:

PyTorch intuition	OpenRath equivalent
`Module.forward(x)` defines computation	`Workflow.forward(session)` defines agent orchestration
Child modules are attached as attributes	`AgentParam` leaves and nested `Workflow` children register by attribute assignment
Tensors move between modules	`Session` moves between agents and workflows
Module tree can be printed and inspected	`repr(workflow)`, `modules()`, and `compile()` expose registered resources

Workflow has a small job: collect directly attached AgentParam values, provide the forward(...) convention, and make instances callable through workflow(session). Execution order, branching, compression, tool injection, and child workflow calls are written explicitly in normal Python code.

Source map#

File	Responsibility
`src/rath/flow/workflow.py`	`Workflow` base class, module-tree registration, placement, `compile()`, repr.
`src/rath/flow/compile.py`	Static manifest, preflight validation, and memory-store lifecycle.
`src/rath/flow/agent_param.py`	`AgentParam` values that can be registered by a workflow.
`src/rath/flow/agent.py`	`Agent` preset workflow wrapping one agent loop.
`src/rath/flow/compressor.py`	`Compressor` preset workflow wrapping compression.
`src/rath/flow/selector.py`	`Selector` preset that chooses the next workflow.
`src/rath/flow/empty.py`	`EmptyWorkflow` identity result for no-match/completion.
`src/rath/session/loop.py`	Runs the LLM loop, tool calls, sandbox transfer, and lineage writeback.
`src/rath/session/select.py`	One-shot selection request and index parsing.
`example/01_hello_agent.py`	Smallest preset `flow.Agent` workflow.
`example/08_compress.py`	Preset `flow.Compressor` workflow.
`example/11_dynamic_selector.py`	Dynamic `if` and bounded `while` routing.

Minimal Workflow#

Inherit from Workflow and implement forward(self, session) -> Session:

from rath.flow import Workflow
from rath.session import Session


class IdentityWorkflow(Workflow):
    def forward(self, session: Session) -> Session:
        return session

Workflow.__call__(session) directly calls forward(session). The base forward(...) raises NotImplementedError, so subclasses must define their own runtime logic.

AgentParam Auto-Registration#

When an AgentParam is assigned as a workflow attribute, Workflow.__setattr__ records it in _agents:

from rath.flow import AgentParam, Provider, Workflow
from rath.session import Session


class PlanningWorkflow(Workflow):
    def __init__(self):
        super().__init__()
        self.planner = AgentParam(
            Session.from_agent_prompt("Plan the work."),
            Provider(api_key="sk-...", model="gpt-5.5"),
        )

That assignment has two effects:

Result	Behavior
Python attribute	Usable through `self.planner`.
workflow registry	Visible through `named_agents()` and `repr(workflow)`.

named_agents() returns a tuple sorted by attribute name. When an attribute is deleted, Workflow.__delattr__ removes the matching registered item from _agents.

Module Tree And Provider Placement#

v1.3.0 also registers nested Workflow attributes in _children:

class Team(Workflow):
    def __init__(self, provider: Provider) -> None:
        super().__init__()
        self.research = ResearchWorkflow(provider)
        self.writer = AgentParam(
            Session.from_agent_prompt("Write the result."),
            provider,
        )

API	Scope
`named_agents()`	Direct `AgentParam` leaves on this node.
`named_children()`	Direct nested `Workflow` / `Agent` children.
`modules()`	This workflow followed by all descendants in depth-first pre-order.
`workflow.to(...)`	Direct agents on this node only; it does not recurse.

Provider placement is chainable:

agent_param.to(Provider(model="gpt-5.5", api_key="..."))
agent_param.to(provider="main")       # reads config now
agent_param.to(model="new-model")     # overlays the current Provider

workflow.to(Provider(model="gpt-5.5", api_key="..."))

# Rebind a nested tree explicitly.
for module in workflow.modules():
    module.to(Provider(model="gpt-5.5", api_key="..."))

provider="name" calls Provider.from_config(...) during .to(...); it is not deferred until the next model request. Provider itself has no .to() method.

Single-Agent To Multi-Agent#

The smallest runnable path can use the preset flow.Agent directly:

from rath import flow
from rath.llm import Provider

agent = flow.Agent(
    system_prompt="Answer clearly.",
    provider=Provider(api_key="sk-...", model="gpt-5.5"),
)

out = agent(user_session)

For multiple roles, define each role as an AgentParam and call run_session_loop(...) step by step in forward(...):

from rath.flow import AgentParam, Provider, Workflow
from rath.session import Session, run_session_loop


class ReviewWorkflow(Workflow):
    def __init__(self, provider: Provider):
        super().__init__()
        self.writer = AgentParam(
            Session.from_agent_prompt("Write a first draft."),
            provider,
        )
        self.reviewer = AgentParam(
            Session.from_agent_prompt("Review the draft and tighten it."),
            provider,
        )

    def forward(self, session: Session) -> Session:
        draft = run_session_loop(
            session,
            self.writer.agent_session,
            agent_provider=self.writer.provider,
        )
        return run_session_loop(
            draft,
            self.reviewer.agent_session,
            agent_provider=self.reviewer.provider,
        )

The first loop output becomes the second loop input. The session graph records the parents for each loop, and the sandbox handle moves from input session to output session.

Session Is The Composition Unit#

Workflow instances communicate through Session. That keeps several composition patterns consistent:

Pattern	Code shape	Use case
Sequential call	`s = step_a(s); s = step_b(s)`	Roles work in a fixed order.
Branching exploration	`left = s.fork(); right = s.fork()`	Derive multiple candidate paths from the same context.
Session-level parallelism	`left = pool.submit(...); right = pool.submit(...)`	Send multiple forked sessions to different agents at the same time.
Detach from history	`clean = s.detach()`	Reuse content while cutting lineage.
Compress context	`compressor(s)`	Shorten history before the next stage.
Nested workflow	`s = self.child.forward(s)`	Split complex flows into smaller modules.

All of these operations still revolve around the session graph. fork(), detach(), run_session_loop(...), and run_session_compress(...) write lineage to output sessions; tool results remain as chunks in the session table; sandbox lifecycle is owned and transferred by the session.

Session-Level Parallelism#

OpenRath multi-agent parallelism is based on session branches, not a special scheduling DSL. After an upstream agent produces a session, use fork() to derive branches, then use normal Python concurrency tools to send those branches to different agents.

from concurrent.futures import ThreadPoolExecutor
from rath.session import Session


def forward(self, session: Session) -> Session:
    analysed = run_session_loop(
        session,
        self.analyst.agent_session,
        agent_provider=self.analyst.provider,
        tools=[market_tool],
    )

    bear_input = analysed.fork()
    bull_input = analysed.fork()

    with ThreadPoolExecutor(max_workers=2) as pool:
        bear_future = pool.submit(
            run_session_loop,
            bear_input,
            self.researcher_bear.agent_session,
            agent_provider=self.researcher_bear.provider,
            tools=None,
        )
        bull_future = pool.submit(
            run_session_loop,
            bull_input,
            self.researcher_bull.agent_session,
            agent_provider=self.researcher_bull.provider,
            tools=None,
        )
        bear_session = bear_future.result()
        bull_session = bull_future.result()

    def last_assistant_text(s: Session) -> str:
        return s.text() or ""

    trader_input = Session.from_user_message(
        "Combine the two research branches.\n\n"
        f"Bear branch:\n{last_assistant_text(bear_session)}\n\n"
        f"Bull branch:\n{last_assistant_text(bull_session)}"
    ).to("local")

    return run_session_loop(
        trader_input,
        self.trader.agent_session,
        agent_provider=self.trader.provider,
        tools=None,
    )

This pattern has three boundaries:

Boundary	Notes
lineage	Both forked sessions keep the same parent, and later loop outputs record their own agent parent.
sandbox	If the source has an open sandbox, `fork()` shares that handle by reference count. If no handle is open, branches inherit the backend target and open lazily.
aggregation	`Session.merge(other)` concatenates rows and keeps the left-hand sandbox; workflows may still choose to summarize branch outputs into a new user message. The example above builds the next input from the last assistant text in each branch.

OpenRath’s parallel unit is therefore the session. Tool stream concurrency belongs to the backend layer, and Provider.parallel_tool_calls belongs to LLM tool-call parameters; both are separate from session-level parallelism.

If branches write to the workspace, assign different directories explicitly. When the source has an open sandbox, forked branches share the same handle; when it only has spec=".", both branches still target the same host directory on lazy open. A safer pattern is to reset a branch-specific workspace after fork:

auth_input = session.fork().to("local", spec=".workspace/auth-branch")
data_input = session.fork().to("local", spec=".workspace/data-branch")

OpenSandbox follows the same rule: retarget a branch before tool execution if it needs an independent container or host bind path.

Preset Workflows#

OpenRath v1.3.0 provides four preset subclasses:

Class	Wraps	Best for
`Agent`	One `AgentParam`, one tools list, one `run_session_loop(...)`	Single-agent calls and quick tool integration.
`Compressor`	One `AgentParam`, one `run_session_compress(...)`	Compressing a long session into a new user-side session.
`Selector`	One routing `AgentParam`, one `select_session(...)`	Choosing the next self-describing workflow.
`EmptyWorkflow`	Identity `forward(...)`	Representing no match or a completed routing loop without returning `None`.

Agent.register_tool(...) deduplicates by tool name. Compressor asks the model to produce a new user message; the compressed result keeps session lineage and continues to hold the input session’s sandbox configuration and handle.

Dynamic Workflow Routing#

Workflow.description is routing metadata. Selector.forward(session, *workflows) builds a numbered menu from those descriptions, makes one model request with tools disabled, and returns a candidate workflow or an EmptyWorkflow:

selector = flow.Selector(provider)
triage = flow.Agent(
    "Triage the request.",
    provider,
    description="Classify a new support request",
)
resolve = flow.Agent(
    "Resolve the technical problem.",
    provider,
    description="Installation, configuration, or runtime errors",
)

chosen = selector.forward(session, triage, resolve)
if not isinstance(chosen, flow.EmptyWorkflow):
    session = chosen(session)

Selection is a decision boundary rather than a session transform:

Property	v1.2.1 behavior
return type	A `Workflow`; never `None` and never the selected workflow’s output `Session`.
completion	`EmptyWorkflow`, which is safe to call as an identity transform.
request	One completion, `tools=None`, `tool_choice="none"`.
parsing	First integer token; missing, negative, or out-of-range becomes no match.
lineage	`select_session(...)` creates no `Session` and stamps no lineage.
streaming	The `Selector(on_event=...)` argument is stored but is not forwarded by v1.3.0 selection.

Keep loops bounded because the model may repeatedly select a candidate. Candidate descriptions should be short, mutually distinguishable, and describe when the workflow applies rather than restating its system prompt.

Static Compilation#

Workflow.compile() walks the registered module tree without running a model or materializing a session:

compiled = workflow.compile()
manifest = compiled.manifest

models = manifest.provider_models()
kinds = manifest.provider_kinds()
problems = compiled.validate()

Each AgentResource records path, a Provider snapshot, has_memory, and agent_session_id. Each DynamicNode records path, kind, and reason. A Selector is dynamic because compile cannot predict its runtime target; its own router AgentParam remains a static resource.

CompiledWorkflow.validate() is intentionally narrow. It checks registered provider kinds and whether a credential resolves through the applicable offline resolver. It does not check model existence, network access, sandbox configuration, tools, memory health, or Selector candidates. LiteLLM is treated as satisfiable because it may resolve vendor-specific credentials internally.

The manifest is a compile-time snapshot. The wrapper still calls the current underlying workflow, so change the module tree or Provider placement only after planning to recompile.

As a context manager, CompiledWorkflow acquires every distinct reachable memory store once and releases them in reverse order. Providers have no lifecycle, and sandbox handles still open lazily from sessions.

Nested Workflow#

A nested workflow is still ordinary Python composition:

class EngineeringProjectWorkflow(Workflow):
    def __init__(self, provider: Provider) -> None:
        super().__init__()
        self.lead = AgentParam(Session.from_agent_prompt(LEAD_ENGINEER_SYSTEM), provider)
        self._squad = FeatureSquadWorkflow(provider)
        self._qa = QualityAssuranceWorkflow(provider)

    def forward(self, session: Session) -> Session:
        s = run_session_loop(
            session,
            self.lead.agent_session,
            agent_provider=self.lead.provider,
            tools=None,
        )
        s = self._squad.forward(s)
        return self._qa.forward(s)

In v1.3.0, self._squad and self._qa register as child workflows. Runtime order still comes only from explicit forward(...) calls; registration makes the tree inspectable and compilable, not automatically executable.

Sequential Multi-Agent Pattern#

A fixed-order multi-role flow usually looks like:

analyst
  researcher_bear
  researcher_bull
  trader
  risk_pm

The first stage may inject a domain tool; later stages read the tool result and assistant content already stored in the session. External tools can be given to a single role, and their results pass to later roles through the session.

Public examples require users to set their own API keys explicitly so a default key is not mistaken for a product capability.

Hierarchical Composition Pattern#

Hierarchical composition keeps parent and child workflows explicit:

Level	Workflow	Execution
L1	`EngineeringProjectWorkflow`	lead plan -> feature squad -> QA.
L2	`FeatureSquadWorkflow`	architect -> backend pair -> frontend.
L3	`BackendPairWorkflow`	backend auth -> backend data.
QA	`QualityAssuranceWorkflow`	Tests and risk checks based on the full session.

The example shows how to organize complex engineering work: each workflow owns its local sequence, the parent workflow chains child workflows, and all stages share the same session-passing chain.

Tool And Sandbox Boundaries#

Inside a workflow, tools and sandbox still take effect through run_session_loop(...):

Item	Where it happens
Tool list merge	At the start of each `run_session_loop(...)`.
Tool call record	Written to the output session as a `tool_result` chunk.
sandbox handle	Shared from the input session via `bind_sandbox` (refcount + 1); the input keeps its reference.
sandbox backend spec	Stored on the output session.
lineage	Output session records both the user session and agent session as parents.

A workflow can therefore give different roles different tools; the same sandbox can move through multiple roles with the session; later agents can see results produced by earlier tools.

Call Path#

workflow(session)
  Workflow.__call__
  subclass.forward(session)
  run_session_loop or child workflow
  returned Session carries new chunks, sandbox, lineage

When using the preset Agent:

flow.Agent.forward(session)
  run_session_loop(
    user_session=session,
    agent_session=self.agent.agent_session,
    agent_provider=self.agent.provider,
    tools=self.tools,
  )

When using the preset Compressor:

flow.Compressor.forward(session)
  run_session_compress(
    user_session=session,
    agent_session=self.agent.agent_session,
    agent_provider=self.agent.provider,
  )

When using the preset Selector:

flow.Selector.forward(session, *workflows)
  select_session(session, routing_prompt, *descriptions)
  parse one zero-based index
  return workflows[index] or EmptyWorkflow()

Current Boundaries#

Behavior	Current implementation
attribute registration	Only attributes assigned to `AgentParam` values enter `_agents`.
child registration	Attributes assigned to `Workflow` instances enter `_children`.
deletion	`__delattr__` removes the same-named agent from `_agents`.
ordering	`named_agents()` sorts by attribute name.
base execution	`Workflow.forward(...)` raises `NotImplementedError`.
nested placement	`Workflow.to(...)` affects direct agents only; iterate `modules()` for the full tree.
compile boundary	The manifest is a static resource snapshot, not a predicted branch/loop/fork execution plan.
selector contract	`Selector.forward(...)` returns a `Workflow`, intentionally differing from the base return annotation.
async support	`Workflow.forward(...)` is currently synchronous.
scheduling policy	Ordering, branching, retries, and concurrency are expressed by the user in Python code.

Code Reading Checkpoints#

In workflow.py, check _agents, _children, and __setattr__.
In workflow.py, check the sorting rule in named_agents().
In agent.py, check how Agent.forward(...) calls run_session_loop(...).
In compressor.py, check how the compression workflow calls run_session_compress(...).
In example/01_hello_agent.py, check the preset flow.Agent path.
In example/08_compress.py, check the preset compression path.
In tests/flow/test_workflow_agent.py, check workflow registration and sandbox transfer tests.
In selector.py and session/select.py, check no-match parsing and the no-lineage selection boundary.
In compile.py, check static traversal, validation scope, and memory acquisition order.

Test Coverage#

Behavior	Tests
workflow registration and agent call	`tests/flow/test_workflow_agent.py`
selector routing and empty fallback	`tests/flow/test_selector.py`, `tests/session/test_select_session.py`
module tree, manifest, validation, lifecycle	`tests/flow/test_module_tree.py`, `test_resource_manifest.py`, `test_compile_validate.py`, `test_compile_lifecycle.py`
import contract	`tests/test_import.py`
session compressor live behavior	`tests/integration/test_session_compress_real.py`

Related Pages#

Page	Covers
AgentParam	Agent-side session, provider, and request assembly.
01 Hello Agent	Smallest preset `flow.Agent` path.
08 Compress	Preset compression workflow.
11 Dynamic Selector	LLM-routed `if` and `while` patterns.
12 Workflow Compile	Static manifest and preflight example.