RunMode — Four Ways to Drive an Agent¶
Documentation update in progress
This Amphibious tutorial is currently outdated and may not match the bridgic-amphibious 0.2.0 API. We are actively updating it. Please use it as conceptual background only until the update is complete.
An amphibious agent has two execution surfaces: on_agent (the LLM decides what to do) and on_workflow (you decide what to do). RunMode selects which surface drives a run — and whether the two cooperate.
There are four modes:
| Mode | Driver | Surfaces used |
|---|---|---|
AGENT | the LLM | on_agent only |
WORKFLOW | you | on_workflow only |
AMPHIFLOW | a peer state machine | both — workflow with agent recovery |
AUTO (default) | auto-detected | resolved from which methods you override |
This tutorial walks each mode and then dives into the centerpiece: how AMPHIFLOW recovers from a failed workflow step.
AUTO resolution¶
You rarely pass a mode explicitly. The default is RunMode.AUTO, which inspects which template methods your class overrides and picks a mode for you:
| You override... | AUTO resolves to |
|---|---|
only on_agent | AGENT |
only on_workflow | WORKFLOW |
both on_agent and on_workflow | AMPHIFLOW |
All overridable template methods must be async generators (yield-driven). The framework validates this at class-creation time; if a body has no real yields, add if False: yield as an unreachable stub.
When you need to be explicit — or want a class that overrides both methods to run as a pure AGENT or pure WORKFLOW — pass mode=RunMode.AGENT / WORKFLOW / AMPHIFLOW to arun.
Setup¶
AGENT and AMPHIFLOW runs need an LLM (they run a CognitiveWorker). Pure WORKFLOW does not. We read credentials from environment variables, so set MODEL_NAME, API_KEY, and BASE_URL before running the LLM-backed cells.
import os
model_name = os.environ.get("MODEL_NAME")
api_key = os.environ.get("API_KEY")
api_base = os.environ.get("BASE_URL")
from bridgic.llms.openai import OpenAILlm, OpenAIConfiguration
llm = OpenAILlm(
api_key=api_key,
api_base=api_base,
timeout=30,
configuration=OpenAIConfiguration(
model=model_name,
temperature=0.0,
max_tokens=16384,
),
)
A tool, an OTAContext, and a Context¶
We model a tiny registration form: each step writes one field to a remote service. The small-loop OTAContext declares the tool the run carries (nothing is auto-injected); the big-loop Context is bare here.
The submit_field tool is deliberately flaky: the captcha field fails the first time it is attempted, so we have a concrete, deterministic step failure to recover from later.
from bridgic.amphibious import OTAContext, Context
# Module-level state so the simulated failure is deterministic and
# observable across calls within one run.
_attempts = {}
_submitted = {}
async def submit_field(field_name: str, value: str) -> str:
"""Submit one form field to the registration service.
The 'captcha' field fails on its first attempt (a transient,
captcha-like check) and succeeds on retry.
"""
_attempts[field_name] = _attempts.get(field_name, 0) + 1
if field_name == "captcha" and _attempts[field_name] == 1:
raise RuntimeError(
"captcha verification failed: the token was rejected"
)
_submitted[field_name] = value
return f"{field_name} accepted: {value!r}"
# Small-loop context: declare exactly the tools this run carries.
class FormOTAContext(OTAContext):
pass
FormOTAContext.tool(submit_field)
# Big-loop context: free-form cross-turn knowledge (bare here).
class FormContext(Context):
pass
AGENT mode — only on_agent¶
Override just on_agent and AUTO resolves to AGENT. The LLM drives: a CognitiveWorker's thinking() assembles a prompt, calls the model, and the framework turns the result into a decision (tool calls, or a finish). on_agent orchestrates the cognitive loop by yielding ThinkUnit("name").
from bridgic.amphibious import (
AmphibiousAutoma, CognitiveWorker, think_unit, ThinkUnit,
)
from bridgic.core.model.types import Message, Role
class FormThink(CognitiveWorker):
async def thinking(self, ota_context, context=None):
messages = [
Message.from_text(ota_context.summary(), role=Role.USER),
]
return await self._llm.aselect_tool(
messages=messages,
tools=[t.to_tool() for t in ota_context.tools],
)
class FormAgent(AmphibiousAutoma[FormOTAContext, FormContext]):
worker = think_unit(FormThink(), max_attempts=10)
async def on_agent(self, ota_context, context=None):
yield ThinkUnit("worker")
# Only on_agent is overridden, so AUTO resolves to AGENT mode.
agent = FormAgent(verbose=True)
answer = await agent.arun(
llm=llm,
user_input=(
"Register a user: submit username='john', "
"email='john@example.com'."
),
)
print(answer)
WORKFLOW mode — only on_workflow¶
Override just on_workflow and AUTO resolves to WORKFLOW. You spell out the exact steps; result = yield ActionCall(name, **args) runs one tool and hands you back a List[ToolResult]. A pure workflow needs no LLM.
(We avoid the flaky captcha field here so the deterministic run succeeds end to end — failure handling is the next section's job.)
from bridgic.amphibious import ActionCall, RETURN
class FormWorkflow(AmphibiousAutoma[FormOTAContext, FormContext]):
async def on_workflow(self, ota_context, context=None):
u = yield ActionCall(
"submit_field", field_name="username", value="john"
)
e = yield ActionCall(
"submit_field", field_name="email",
value="john@example.com",
)
u_val = u[0].result if u else "N/A"
e_val = e[0].result if e else "N/A"
yield RETURN(f"{u_val} | {e_val}")
# Reset the simulated service before the run.
_attempts.clear()
_submitted.clear()
workflow = FormWorkflow(verbose=True)
result = await workflow.arun(user_input="Register john") # no llm
print(result)
AMPHIFLOW mode — workflow with agent recovery¶
This is the centerpiece. Override both methods and AUTO resolves to AMPHIFLOW: a peer state machine runs your on_workflow deterministically, but when a step fails it can call on on_agent to recover — without you writing any try/except.
What happens when a step fails¶
When a yielded atomic Call (ActionCall / HumanCall / LLMCall) raises, the framework decides between two responses using a single counter:
consecutive_failures += 1. Every successful atomic Call resets this counter back to0.- If
consecutive_failures >= max_consecutive_fallbacks→ full fallback. The workflow generator is closed andon_agentruns for the rest of the task (inheriting the originaluser_input). - Otherwise → step-level recovery. The framework runs a bounded
on_agentsub-run — a fresh OTA episode whoseuser_inputdescribes the failed step and its error. The sub-run's conclusion is shaped into the failed step's return type andasend()-ed back into the suspended workflow, which resumes at the next instruction.
There is no injected tool and no toolset mutation — the recovery sub-run's own conclusion is the resolution. (A workflow generator-internal error — helper code between yields raising — can't be resumed in place, so it escalates straight to full fallback.)
max_consecutive_fallbacks (default 1) is the threshold that bounds consecutive recoveries before full fallback takes over.
The scenario¶
Our submit_field tool fails the first time it submits the captcha field. We run a four-step form workflow that includes that field, with max_consecutive_fallbacks=2.
on_workflowlists the deterministic steps (username → email → captcha → finish).on_agentis a small "fixer" think unit. When thecaptchastep raises, the framework spins up a bounded recovery sub-run against a goal describing that failure; the fixer's conclusion is fed back as the failed step's result, and the workflow continues.
Because max_consecutive_fallbacks=2, a single isolated failure stays step-level (1 < 2) and the deterministic flow survives. Only two failures in a row (with no successful step in between to reset the counter) would escalate to full fallback.
from bridgic.amphibious import RunMode
class FormFixer(CognitiveWorker):
"""Recovery brain: re-attempts the failed step."""
async def thinking(self, ota_context, context=None):
messages = [
Message.from_text(
"A registration step just failed. Read the failure "
"below, then re-submit the same field to recover.\n\n"
+ ota_context.summary(),
role=Role.USER,
),
]
return await self._llm.aselect_tool(
messages=messages,
tools=[t.to_tool() for t in ota_context.tools],
)
class FormAmphiflow(AmphibiousAutoma[FormOTAContext, FormContext]):
# The recovery sub-run drives this think unit.
fixer = think_unit(FormFixer(), max_attempts=5)
async def on_agent(self, ota_context, context=None):
# Bounded recovery sub-run: ota_context.user_input already
# describes the failed step and its error.
yield ThinkUnit("fixer")
async def on_workflow(self, ota_context, context=None):
yield ActionCall(
"submit_field", field_name="username", value="john"
)
yield ActionCall(
"submit_field", field_name="email",
value="john@example.com",
)
# This step raises on its FIRST attempt -> step-level recovery.
yield ActionCall(
"submit_field", field_name="captcha", value="7QF3"
)
yield RETURN("Registration complete.")
# Reset the simulated service so 'captcha' fails exactly once.
_attempts.clear()
_submitted.clear()
# Both methods overridden -> AUTO would pick AMPHIFLOW; we pass it
# explicitly for clarity, along with the fallback threshold.
agent = FormAmphiflow(verbose=True)
result = await agent.arun(
llm=llm,
user_input="Register john and clear the captcha.",
mode=RunMode.AMPHIFLOW,
max_consecutive_fallbacks=2,
)
print(result)
print("captcha attempts:", _attempts.get("captcha"))
Reading the trace¶
Step by step, with max_consecutive_fallbacks=2:
submit_field(username)succeeds → counter stays0.submit_field(email)succeeds → counter stays0.submit_field(captcha)raises on its first attempt →consecutive_failuresbecomes1. Since1 < 2, the framework runs the boundedon_agentrecovery sub-run against a goal describing the captcha failure. The fixer re-submits the captcha (now its second attempt, which succeeds); that conclusion isasend()-ed back as the failed step's result.- The workflow resumes at the next instruction and yields
RETURN("Registration complete.").
Had the counter reached 2 (two consecutive failures with no successful step between them), the workflow generator would have been closed and on_agent would have taken over for the rest of the run (full fallback). And note EnterAgent is orthogonal to all of this: it is your explicit mode-switch for an open-ended sub-task, not a failure recovery, and it never touches the counter.
The four modes at a glance¶
| Mode | Driver | Best for | Fallback |
|---|---|---|---|
AGENT | the LLM (on_agent) | open-ended, adaptive tasks | N/A |
WORKFLOW | you (on_workflow) | known, repeatable processes | N/A |
AMPHIFLOW | peer state machine (both) | robust hybrid execution | step-level recovery sub-run, then full fallback |
AUTO (default) | auto-detected from overrides | most subclasses | inherits from the resolved mode |
Rule of thumb: override only the method you need and let AUTO pick. Override both when you want a deterministic happy path that can lean on the LLM to recover from flaky steps — that is AMPHIFLOW.
Next Steps¶
- Dual-Mode Orchestration —
on_agentvson_workflow, and switching withEnterAgent. - CognitiveWorker & think_unit — the atomic think unit in depth.
- Built-in Tools — declare the shipped tools on your OTA context.
- Custom Contexts — shape the small-loop and big-loop state.
- Execution Tracing — inspect every observe-think-act round.