rath.flow

Workflow composition layer. Normal transforming workflows organize Session -> Session paths, AgentParam stores agent-side state, and Selector is the explicit routing exception that returns a workflow choice.

Source

Module	Source
rath.flow.workflow	src/rath/flow/workflow.py
rath.flow.agent_param	src/rath/flow/agent_param.py
rath.flow.agent	src/rath/flow/agent.py
rath.flow.compressor	src/rath/flow/compressor.py
rath.flow.selector	src/rath/flow/selector.py
rath.flow.empty	src/rath/flow/empty.py
rath.flow.compile	src/rath/flow/compile.py

Public contract

Workflow

Method	Returns	Behavior
forward(session)	Session	Execution logic implemented by subclasses.
__call__(session)	Session	Calls forward(session).
named_agents()	tuple[tuple[str, AgentParam], ...]	Returns agent params registered as attributes.
named_children()	tuple[tuple[str, Workflow], ...]	Returns directly registered nested workflows.
modules()	list[Workflow]	Returns this workflow and descendants in depth-first pre-order.
to(target=None, *, provider=None, model=None)	Workflow	Rebinds direct AgentParams only and returns self.
compile()	CompiledWorkflow	Builds a static resource snapshot without running a model.

Workflow(description="...") stores an optional description used when the workflow is offered to a Selector.

When an AgentParam is assigned to a workflow as an attribute, Workflow.__setattr__ adds it to _agents.

AgentParam

Field	Type	Description
agent_session	Session	Agent/system transcript.
provider	Provider	Model and request parameters.
memory	MemoryStore | None	Optional memory store bound to the agent.

AgentParam.to(...) accepts an explicit Provider, provider="config-name", or model="...". A named provider is resolved immediately. There is no Provider.to() method.

Preset workflows

Class	Constructor arguments	Behavior
Agent	system_prompt, provider=None, tools=None, model=None, on_event=None, memory=None, memory_inject=None, commit_on_forward=False, description=""	Creates an agent session and stores provider/runtime options. model= is a shortcut for a default Provider; memory= attaches a MemoryStore or provider spec. forward(...) calls run_session_loop(...).
Compressor	compress_instruction, provider, on_event=None, description=""	forward(...) calls run_session_compress(...).
Selector	provider, select_instruction=..., description="", on_event=None	forward(session, *workflows) returns the chosen Workflow or EmptyWorkflow. In v1.3.0 the stored on_event callback is not forwarded.
EmptyWorkflow	description=""	forward(session) returns the input session unchanged.

Agent.register_tool(tool) adds tools and deduplicates by name. Agent.unregister_tool(tool_name) removes the tool with the same name.

Dynamic routing

Selector.forward(...) intentionally does not follow the normal Session -> Session return contract. It makes a routing decision; the caller then dispatches the returned workflow.

Result	Meaning
Candidate Workflow	The model selected its zero-based menu index.
EmptyWorkflow	No candidate, -1, missing integer, or out-of-range integer.

Descriptions are the selector’s routing surface. Empty descriptions are valid but usually provide too little information for a reliable choice.

Static compilation

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

Type	Key members
AgentResource	path, provider, has_memory, agent_session_id
DynamicNode	path, kind, reason
ResourceManifest	agents, dynamic_nodes, provider_models(), provider_kinds()
CompiledWorkflow	workflow, manifest, validate(...), callable/context-manager behavior

The manifest is a snapshot. validate() checks only provider registration and offline credential resolution. The compiled wrapper delegates execution to the current workflow; context entry/exit manages distinct bound memory stores.

Agent memory helpers

Method	Behavior
Agent.remember_memory(content, *, scope="user", category="preferences", wait=False)	Writes an explicit memory entry through the attached store.
Agent.recall_memory(query, *, top_k=4, target_uri=None)	Retrieves relevant memory entries.
Agent.commit_memory(session, *, wait=False)	Commits the session transcript through the attached store.
commit_on_forward=True	Runs a best-effort commit after each forward call.

Runnable workflow examples

Example	Path	Description
Hello agent	example/01_hello_agent.py	Minimal provider and Agent call.
Session lineage	example/02_session_lineage.py	Key-free fork/detach/merge graph mechanics.
Memory	example/09_memory.py	Key-free local memory plus optional model-assisted commit.
Provider variation	example/10_provider_variation.py	Provider config, Anthropic, embeddings, and VLM setup.
Dynamic selector	example/11_dynamic_selector.py	Model-routed branching and bounded loops.
Workflow compile	example/12_compile.py	Key-free static manifest and lifecycle inspection.

These examples use the public Workflow, AgentParam, Provider, and run_session_loop(...) APIs, so they are useful source references for multi-agent composition.

Autodoc

class rath.flow.Workflow(description: str = '')

Collects attached AgentParam instances and subclasses run sessions here.

named_agents() → tuple[tuple[str, AgentParam], ...]

Agent params registered directly on this workflow (sorted by name).

named_children() → tuple[tuple[str, Workflow], ...]

Nested Workflow/Agent children registered by attribute (sorted).

modules() → list[Workflow]

This workflow followed by every descendant (pre-order, depth-first).

to(target: Provider | None = None, *, provider: str | None = None, model: str | None = None) → Workflow

Rebind the provider on every registered AgentParam (chainable).

Fans AgentParam.to() out to each agent from named_agents(), so workflow.to(Provider(...)) / workflow.to(provider="name") / workflow.to(model="m") apply uniformly. A workflow with no agents is a no-op. A bare positional string is rejected (same rule as AgentParam.to()).

compile() → object

Return a CompiledWorkflow for this workflow.

A static pass over the module tree (P5.1) that builds a resource manifest for pre-flight validation, deterministic resource lifecycle, and inspection. Opt-in and non-breaking: the returned object is callable exactly like this workflow. Runs no model and materializes no session.

compile_plan(*, revision_id: UUID) → ExecutionPlan

Compile explicit @step boundaries into an immutable v2 plan.

inspect_resources() → object

Return the v1 static resource inventory without compiling a v2 plan.

forward(session: Session) → Session

Subclasses orchestrate Sessions (blocking).

class rath.flow.AgentParam(agent_session: Session, provider: Provider, memory: MemoryStore | None = None)

System session plus LLM options for run_session_loop.

to(target: Provider | None = None, *, provider: str | None = None, model: str | None = None) → AgentParam

Rebind this param’s Provider (chainable, returns self).

Type-dispatched, mirroring Session.to for sandboxes:

• ap.to(Provider(...)) — bind an explicit provider (positional);

• ap.to(provider="name") — resolve a config preset lazily;

• ap.to(model="m") — overlay just the model on the current provider.

The positional argument accepts only a Provider; a bare string is rejected because — unlike Session.to("local") (a sandbox backend name) — the LLM path has no unambiguous string form. Use provider="name" for a config preset instead.

property data: Mapping[str, Any]

Read-only mapping of underlying agent_session, provider and memory.

class rath.flow.Agent(system_prompt: str, provider: Provider | None = None, tools: list[FlowToolCall] | None = None, *, model: str | None = None, on_event: Callable[[RathLLMStreamDelta], None] | None = None, memory: MemoryStore | MemoryStoreSpec | str | None = None, memory_inject: MemoryInjectionPolicy | None = None, commit_on_forward: bool = False, description: str = '')

forward(session: Session) → Session

Subclasses orchestrate Sessions (blocking).

remember_memory(content: str, *, scope: str = 'user', category: str = 'preferences', wait: bool = False) → object

Persist a free-form note under memory://{scope}/memories/{category}/....

scope is intentionally permissive (user / agent / session) so user code can decide which namespace to target; the URI prefix is adapter-coupled and other backends may rewrite it. See OpenVikingBackend.

recall_memory(query: str, *, top_k: int = 4, target_uri: str | None = None) → object

Issue a MemoryOpFind against the bound store and return the result.

commit_memory(session: Session, *, wait: bool = False) → object

Commit session’s chat transcript into memory for extraction.

close() → None

Release the memory store reference acquired in __init__ (idempotent).

class rath.flow.Compressor(compress_instruction: str, provider: Provider, *, on_event: Callable[[RathLLMStreamDelta], None] | None = None, description: str = '')

forward(session: Session) → Session

Subclasses orchestrate Sessions (blocking).

class rath.flow.Selector(provider: Provider, *, select_instruction: str = 'You are a router. Given the conversation and a numbered menu of candidate workflows, reply with the single best index, or -1 if none applies / the task is already complete.', description: str = '', on_event: Callable[[RathLLMStreamDelta], None] | None = None)

Pick the next Workflow for a session.

Sibling of Agent / Compressor.

forward(session: Session, *workflows: Workflow) → Workflow

Return the workflow the model picks for session, or an EmptyWorkflow when no candidate fits / the session is complete.

NOTE: this deliberately deviates from the base forward(session) -> Session contract. Selector is a routing decision component, not a session transformer: it returns the chosen Workflow (never None), and the caller dispatches it (session = chosen(session)). Completion is signalled by returning an EmptyWorkflow (a no-op), detected via isinstance(result, EmptyWorkflow).

class rath.flow.EmptyWorkflow(description: str = '')

No-op workflow: forward returns the input session unchanged.

Returned by Selector when no candidate fits / the session is complete, so callers can dispatch unconditionally and detect completion via isinstance(result, EmptyWorkflow).

forward(session: Session) → Session

Subclasses orchestrate Sessions (blocking).

class rath.flow.CompiledWorkflow(workflow: Workflow)

Static, callable wrapper around a Workflow.

Produced by Workflow.compile(). It is callable exactly like the workflow — cw(session) delegates to workflow.forward — so compiling is opt-in and non-breaking. It also exposes the static ResourceManifest, the module tree, and a graph repr.

Compiling runs no model and materializes no session; it only walks the static module tree (P5.1) to build the manifest.

named_children()

The compiled workflow’s registered children (delegates).

validate(*, raise_on_error: bool = False) → list[str]

Pre-flight check every reachable provider (offline; no model call).

For each agent in the manifest, verify (1) its provider_kind is a registered chat-client kind, and (2) a credential resolves for it via the same Provider → env → config chain the client uses at construction — without building an SDK client or hitting the network.

Returns a list of human-readable problems (empty when clean). With raise_on_error=True, raises ValueError if any problem is found. This lets callers fail fast before a run instead of deep inside the first completion.

class rath.flow.ResourceManifest(agents: list[~rath.flow.compile.AgentResource] = <factory>, dynamic_nodes: list[~rath.flow.compile.DynamicNode] = <factory>)

The static resource inventory of a compiled workflow.

provider_models() → list[str]

Distinct, sorted provider model names reachable in the workflow.

provider_kinds() → list[str]

Distinct, sorted provider kinds reachable (None -> "openai").

class rath.flow.compile.AgentResource(path: str, provider: Provider, has_memory: bool, agent_session_id: str)

One reachable AgentParam in the compiled module tree.

class rath.flow.compile.DynamicNode(path: str, kind: str, reason: str)

A node whose runtime behavior compile cannot statically resolve.

← API Reference