CognitiveWorker & think_unit¶
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.
CognitiveWorker is the framework's atomic in-process think unit — one observe-think-act cycle anchored on a BaseLlm. It owns exactly one job: the thinking step. It decides what to do; the framework handles observe before it and act after it.
A worker has a single override point — async def thinking(self, ota_context, context=None). You assemble a prompt from the two contexts, call self._llm however the model needs, and return that call's natural result. The framework's _assemble_decision adapts whatever you return into a flat decision (think text + tool calls).
think_unit(worker, ...) is the declarative wrapper: a descriptor you place as a class variable, then drive from on_agent via yield ThinkUnit("name"). It carries the thinking-orchestration knobs (until / max_attempts / on_error / max_retries) — never tools.
This tutorial builds a small travel-planning assistant to exercise every shape: tool-calling, content-only, structured output, looping, and error strategies.
Initialize¶
First, set up the LLM. The examples read credentials from environment variables, so set MODEL_NAME, API_KEY, and BASE_URL before running.
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,
),
)
Define a tool and the two contexts¶
A tool is just an async function with a docstring. Tools are declared on the OTA context via OTAContext.tool(...) (decorator or call) — nothing is auto-injected. The worker's thinking() reads them off ota_context.tools.
Every amphibious agent is parameterized by two contexts — AmphibiousAutoma[OTAContext, Context]:
OTAContext(small loop, framework-owned) — this run'suser_input, the observe-think-act round trace (ota_record), and thetoolsit carries.Context(big loop, free-form) — cross-turn knowledge; overridesummary()to render it into the prompt.
from bridgic.amphibious import OTAContext, Context
async def search_attractions(city: str) -> str:
"""Search for tourist attractions in a city."""
data = {
"Tokyo": "Senso-ji Temple, Shibuya Crossing, Tokyo Tower, Meiji Shrine",
"Kyoto": "Fushimi Inari Shrine, Kinkaku-ji, Arashiyama Bamboo Grove",
"Osaka": "Osaka Castle, Dotonbori, Universal Studios Japan",
}
return data.get(city, f"No attraction data for {city}")
# Small-loop context: declare the tools this run carries.
class TravelOTAContext(OTAContext):
pass
TravelOTAContext.tool(search_attractions)
# Big-loop context: free-form knowledge (bare here — nothing extra to render).
class TravelContext(Context):
pass
A basic worker — tool-calling with aselect_tool¶
Subclass CognitiveWorker and implement thinking(). There is no default thinking() (and no inline()/from_prompt() shortcuts) — every subclass implements it.
Here thinking() builds messages from ota_context.summary(), hands the declared tools to the model, and returns aselect_tool's natural result — a (tool_calls, content) tuple. _assemble_decision turns that into a tool-calling decision; the framework runs the tools, feeds the result back as the next round's observation, and loops. A reply with no tool calls is the finish.
from bridgic.amphibious import AmphibiousAutoma, CognitiveWorker, think_unit, ThinkUnit
from bridgic.core.model.types import Message, Role
class TravelThink(CognitiveWorker):
async def thinking(self, ota_context, context=None):
messages = [Message.from_text(ota_context.summary(), role=Role.USER)]
# aselect_tool returns (tool_calls, content) — tool_calls FIRST.
return await self._llm.aselect_tool(
messages=messages,
tools=[t.to_tool() for t in ota_context.tools],
)
Wrap the worker in think_unit as a class variable, then drive it from on_agent with yield ThinkUnit("name"). The asend() result is the finishing think step's step_content — a str. max_attempts caps the OTA cycles. Only on_agent is overridden, so RunMode.AUTO resolves to AGENT.
class TravelAgent(AmphibiousAutoma[TravelOTAContext, TravelContext]):
planner = think_unit(TravelThink(), max_attempts=5)
async def on_agent(self, ota_context, context=None):
yield ThinkUnit("planner")
agent = TravelAgent(verbose=True)
answer = await agent.arun(
llm=llm,
user_input="Find the top attractions in Tokyo and Kyoto, then summarize.",
)
print(answer) # the finishing think step's step_content
print(agent.final_answer)
The shapes thinking() can return¶
thinking() returns the bridgic protocol's natural result; _assemble_decision adapts each shape:
thinking() returns | Produced by | Becomes |
|---|---|---|
Response | achat | content-only (step_content = reply text) |
(tool_calls, content) | aselect_tool | tool-calling (tool_calls FIRST) |
a pydantic BaseModel / dict | astructured_output | content-only; value serialized to JSON in step_content |
str | plain text / accumulated stream | content-only |
A decision with no tool calls IS the finish.
Content-only — return a Response from achat¶
For models without native function-calling (or when you just want a textual answer), call achat and return its Response. That is a content-only finish — step_content is the reply text and the loop ends immediately.
class SummaryThink(CognitiveWorker):
async def thinking(self, ota_context, context=None):
messages = [Message.from_text(
"Write a one-paragraph trip overview.\n\n" + ota_context.summary(),
role=Role.USER,
)]
# achat returns a Response -> content-only finish.
return await self._llm.achat(messages)
class SummaryAgent(AmphibiousAutoma[TravelOTAContext, TravelContext]):
summarizer = think_unit(SummaryThink(), max_attempts=1)
async def on_agent(self, ota_context, context=None):
text = yield ThinkUnit("summarizer") # the reply text
print(text)
answer = await SummaryAgent().arun(
llm=llm,
user_input="Overview of a 3-day Tokyo trip.",
)
print(answer)
Structured output — return a pydantic BaseModel from astructured_output¶
There is no output_schema knob — return a Pydantic model straight from thinking(). Use PydanticModel(model=Schema) as the constraint. _assemble_decision serializes the model into the decision's step_content (JSON), so the yield ThinkUnit(...) result is a JSON string — parse it with Schema.model_validate_json(result).
from pydantic import BaseModel, Field
from bridgic.core.model.protocols import PydanticModel
from bridgic.amphibious import RETURN
class TripPlan(BaseModel):
phases: list[str] = Field(description="Execution phases")
estimated_days: int = Field(description="Total days needed")
class PlannerThink(CognitiveWorker):
async def thinking(self, ota_context, context=None):
messages = [Message.from_text(
"Create a phased trip plan.\n\n" + ota_context.summary(),
role=Role.USER,
)]
# astructured_output returns a TripPlan instance.
return await self._llm.astructured_output(messages, PydanticModel(model=TripPlan))
class PlannerAgent(AmphibiousAutoma[TravelOTAContext, TravelContext]):
planner = think_unit(PlannerThink(), max_attempts=1)
async def on_agent(self, ota_context, context=None):
plan_json = yield ThinkUnit("planner") # JSON string
plan = TripPlan.model_validate_json(plan_json) # parse it back
yield RETURN(f"{plan.estimated_days} days across {len(plan.phases)} phases")
result = await PlannerAgent().arun(
llm=llm,
user_input="Plan a 3-day trip to Kyoto.",
)
print(result)
Declarative configuration — max_attempts and until¶
think_unit(...) owns the thinking-orchestration knobs:
max_attempts— caps the OTA cycles a singleThinkUnitruns (default1).until— a predicate over the context (e.g. number of rounds recorded). The unit loops until the predicate returns true, still bounded bymax_attempts.
Each yield ThinkUnit("name", ...) overlays the descriptor's defaults — pass until= / max_attempts= to override them for that single yield; None means "use the descriptor's value". (on_error / max_retries are descriptor-only — no per-yield overlay.)
class IterativeAgent(AmphibiousAutoma[TravelOTAContext, TravelContext]):
researcher = think_unit(TravelThink(), max_attempts=10)
async def on_agent(self, ota_context, context=None):
# Loop until at least 3 OTA rounds are recorded; per-call overrides allowed.
result = yield ThinkUnit(
"researcher",
until=lambda ota: len(ota.ota_record) >= 3,
max_attempts=20,
)
yield RETURN(result)
result = await IterativeAgent().arun(
llm=llm,
user_input="Research attractions across Tokyo, Kyoto, and Osaka.",
)
print(result)
Error strategies¶
ErrorStrategy governs failures in the observe-think-act cycle itself — a failing hook, an LLM API timeout, a validation error. (Tool-execution errors are handled separately: they are caught and fed back to the LLM as error text to reason about, not raised here.) The strategy lives on the descriptor:
ErrorStrategy.RAISE(default) — re-raise on the first error.ErrorStrategy.IGNORE— silently skip the failed cycle and continue.ErrorStrategy.RETRY— retry the cycle up tomax_retriestimes.
on_error and max_retries are descriptor-only — there is no per-yield overlay for them.
from bridgic.amphibious import ErrorStrategy
# Simulate a flaky data source in the worker's observation() hook.
obs_calls = 0
class FlakyThink(CognitiveWorker):
async def observation(self, ota_context, context=None):
global obs_calls
obs_calls += 1
if obs_calls <= 2:
raise ConnectionError(f"[attempt {obs_calls}] live data fetch failed")
return "Live data: Tokyo hotels avg $120/night."
async def thinking(self, ota_context, context=None):
messages = [Message.from_text(ota_context.summary(), role=Role.USER)]
return await self._llm.achat(messages)
RAISE (the default) crashes on the first ConnectionError:
obs_calls = 0
class RaiseAgent(AmphibiousAutoma[TravelOTAContext, TravelContext]):
worker = think_unit(FlakyThink(), max_attempts=5, on_error=ErrorStrategy.RAISE)
async def on_agent(self, ota_context, context=None):
yield ThinkUnit("worker")
try:
await RaiseAgent().arun(llm=llm, user_input="Recommend a hotel.")
except Exception as exc:
print("RAISE ->", exc)
RETRY retries the cycle up to max_retries times — the 3rd attempt here succeeds:
obs_calls = 0
class RetryAgent(AmphibiousAutoma[TravelOTAContext, TravelContext]):
worker = think_unit(
FlakyThink(),
max_attempts=5,
on_error=ErrorStrategy.RETRY,
max_retries=3,
)
async def on_agent(self, ota_context, context=None):
result = yield ThinkUnit("worker")
yield RETURN(result)
result = await RetryAgent().arun(llm=llm, user_input="Recommend a hotel.")
print("RETRY ->", result)
ErrorStrategy.IGNORE instead skips the failed cycle and moves on to the next max_attempts iteration — use it when individual cycle failures are acceptable.
A note on worker hooks¶
Beyond thinking(), a CognitiveWorker exposes three optional hooks that refine the cycle — all payload-free (read the current round's state off ota_context):
observation(self, ota_context, context=None)— perception before thinking; read/set viaota_context.obs_result.before_action(self, ota_context, context=None)— inspect/override the pending decision atota_context.think_result.after_action(self, ota_context, context=None)— react toota_context.action_result.
Each hook is either a coroutine returning _DELEGATE (chain to the matching AmphibiousAutoma-level hook) or a value, or an async generator (yield ActionCall / HumanCall / LLMCall, then optionally RETURN(value)). We only sketch them here — the full hook deep-dive lives in Custom OTA.
from bridgic.amphibious import CognitiveWorker, _DELEGATE
class HintedThink(CognitiveWorker):
async def observation(self, ota_context, context=None):
# Coroutine form: a plain value becomes this round's obs_result.
# Return _DELEGATE instead to fall through to the agent-level hook.
return "Tip: group nearby attractions to cut travel time."
async def before_action(self, ota_context, context=None):
# Read the pending decision off the round; _DELEGATE = leave it untouched.
_ = ota_context.think_result
return _DELEGATE
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],
)
Next Steps¶
- Dual-Mode Orchestration —
on_agentvson_workflow, and switching withEnterAgent. - Custom OTA — the observation / before_action / after_action hooks in depth, and
_DELEGATE. - Built-in Tools — declare the shipped tools on your OTA context.
- Execution Tracing — inspect every think and tool call after a run.