Skip to content

control.md

Latest commit

 

History

History
200 lines (152 loc) · 6 KB

control.md

File metadata and controls

200 lines (152 loc) · 6 KB
Raw

Control

Graph is the control surface. Flow.start(...) creates a graph, and every Flow.step(graph) returns a fresh advanced snapshot. Save/load, rewind, branch, inject, and resume are all graph operations.

Step Loop

agent = rflow.Flow(rflow.OpenAIClient(model="gpt-5"), max_depth=2)
graph = agent.start(query)
while not graph.finished:
    graph = agent.step(graph)

agent.run(query) drives the same loop and returns graph.result(). agent.chat(messages) is the LLMClient interface; the latest user message becomes the query and the recursive loop runs under the hood.

Each step(graph) advances one observation-to-observation transition for every agent that is ready to move. A model turn is usually two steps: LLM call (obs -> LLMAction -> LLMOutput) and code execution (LLMOutput -> ExecAction -> CodeObservation). See node_model.md for the typed node flow.

Eager Children

By default, children advance in synchronized batches. If child A's current step takes 10 seconds and child B's current step takes 2 seconds, child B waits for that batch before starting its next step.

Set eager_children=True for a work-conserving drain after a parent awaits a launcher:

agent = rflow.Flow(
    rflow.OpenAIClient(model="gpt-5"),
    max_depth=2,
    child_max_iters=20,
    max_concurrency=8,
    eager_children=True,
)

Children still do not run before the parent reaches await launch_subagents([...]). Once the parent is supervising, runnable children refill the worker pool until all waited-on descendants finish.

See examples/control/delegation/eager_children.py for a deterministic offline demo.

Save And Resume

A saved graph directory is the durable run:

graph.save("runs/deep_research")

resumed = rflow.Graph.load("runs/deep_research")
while not resumed.finished:
    resumed = agent.step(resumed)

For live checkpointing, save after every step. The same path is overwritten with the latest complete graph/run layout.

Rewind And Branch

Keep every Graph snapshot in a list and resume any one of them:

history = [agent.start(query)]
while not history[-1].finished:
    history.append(agent.step(history[-1]))

graph = history[-5]
while not graph.finished:
    graph = agent.step(graph)

Branch by copying or loading a graph and saving the result somewhere else:

branch = history[-5].copy(deep=True)
while not branch.finished:
    branch = agent.step(branch)
branch.save("runs/repair-branch")

Node Injection

Controllers can append typed nodes to a graph and commit them through the normal step loop. This is useful for budget nudges, human feedback, and forced finalization:

graph = graph.inject(
    target="root.worker",
    node=rflow.ExecOutput(
        output="Injected controller observation: answer now.",
        content="Injected controller observation: answer now.",
    ),
)
graph = agent.step(graph)

graph = graph.inject(
    target="root.worker",
    node=rflow.ExecAction(code='done("best available answer")'),
)
graph = agent.step(graph)

See injections.md and examples/control/controller_injection.py.

Delegation

Agents delegate through one launcher, which must be awaited:

# One child — still pass a one-item list of dict specs, and unpack the result.
[answer] = await launch_subagents([
    {"name": "single", "query": query, "inputs": {"data": data}},
])

# Many children in parallel — returns answers in spec order.
results = await launch_subagents([
    {"name": "a", "query": "...", "inputs": {"chunk": chunk_a}},
    {"name": "b", "query": "...", "inputs": {"chunk": chunk_b}},
])
  • Sequential dependent steps: chain one-item await launch_subagents([...]) calls, feeding each result into the next child's inputs.
  • Parallel independent work: pass every spec in one call so the engine schedules them concurrently.
  • Child data: put payloads in each spec's inputs dict. The child sees only its query and its own INPUTS.

Custom Runtime

Subclass Runtime and implement open(agent) to mint a backend:

class MyRuntime(rflow.Runtime):
    def open(self, agent: rflow.Graph) -> rflow.ReplBackend:
        return MyBackend(...)

Most users should pass LocalRuntime, DockerRuntime, or a sandbox runtime. See runtimes.md.

Custom Tools

Register tools on the runtime before constructing or stepping the flow:

@rflow.tool("Search files for a regex.")
def search(pattern: str, path: str = ".") -> str:
    ...

runtime = rflow.LocalRuntime(working_directory=".")
runtime.register_tool(search)
runtime.register_tools(rflow.FILE_TOOLS)
agent = rflow.Flow(rflow.OpenAIClient(model="gpt-5"), runtime=runtime)

Custom Prompt

For a fuller guide, see prompt_customization.md.

from rflow.prompts import DEFAULT_BUILDER

GUARDRAILS = """
- Verify before `done()`. Empty/zero/surprising results -> one sanity check first.
- Ask children for structured output when shape matters.
"""

agent = rflow.Flow(rflow.OpenAIClient(model="gpt-5"))
agent.prompt_builder = (
    DEFAULT_BUILDER
    .section("role", "You are a security auditor.", title="Role")
    .section("guardrails", GUARDRAILS, title="Guardrails", after="strategy")
)

You can also subclass Flow and override build_system_prompt, build_messages, format_exec_output, first_prompt, or step.

Walkthroughs