The Thinking Atom¶
CognitiveWorker is the atomic building block of the Bridgic Amphibious framework — a pure thinking unit that only decides what to do, never how to execute. Combined with the think_unit descriptor and error strategies, it forms the foundation of every amphibious agent.
In this tutorial, we'll build a travel planning assistant that uses different CognitiveWorkers for different thinking responsibilities: one analyzes destinations, another creates itineraries.
Initialize¶
In [1]:
Copied!
import os
model_name = os.environ.get("MODEL_NAME")
api_key = os.environ.get("API_KEY")
api_base = os.environ.get("BASE_URL")
import os model_name = os.environ.get("MODEL_NAME") api_key = os.environ.get("API_KEY") api_base = os.environ.get("BASE_URL")
In [ ]:
Copied!
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,
),
)
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, ), )
Let's define some travel-related tools.
In [3]:
Copied!
from bridgic.core.agentic.tool_specs import FunctionToolSpec
async def search_attractions(city: str) -> str:
"""Search for tourist attractions in a city"""
attractions = {
"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 attractions.get(city, f"No attraction data for {city}")
async def get_travel_cost(origin: str, destination: str) -> str:
"""Get estimated travel cost between two cities"""
return f"Estimated cost from {origin} to {destination}: $150-300 (train/flight)"
async def check_hotels(city: str, checkin: str, checkout: str) -> str:
"""Check hotel availability in a city for given dates"""
return f"3 hotels available in {city} from {checkin} to {checkout}, starting at $80/night"
search_attractions_tool = FunctionToolSpec.from_raw(search_attractions)
get_travel_cost_tool = FunctionToolSpec.from_raw(get_travel_cost)
check_hotels_tool = FunctionToolSpec.from_raw(check_hotels)
from bridgic.core.agentic.tool_specs import FunctionToolSpec async def search_attractions(city: str) -> str: """Search for tourist attractions in a city""" attractions = { "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 attractions.get(city, f"No attraction data for {city}") async def get_travel_cost(origin: str, destination: str) -> str: """Get estimated travel cost between two cities""" return f"Estimated cost from {origin} to {destination}: $150-300 (train/flight)" async def check_hotels(city: str, checkin: str, checkout: str) -> str: """Check hotel availability in a city for given dates""" return f"3 hotels available in {city} from {checkin} to {checkout}, starting at $80/night" search_attractions_tool = FunctionToolSpec.from_raw(search_attractions) get_travel_cost_tool = FunctionToolSpec.from_raw(get_travel_cost) check_hotels_tool = FunctionToolSpec.from_raw(check_hotels)
Part 1: CognitiveWorker — The Pure Thinking Unit¶
In the observe-think-act cycle, a CognitiveWorker handles the think phase. It receives context (what the agent knows) and decides what action to take next — but never executes the action itself.
Two Ways to Create a CognitiveWorker¶
Method 1: CognitiveWorker.inline() — for simple cases where a prompt string is enough.
In [23]:
Copied!
from bridgic.amphibious import CognitiveWorker
# Quick creation with a prompt string
simple_worker = CognitiveWorker.inline(
"Analyze the travel destination and suggest the best attractions to visit."
)
from bridgic.amphibious import CognitiveWorker # Quick creation with a prompt string simple_worker = CognitiveWorker.inline( "Analyze the travel destination and suggest the best attractions to visit." )
Method 2: Subclass CognitiveWorker — for custom observation and thinking logic.
In [24]:
Copied!
from bridgic.amphibious import CognitiveContext
class DestinationAnalyzer(CognitiveWorker):
"""A worker that analyzes destinations with custom observation."""
async def thinking(self) -> str:
return "Analyze the destination based on the observation and suggest a day-by-day plan."
async def observation(self, context: CognitiveContext):
"""Inject extra environmental information before thinking."""
return (
f"Current goal: {context.goal}\n"
f"Travel tip: Consider visiting attractions early morning to avoid crowds."
)
from bridgic.amphibious import CognitiveContext class DestinationAnalyzer(CognitiveWorker): """A worker that analyzes destinations with custom observation.""" async def thinking(self) -> str: return "Analyze the destination based on the observation and suggest a day-by-day plan." async def observation(self, context: CognitiveContext): """Inject extra environmental information before thinking.""" return ( f"Current goal: {context.goal}\n" f"Travel tip: Consider visiting attractions early morning to avoid crowds." )
The observation() method lets you inject additional context before the worker thinks. By default, it delegates to the agent-level observation. When you override it, the worker gets a custom perception of its environment.
Part 2: think_unit — Declarative Execution Configuration¶
A think_unit binds a CognitiveWorker with execution parameters and declares it as a class attribute on the agent. Each await self.think_unit triggers one full observe-think-act cycle.
In [25]:
Copied!
from bridgic.amphibious import AmphibiousAutoma, CognitiveContext, think_unit
class TravelPlanner(AmphibiousAutoma[CognitiveContext]):
# Declare think units as class attributes
analyzer = think_unit(
CognitiveWorker.inline("Analyze the destination and find key attractions. Respond in JSON format."),
max_attempts=3,
)
planner = think_unit(
CognitiveWorker.inline("Create a detailed day-by-day itinerary based on what we know. Respond in JSON format."),
max_attempts=5,
)
async def on_agent(self, ctx: CognitiveContext):
# Each await triggers one observe-think-act cycle
await self.analyzer
await self.planner
from bridgic.amphibious import AmphibiousAutoma, CognitiveContext, think_unit class TravelPlanner(AmphibiousAutoma[CognitiveContext]): # Declare think units as class attributes analyzer = think_unit( CognitiveWorker.inline("Analyze the destination and find key attractions. Respond in JSON format."), max_attempts=3, ) planner = think_unit( CognitiveWorker.inline("Create a detailed day-by-day itinerary based on what we know. Respond in JSON format."), max_attempts=5, ) async def on_agent(self, ctx: CognitiveContext): # Each await triggers one observe-think-act cycle await self.analyzer await self.planner
In [7]:
Copied!
agent = TravelPlanner(llm=llm, verbose=True)
result = await agent.arun(
goal="Plan a 3-day trip to Tokyo",
tools=[search_attractions_tool, get_travel_cost_tool, check_hotels_tool],
)
print(result)
agent = TravelPlanner(llm=llm, verbose=True) result = await agent.arun( goal="Plan a 3-day trip to Tokyo", tools=[search_attractions_tool, get_travel_cost_tool, check_hotels_tool], ) print(result)
[13:16:15.086] [Router] (_amphibious_automa.py:1543) Auto-detecting execution mode [13:16:15.087] [Router] (_amphibious_automa.py:1549) Detected AGENT mode [13:16:15.088] [Observe] (_amphibious_automa.py:840) _PromptWorker: None [13:16:20.161] [Think] (_amphibious_automa.py:846) _PromptWorker: finish=False, step=Starting to plan a 3-day trip to Tokyo. First, I need to gather information about key tourist attractions in Tokyo to help plan the itinerary. [13:16:20.163] [Act] (_amphibious_automa.py:852) _PromptWorker: { "content": "Starting to plan a 3-day trip to Tokyo. First, I need to gather information about key tourist attractions in Tokyo to help plan the itinerary.", "result": { "results": [ { "tool_id": "call_0", "tool_name": "search_attractions", "tool_arguments": { "city": "Tokyo" }, "tool_result": "Senso-ji Temple, Shibuya Crossing, Tokyo Tower, Meiji Shrine", "success": true, "error": null } ] }, "metadata": {}, "status": null } [13:16:20.163] [Observe] (_amphibious_automa.py:840) _PromptWorker: None [13:17:50.486] [Think] (_amphibious_automa.py:846) _PromptWorker: finish=True, step=Now that I have the key attractions (Senso-ji Temple, Shibuya Crossing, Tokyo Tower, Meiji Shrine), I'll structure a 3-day itinerary grouping nearby locations. Day 1 focuses on Asakusa (Senso-ji + Nakamise Street), Day 2 on Harajuku/Shibuya (Meiji Shrine + Shibuya Crossing), and Day 3 on Minato (Tokyo Tower + Odaiba). Hotel checks require specific dates, which aren't provided yet. Finalizing the itinerary now. [13:17:50.488] [Act] (_amphibious_automa.py:852) _PromptWorker: { "content": "Now that I have the key attractions (Senso-ji Temple, Shibuya Crossing, Tokyo Tower, Meiji Shrine), I'll structure a 3-day itinerary grouping nearby locations. Day 1 focuses on Asakusa (Senso-ji + Nakamise Street), Day 2 on Harajuku/Shibuya (Meiji Shrine + Shibuya Crossing), and Day 3 on Minato (Tokyo Tower + Odaiba). Hotel checks require specific dates, which aren't provided yet. Finalizing the itinerary now.", "result": null, "metadata": { "tool_calls": [] }, "status": null } ================================================== TravelPlanner-dc7b2ea4 | Completed Tokens: 860 | Time: 95.40s ================================================== Now that I have the key attractions (Senso-ji Temple, Shibuya Crossing, Tokyo Tower, Meiji Shrine), I'll structure a 3-day itinerary grouping nearby locations. Day 1 focuses on Asakusa (Senso-ji + Nakamise Street), Day 2 on Harajuku/Shibuya (Meiji Shrine + Shibuya Crossing), and Day 3 on Minato (Tokyo Tower + Odaiba). Hotel checks require specific dates, which aren't provided yet. Finalizing the itinerary now.
Conditional Loops with until()¶
Use until() to run a think_unit repeatedly until a condition is met. This is useful when the agent needs multiple cycles to complete a task.
In [11]:
Copied!
class IterativePlanner(AmphibiousAutoma[CognitiveContext]):
researcher = think_unit(
CognitiveWorker.inline(
"Research ONE aspect of the trip (attractions, costs, or hotels). "
"Focus on what hasn't been researched yet. Respond in JSON format."
),
max_attempts=10,
)
async def on_agent(self, ctx: CognitiveContext):
# Run until at least 3 research steps are completed
# Here, the mechanical judgment will automatically terminate
# when there are three records in the execution history.
await self.researcher.until(
lambda ctx: len(ctx.cognitive_history) >= 3,
)
agent = IterativePlanner(llm=llm)
result = await agent.arun(
goal="Research a 3-day trip to Kyoto: find attractions, estimate costs, and check hotels for June 15-18",
tools=[search_attractions_tool, get_travel_cost_tool, check_hotels_tool],
)
print(f"Goal: {result.get('goal')}")
print(f"Research History: {result.get('cognitive_history')}")
class IterativePlanner(AmphibiousAutoma[CognitiveContext]): researcher = think_unit( CognitiveWorker.inline( "Research ONE aspect of the trip (attractions, costs, or hotels). " "Focus on what hasn't been researched yet. Respond in JSON format." ), max_attempts=10, ) async def on_agent(self, ctx: CognitiveContext): # Run until at least 3 research steps are completed # Here, the mechanical judgment will automatically terminate # when there are three records in the execution history. await self.researcher.until( lambda ctx: len(ctx.cognitive_history) >= 3, ) agent = IterativePlanner(llm=llm) result = await agent.arun( goal="Research a 3-day trip to Kyoto: find attractions, estimate costs, and check hotels for June 15-18", tools=[search_attractions_tool, get_travel_cost_tool, check_hotels_tool], ) print(f"Goal: {result.get('goal')}") print(f"Research History: {result.get('cognitive_history')}")
Goal: Goal: Research a 3-day trip to Kyoto: find attractions, estimate costs, and check hotels for June 15-18
Research History: Execution History:
[Working Memory (0-2)]
[0] Starting the Kyoto trip research. Since no previous research has been done, I'll begin by searching for tourist attractions in Kyoto. This is a logical first step as understanding what to see helps inform other decisions like hotel location and budget.
Result: results=[ActionStepResult(tool_id='call_0', tool_name='search_attractions', tool_arguments={'city': 'Kyoto'}, tool_result='Fushimi Inari Shrine, Kinkaku-ji, Arashiyama Bamboo Grove', success=True, error=None)]
[1] Since attractions in Kyoto have already been researched, the next logical step is to check hotel availability for the specified dates (June 15-18). This will help determine accommodation options and costs, which is critical for trip planning. Travel costs between cities cannot be estimated yet without knowing the origin city, so hotels are prioritized.
Result: results=[ActionStepResult(tool_id='call_0', tool_name='check_hotels', tool_arguments={'city': 'Kyoto', 'checkin': '2024-06-15', 'checkout': '2024-06-18'}, tool_result='3 hotels available in Kyoto from 2024-06-15 to 2024-06-18, starting at $80/night', success=True, error=None)]
[2] The remaining aspect to research is travel costs. However, the origin city is not specified in the context. Assuming a common origin like Tokyo for estimation purposes, I'll proceed to calculate travel costs between Tokyo and Kyoto.
Result: results=[ActionStepResult(tool_id='call_0', tool_name='get_travel_cost', tool_arguments={'origin': 'Tokyo', 'destination': 'Kyoto'}, tool_result='Estimated cost from Tokyo to Kyoto: $150-300 (train/flight)', success=True, error=None)]
Filtering Available Tools¶
You can restrict which tools are visible to a think_unit using the tools parameter.
In [ ]:
Copied!
class FocusedPlanner(AmphibiousAutoma[CognitiveContext]):
# This worker can only see search_attractions
attraction_finder = think_unit(
CognitiveWorker.inline(
"Find the best attractions to visit. Respond in JSON format.",
verbose_prompt=True, # Log prompt of the cognitive worker
),
max_attempts=3,
)
# This worker can see all tools
full_planner = think_unit(
CognitiveWorker.inline(
"Create a complete travel plan with costs and hotels. Respond in JSON format.",
verbose_prompt=True, # # Log prompt of the cognitive worker
),
max_attempts=5,
)
async def on_agent(self, ctx: CognitiveContext):
await self.attraction_finder.until(
lambda ctx: True, # Run once
tools=["search_attractions"],
)
await self.full_planner
agent = FocusedPlanner(llm=llm, verbose=True)
result = await agent.arun(
goal="Plan a trip to Osaka",
tools=[search_attractions_tool, get_travel_cost_tool, check_hotels_tool],
)
print(result)
class FocusedPlanner(AmphibiousAutoma[CognitiveContext]): # This worker can only see search_attractions attraction_finder = think_unit( CognitiveWorker.inline( "Find the best attractions to visit. Respond in JSON format.", verbose_prompt=True, # Log prompt of the cognitive worker ), max_attempts=3, ) # This worker can see all tools full_planner = think_unit( CognitiveWorker.inline( "Create a complete travel plan with costs and hotels. Respond in JSON format.", verbose_prompt=True, # # Log prompt of the cognitive worker ), max_attempts=5, ) async def on_agent(self, ctx: CognitiveContext): await self.attraction_finder.until( lambda ctx: True, # Run once tools=["search_attractions"], ) await self.full_planner agent = FocusedPlanner(llm=llm, verbose=True) result = await agent.arun( goal="Plan a trip to Osaka", tools=[search_attractions_tool, get_travel_cost_tool, check_hotels_tool], ) print(result)
[13:40:17.603] [Router] (_amphibious_automa.py:1543) Auto-detecting execution mode [13:40:17.603] [Router] (_amphibious_automa.py:1549) Detected AGENT mode [13:40:17.604] [Observe] (_amphibious_automa.py:840) _PromptWorker: None [13:40:21.954] [Think] (_cognitive_worker.py:332) Message 1 (system, 274 tokens): Find the best attractions to visit. Respond in JSON format. # Available Tools (with parameters): • search_attractions: Search for tourist attractions in a city - city (string) [required] # Context Acquiring If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty. ## Field format: - **details**: [{field: "skills", index: 0}, ...] Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step) # Output Fields - **step_content**: Your analysis and reasoning for this step - **finish**: Set True when the sub-task is fully complete (default: False) - **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...] - **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}] [13:40:21.954] [Think] (_cognitive_worker.py:332) Message 2 (user, 27 tokens): Based on the context below, decide your next action. Goal: Plan a trip to Osaka Execution History: (none) [13:40:21.954] [Think] (_cognitive_worker.py:332) Total: 301 tokens (cumulative: 301) [13:40:21.958] [Think] (_amphibious_automa.py:846) _PromptWorker: finish=False, step=Starting to plan a trip to Osaka. The first step is to find tourist attractions in Osaka that would be worth visiting. I'll use the search_attractions tool to get a list of popular attractions in the city. [13:40:21.961] [Act] (_amphibious_automa.py:852) _PromptWorker: { "content": "Starting to plan a trip to Osaka. The first step is to find tourist attractions in Osaka that would be worth visiting. I'll use the search_attractions tool to get a list of popular attractions in the city.", "result": { "results": [ { "tool_id": "call_0", "tool_name": "search_attractions", "tool_arguments": { "city": "Osaka" }, "tool_result": "Osaka Castle, Dotonbori, Universal Studios Japan", "success": true, "error": null } ] }, "metadata": {}, "status": null } [13:40:21.961] [Observe] (_amphibious_automa.py:840) _PromptWorker: None [13:40:40.976] [Think] (_cognitive_worker.py:332) Message 1 (system, 351 tokens): Create a complete travel plan with costs and hotels. Respond in JSON format. # Available Tools (with parameters): • search_attractions: Search for tourist attractions in a city - city (string) [required] • get_travel_cost: Get estimated travel cost between two cities - origin (string) [required] - destination (string) [required] • check_hotels: Check hotel availability in a city for given dates - city (string) [required] - checkin (string) [required] - checkout (string) [required] # Context Acquiring If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty. ## Field format: - **details**: [{field: "skills", index: 0}, ...] Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step) # Output Fields - **step_content**: Your analysis and reasoning for this step - **finish**: Set True when the sub-task is fully complete (default: False) - **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...] - **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}] [13:40:40.976] [Think] (_cognitive_worker.py:332) Message 2 (user, 138 tokens): Based on the context below, decide your next action. Goal: Plan a trip to Osaka Execution History: [Working Memory (0-0)] [0] Starting to plan a trip to Osaka. The first step is to find tourist attractions in Osaka that would be worth visiting. I'll use the search_attractions tool to get a list of popular attractions in the city. Result: results=[ActionStepResult(tool_id='call_0', tool_name='search_attractions', tool_arguments={'city': 'Osaka'}, tool_result='Osaka Castle, Dotonbori, Universal Studios Japan', success=True, error=None)] [13:40:40.976] [Think] (_cognitive_worker.py:332) Total: 489 tokens (cumulative: 489) [13:40:40.979] [Think] (_amphibious_automa.py:846) _PromptWorker: finish=False, step=Now that attractions in Osaka are identified, the next steps are to estimate travel costs from a major origin city (assuming Tokyo as a common departure point) and check hotel availability for a typical 3-night stay. Placeholder dates (2024-07-01 to 2024-07-04) are used since specific dates weren't provided. [13:40:40.984] [Act] (_amphibious_automa.py:852) _PromptWorker: { "content": "Now that attractions in Osaka are identified, the next steps are to estimate travel costs from a major origin city (assuming Tokyo as a common departure point) and check hotel availability for a typical 3-night stay. Placeholder dates (2024-07-01 to 2024-07-04) are used since specific dates weren't provided.", "result": { "results": [ { "tool_id": "call_0", "tool_name": "get_travel_cost", "tool_arguments": { "origin": "Tokyo", "destination": "Osaka" }, "tool_result": "Estimated cost from Tokyo to Osaka: $150-300 (train/flight)", "success": true, "error": null }, { "tool_id": "call_1", "tool_name": "check_hotels", "tool_arguments": { "city": "Osaka", "checkin": "2024-07-01", "checkout": "2024-07-04" }, "tool_result": "3 hotels available in Osaka from 2024-07-01 to 2024-07-04, starting at $80/night", "success": true, "error": null } ] }, "metadata": {}, "status": null } [13:40:40.984] [Observe] (_amphibious_automa.py:840) _PromptWorker: None [13:40:55.656] [Think] (_cognitive_worker.py:332) Message 1 (system, 351 tokens): Create a complete travel plan with costs and hotels. Respond in JSON format. # Available Tools (with parameters): • search_attractions: Search for tourist attractions in a city - city (string) [required] • get_travel_cost: Get estimated travel cost between two cities - origin (string) [required] - destination (string) [required] • check_hotels: Check hotel availability in a city for given dates - city (string) [required] - checkin (string) [required] - checkout (string) [required] # Context Acquiring If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty. ## Field format: - **details**: [{field: "skills", index: 0}, ...] Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step) # Output Fields - **step_content**: Your analysis and reasoning for this step - **finish**: Set True when the sub-task is fully complete (default: False) - **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...] - **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}] [13:40:55.656] [Think] (_cognitive_worker.py:332) Message 2 (user, 346 tokens): Based on the context below, decide your next action. Goal: Plan a trip to Osaka Execution History: [Working Memory (0-1)] [0] Starting to plan a trip to Osaka. The first step is to find tourist attractions in Osaka that would be worth visiting. I'll use the search_attractions tool to get a list of popular attractions in the city. Result: results=[ActionStepResult(tool_id='call_0', tool_name='search_attractions', tool_arguments={'city': 'Osaka'}, tool_result='Osaka Castle, Dotonbori, Universal Studios Japan', success=True, error=None)] [1] Now that attractions in Osaka are identified, the next steps are to estimate travel costs from a major origin city (assuming Tokyo as a common departure point) and check hotel availability for a typical 3-night stay. Placeholder dates (2024-07-01 to 2024-07-04) are used since specific dates weren't provided. Result: results=[ActionStepResult(tool_id='call_0', tool_name='get_travel_cost', tool_arguments={'origin': 'Tokyo', 'destination': 'Osaka'}, tool_result='Estimated cost from Tokyo to Osaka: $150-300 (train/flight)', success=True, error=None), ActionStepResult(tool_id='call_1', tool_name='check_hotels', tool_arguments={'city': 'Osaka', 'checkin': '2024-07-01', 'checkout': '2024-07-04'}, tool_result='3 hotels available in Osaka from 2024-07-01 to 2024-07-04, starting at $80/night', success=True, error=Non... [13:40:55.656] [Think] (_cognitive_worker.py:332) Total: 697 tokens (cumulative: 1186) [13:40:55.658] [Think] (_amphibious_automa.py:846) _PromptWorker: finish=True, step=Compile the gathered information into a structured travel plan including attractions, travel costs, hotel options, and estimated total budget. Since all necessary data has been collected (attractions, travel cost range, hotel availability/pricing), the plan can now be finalized. [13:40:55.658] [Act] (_amphibious_automa.py:852) _PromptWorker: { "content": "Compile the gathered information into a structured travel plan including attractions, travel costs, hotel options, and estimated total budget. Since all necessary data has been collected (attractions, travel cost range, hotel availability/pricing), the plan can now be finalized.", "result": null, "metadata": { "tool_calls": [] }, "status": null } ================================================== FocusedPlanner-137b7f1b | Completed Tokens: 1487 | Time: 38.06s ================================================== Compile the gathered information into a structured travel plan including attractions, travel costs, hotel options, and estimated total budget. Since all necessary data has been collected (attractions, travel cost range, hotel availability/pricing), the plan can now be finalized.
As shown above, after configuring the tools, during the process of the agent running the thinking unit, only the information of the specified tools will be visible.
Part 3: ErrorStrategy — When Things Go Wrong¶
ErrorStrategy handles errors in the observe-think-act cycle itself — for example, a failing observation() hook, an LLM API timeout, or a Pydantic validation error. It does not handle tool execution errors; those are caught internally and returned to the LLM as error messages for it to reason about.
To demonstrate, we'll create a CognitiveWorker with a custom observation() that reads from an unstable data source — it fails on the first few attempts, simulating a flaky API.
In [ ]:
Copied!
from bridgic.amphibious import ErrorStrategy
# Simulate an unstable data source
obs_call_count = 0
class UnstableWorker(CognitiveWorker):
"""A worker whose observation() fails on the first 2 calls."""
async def observation(self, context: CognitiveContext):
global obs_call_count
obs_call_count += 1
if obs_call_count <= 2:
raise ConnectionError(
f"[Attempt {obs_call_count}] Failed to fetch live market data from API"
)
return "Live market data: Tokyo hotels avg $120/night, peak season surcharge +20%"
async def thinking(self) -> str:
return "Based on the live market data, recommend the best hotel to book."
from bridgic.amphibious import ErrorStrategy # Simulate an unstable data source obs_call_count = 0 class UnstableWorker(CognitiveWorker): """A worker whose observation() fails on the first 2 calls.""" async def observation(self, context: CognitiveContext): global obs_call_count obs_call_count += 1 if obs_call_count <= 2: raise ConnectionError( f"[Attempt {obs_call_count}] Failed to fetch live market data from API" ) return "Live market data: Tokyo hotels avg $120/night, peak season surcharge +20%" async def thinking(self) -> str: return "Based on the live market data, recommend the best hotel to book."
In [23]:
Copied!
# --- RAISE (default): fails immediately on first error ---
obs_call_count = 0
class RaiseAgent(AmphibiousAutoma[CognitiveContext]):
booker = think_unit(
UnstableWorker(),
max_attempts=5,
on_error=ErrorStrategy.RAISE
)
async def on_agent(self, ctx: CognitiveContext):
await self.booker
try:
agent = RaiseAgent(llm=llm, verbose=True)
result = await agent.arun(goal="Book a hotel in Tokyo", tools=[check_hotels_tool])
print(result)
except RuntimeError as e:
print(e)
# --- RAISE (default): fails immediately on first error --- obs_call_count = 0 class RaiseAgent(AmphibiousAutoma[CognitiveContext]): booker = think_unit( UnstableWorker(), max_attempts=5, on_error=ErrorStrategy.RAISE ) async def on_agent(self, ctx: CognitiveContext): await self.booker try: agent = RaiseAgent(llm=llm, verbose=True) result = await agent.arun(goal="Book a hotel in Tokyo", tools=[check_hotels_tool]) print(result) except RuntimeError as e: print(e)
[14:49:56.331] [Router] (_amphibious_automa.py:1543) Auto-detecting execution mode [14:49:56.331] [Router] (_amphibious_automa.py:1549) Detected AGENT mode Worker 'UnstableWorker' failed during observe-think-act cycle: [Attempt 1] Failed to fetch live market data from API
In [22]:
Copied!
# --- RETRY: retries up to max_retries times, then succeeds ---
obs_call_count = 0
class RetryAgent(AmphibiousAutoma[CognitiveContext]):
booker = think_unit(
UnstableWorker(),
max_attempts=5,
on_error=ErrorStrategy.RETRY,
max_retries=3, # Will retry up to 3 times
)
async def on_agent(self, ctx: CognitiveContext):
await self.booker
try:
agent = RetryAgent(llm=llm, verbose=True)
result = await agent.arun(goal="Book a hotel in Tokyo", tools=[check_hotels_tool])
print(f"RETRY strategy → Success: {result}")
except Exception as e:
print(e)
# --- RETRY: retries up to max_retries times, then succeeds --- obs_call_count = 0 class RetryAgent(AmphibiousAutoma[CognitiveContext]): booker = think_unit( UnstableWorker(), max_attempts=5, on_error=ErrorStrategy.RETRY, max_retries=3, # Will retry up to 3 times ) async def on_agent(self, ctx: CognitiveContext): await self.booker try: agent = RetryAgent(llm=llm, verbose=True) result = await agent.arun(goal="Book a hotel in Tokyo", tools=[check_hotels_tool]) print(f"RETRY strategy → Success: {result}") except Exception as e: print(e)
[14:49:45.351] [Router] (_amphibious_automa.py:1543) Auto-detecting execution mode [14:49:45.352] [Router] (_amphibious_automa.py:1549) Detected AGENT mode [14:49:45.352] [Observe] (_amphibious_automa.py:840) UnstableWorker: Live market data: Tokyo hotels avg $120/night, peak season surcharge +20% [14:49:45.485] [Observe] (_amphibious_automa.py:840) UnstableWorker: Live market data: Tokyo hotels avg $120/night, peak season surcharge +20% [14:49:45.486] [Observe] (_amphibious_automa.py:840) UnstableWorker: Live market data: Tokyo hotels avg $120/night, peak season surcharge +20% Worker 'UnstableWorker' failed after 4 retries: Model operation `astructured_output` failed with non-recoverable error
In these examples, the observation() hook fails with a ConnectionError on the first 2 calls, simulating a flaky API. This error occurs in the OTC cycle itself — not inside a tool.
RAISE: The firstConnectionErrorimmediately crashes the agent. Use when OTC failures are unrecoverable.RETRY: The framework retries the entire observe-think-act cycle up tomax_retriestimes. After 2 failed attempts, the 3rd succeeds and the agent completes normally. Use for transient errors like network timeouts.IGNORE: Silently skips the failed cycle and continues to the nextmax_attemptsiteration. Use when individual cycle failures are acceptable.
What have we learnt?¶
In this tutorial, we explored the fundamental building blocks of the Bridgic Amphibious framework:
- CognitiveWorker is a pure thinking unit in the observe-think-act cycle. It only decides what to do — execution is handled by the framework. Create one quickly with
CognitiveWorker.inline("prompt")or subclass it for customthinking()andobservation()logic. - think_unit is a declarative descriptor that binds a worker with execution parameters (
max_attempts,on_error,max_retries). Declare it as a class attribute on your agent. await self.think_unittriggers one complete observe-think-act cycle. Useuntil()to loop until a condition is met, andtools=/skills=to filter what the worker can see.- ErrorStrategy controls failure behavior:
RAISEstops on error,IGNOREskips errors,RETRYautomatically retries transient failures.
Next, we'll learn how to orchestrate think units using the two modes: on_agent for LLM-driven logic and on_workflow for deterministic steps.