`rath.flow`#

Workflow composition layer. Workflow organizes Session -> Session transforms, AgentParam stores the agent-side session and provider, and preset workflows wrap common paths.

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`

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.

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.

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`	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`	`forward(...)` calls `run_session_compress(...)`.

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

Agent memory helpers#

Method	Behavior
`Agent.remember_memory(content, metadata=None, namespace=None)`	Writes an explicit memory entry through the attached store.
`Agent.recall_memory(query, limit=...)`	Retrieves relevant memory entries.
`Agent.commit_memory(session, instruction=None)`	Asks the agent to extract durable memory from a session.
`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.

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[source]#

Collects attached AgentParam instances and subclasses run sessions here.

named_agents() → tuple[tuple[str, AgentParam], ...][source]#: Agent params registered via attribute assignment.

forward(session: Session) → Session[source]#: Subclasses orchestrate Sessions (blocking).

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

System session plus LLM options for run_session_loop.

property data: Mapping[str, Any]#: Read-only mapping of underlying agent_session, provider and memory.

forward(session: Session) → Session[source]#: Subclasses orchestrate Sessions (blocking).

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

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[source]#: Issue a MemoryOpFind against the bound store and return the result.

commit_memory(session: Session, *, wait: bool = False) → object[source]#: Commit session’s chat transcript into memory for extraction.

close() → None[source]#: 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)[source]#

forward(session: Session) → Session[source]#: Subclasses orchestrate Sessions (blocking).

← API Reference

rath.flow#