Dual-Mode Orchestration¶
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 can be orchestrated in two modes, and you choose by overriding one (or both) of two template methods:
on_agent— LLM-driven. You declare cognitive steps and let the model decide what to do at each turn. The body yieldsThinkUnit/ThinkAgent/RETURNonly; the LLM does the tool / human / LLM work inside a think unit.on_workflow— deterministic. You write the exact steps. The body yields atomic calls —ActionCall/HumanCall/LLMCall— plusEnterAgent/RETURN.
And one primitive bridges them: EnterAgent(goal=...) suspends a running workflow, runs a fresh on_agent sub-run scoped to a sub-goal, then resumes the workflow where it left off.
on_agent | on_workflow | |
|---|---|---|
| Who picks the next step | the LLM | you |
| Yields | ThinkUnit / ThinkAgent / RETURN | ActionCall / HumanCall / LLMCall / EnterAgent / RETURN |
| Needs an LLM | yes | no (for a pure workflow) |
| Best for | open-ended, adaptive tasks | known, repeatable processes |
Under the default RunMode.AUTO, the framework reads which methods you overrode: on_agent only resolves to AGENT, on_workflow only resolves to WORKFLOW, and both resolves to AMPHIFLOW.
A Scenario — a price monitor¶
We'll track a product's price across a couple of stores. The same domain lets us see all three orchestration shapes:
- Agent mode — the LLM decides which stores to check and how to summarize.
- Workflow mode — we spell out the exact stores and order.
- Switching — a workflow that checks prices deterministically, then hands an open-ended judgement call to the agent via
EnterAgent.
Initialize¶
Set up the LLM. The examples read credentials from environment variables, so set MODEL_NAME, API_KEY, and BASE_URL before running. (Pure workflow cells below need no LLM.)
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,
),
)
Tools and contexts¶
A tool is just an async function with a docstring. We declare two mock store lookups on the OTA context (the small loop) via OTAContext.tool — nothing is auto-injected, so the run carries exactly what we declare. The big-loop Context is free-form cross-turn knowledge; here it tracks the product we're monitoring and renders it into the prompt via summary().
async def amazon_price(product: str) -> str:
"""Look up the current price of a product on Amazon."""
return f"{product}: $1299 on Amazon"
async def ebay_price(product: str) -> str:
"""Look up the current price of a product on eBay."""
return f"{product}: $1249 on eBay"
from bridgic.amphibious import OTAContext, Context
# Small-loop context: declare the tools this run carries.
class PriceOTAContext(OTAContext):
pass
PriceOTAContext.tool(amazon_price)
PriceOTAContext.tool(ebay_price)
# Big-loop context: free-form knowledge, rendered into the prompt.
class PriceContext(Context):
product: str = ""
def summary(self, fields):
return f"Product under watch: {fields['product']}"
on_agent — the LLM decides¶
In Agent mode you declare a think unit (a CognitiveWorker) and orchestrate it in on_agent. The worker owns one method, thinking(): it assembles a prompt from the contexts, calls the model, and returns the call's natural result — the framework adapts that into a decision and executes any tool calls. on_agent yields ThinkUnit("name") to drive the cognitive loop (capped by max_attempts, ended when the LLM stops calling tools).
Note the scope rule: on_agent only allows ThinkUnit / ThinkAgent / RETURN. Atomic calls (ActionCall / HumanCall / LLMCall) are forbidden here — when the LLM needs a tool, a human, or an LLM call, that happens inside a think unit (the worker's tool-selection phase), not by yielding from on_agent.
from bridgic.amphibious import (
AmphibiousAutoma, CognitiveWorker, think_unit, ThinkUnit,
)
from bridgic.core.model.types import Message, Role
class PriceThink(CognitiveWorker):
async def thinking(self, ota_context, context=None):
# Fold the big-loop knowledge (context) ahead of the small-loop task;
# nothing is auto-injected — the worker decides what reaches the model.
knowledge = context.summary() if context is not None else ""
prompt = f"{knowledge}\n\n{ota_context.summary()}" if knowledge else ota_context.summary()
messages = [Message.from_text(prompt, role=Role.USER)]
# aselect_tool returns (tool_calls, content); a reply with no
# tool calls is the finish.
return await self._llm.aselect_tool(
messages=messages,
tools=[t.to_tool() for t in ota_context.tools],
)
class PriceAgent(AmphibiousAutoma[PriceOTAContext, PriceContext]):
planner = think_unit(PriceThink(), max_attempts=5)
async def on_agent(self, ota_context, context=None):
# Only ThinkUnit / ThinkAgent / RETURN are allowed in agent scope.
yield ThinkUnit("planner")
# Only on_agent is overridden, so AUTO resolves to AGENT mode.
agent = PriceAgent(verbose=True)
answer = await agent.arun(
llm=llm,
user_input="Find the cheapest store for the laptop and summarize.",
context=PriceContext(product="laptop"),
)
print(answer) # the finishing think step's step_content
print(agent.final_answer)
on_workflow — you decide¶
In Workflow mode you write the exact steps in on_workflow, yielding ActionCall to run a tool. result = yield ActionCall(name, **args) hands you back a List[ToolResult]; read each result off result[i].result. A pure workflow needs no LLM — only on_workflow is overridden, so AUTO resolves to WORKFLOW.
Workflow scope also allows HumanCall(prompt=, channel=) (returns a str) and LLMCall.chat / .structure_output / .tool_selector for deterministic human or LLM steps — both omitted here to keep the flow pure (and LLM-free).
from bridgic.amphibious import ActionCall, RETURN
class PriceWorkflow(AmphibiousAutoma[PriceOTAContext, PriceContext]):
async def on_workflow(self, ota_context, context=None):
amazon = yield ActionCall("amazon_price", product="laptop") # List[ToolResult]
ebay = yield ActionCall("ebay_price", product="laptop")
amazon_val = amazon[0].result if amazon else "N/A"
ebay_val = ebay[0].result if ebay else "N/A"
yield RETURN(f"{amazon_val} | {ebay_val}")
workflow = PriceWorkflow(verbose=True)
result = await workflow.arun(user_input="Compare laptop prices") # no llm needed
print(result)
EnterAgent — switching mid-workflow¶
Override both methods and AUTO resolves to AMPHIFLOW: the workflow runs deterministically, and you can hand an open-ended sub-task to the agent on demand with EnterAgent(goal=...).
EnterAgent is a mode-switch signal, not a function call. The dispatcher suspends the workflow generator, builds a fresh OTA context with goal as its user_input (carrying the OTA context class's declared tools; the big-loop Context is shared read-only), and runs on_agent as a sub-run. When that agent generator naturally exhausts, the workflow resumes at the next instruction.
It takes only goal= — it controls what sub-task the agent gets, not how it thinks. There is no worker= / max_attempts= / tools= / skills=; for fine-grained cognitive control, yield ThinkUnit(...) from inside on_agent.
from bridgic.amphibious import EnterAgent, ActionCall, ThinkUnit, RETURN
class PriceMonitor(AmphibiousAutoma[PriceOTAContext, PriceContext]):
analyst = think_unit(PriceThink(), max_attempts=5)
# LLM-driven sub-task: reused by EnterAgent during the workflow.
async def on_agent(self, ota_context, context=None):
yield ThinkUnit("analyst")
# Deterministic spine, with one open-ended hand-off to the agent.
async def on_workflow(self, ota_context, context=None):
yield ActionCall("amazon_price", product="laptop")
yield ActionCall("ebay_price", product="laptop")
# Suspend the workflow; the agent runs a fresh OTA sub-run scoped
# to this goal, then control returns here.
yield EnterAgent(
goal="Decide whether the laptop price is a good deal and explain why.",
)
# Resumes after the agent generator exhausts.
yield RETURN("Price check complete.")
# Both on_agent and on_workflow are overridden, so AUTO resolves to
# AMPHIFLOW. EnterAgent needs the LLM for its agent sub-run.
monitor = PriceMonitor(verbose=True)
result = await monitor.arun(
llm=llm,
user_input="Monitor the laptop price.",
context=PriceContext(product="laptop"),
)
print(result)
When to use which¶
| Mode | Override | AUTO resolves to | Yields | LLM | Use it when |
|---|---|---|---|---|---|
| Agent | on_agent | AGENT | ThinkUnit / ThinkAgent / RETURN | required | the path is open-ended; let the model adapt |
| Workflow | on_workflow | WORKFLOW | ActionCall / HumanCall / LLMCall / EnterAgent / RETURN | optional | the steps are known and repeatable |
| Switch | both | AMPHIFLOW | workflow yields + EnterAgent to a sub-run | required | a deterministic spine with open-ended pockets |
Rule of thumb: start with a workflow for the parts you can pin down, and reach for EnterAgent (or a full agent mode) exactly where the task needs the model's judgement.
Next Steps¶
- CognitiveWorker & think_unit — the atomic think unit (
thinking(), loops, hooks) in depth. - RunMode —
AGENT/WORKFLOW/AMPHIFLOW/AUTOand the step-level fallback mechanism.