Skip to main content

Flows

View Markdown

The definition​

interface FlowDefinition<I : Any, O : Any> {
    val type: FlowType<I, O>
    fun execute(input: I, flow: FlowScope): O
}

Not suspend. The engine owns scheduling: the body blocks on FlowScope calls and the engine parks it durably. Read Determinism before writing one.

Sequence​

val first = flow.task(Echo.type, EchoInput("hi"), quick)
val second = flow.task(Echo.type, EchoInput(first.text + "!"), quick)

Fan-out​

val pending = statements.map { flow.startTask(Classify.type, it, options) }
val results = pending.map { it.await() }

Waiting for a person​

val approvals = flow.signals<Approval>("approval")
val approval = approvals.receive(30.minutes)
if (approval?.approved != true) { /* declined or timed out */ }

Sending, from the API or the CLI:

orchestrator.handle(AgentRun.type, RunId("agent-42")).signal("approval", Approval(true, "abiola"))
lertha flow signal --id agent-42 --name approval --input '{"approved":true,"by":"cli"}'

Sleeping​

flow.sleep(7.days)

A durable timer. The process can die and the timer still fires; a run can sleep for a year and cost nothing while it does.

Catching a failed task​

val observation = try {
    flow.task(RunTool.type, ToolInput(plan.tool, plan.args), tooling).observation
} catch (e: TaskFailed) {
    "Tool ${plan.tool} failed: ${e.message}"
}

e.state says whether the task failed, timed out or was cancelled. Never catch Throwable around a scope call: cancellation of the flow itself travels through that path.

Starting a flow​

val options = RunOptions(
    id = "agent-${conversationId}",
    queue = "lab",
    attemptTimeout = 1.hours,
    overallTimeout = 1.hours,
    retry = RetryPolicy.NONE,
)
val handle: FlowHandle<AgentOutput> = orchestrator.start(AgentRun.type, AgentInput(goal), options)

A FlowHandle is a RunHandle plus signal().

The agent shape​

Plan step, tool step, a human before the consequential action, loop until the planner says done. The planner and the tools are tasks; the loop and the waiting are the flow. The lab repository ships this as AgentRun, with the planner scripted so the lab is deterministic and free. Swap the script for a model call and nothing else changes.