Execution Tracing¶

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.

Every arun() can record exactly what happened — turn by turn — into an AgentTrace. Each yield primitive the run dispatches (ActionCall, HumanCall, LLMCall, EnterAgent, and the cognitive steps ThinkUnit / ThinkAgent) becomes one TraceStep in the trace history, in order.

Why trace?

Debugging — see the exact tool calls, arguments, results, and failures.
Optimization — inspect token spend and timing in the run metadata.
Auditing — persist a complete, replayable record of every step.

Tracing is controlled by two orthogonal arun knobs:

trace=True activates an in-memory AgentTrace, kept on agent._agent_trace after the run.
workdir="./.bridgic" materializes a run directory at <workdir>/runs/<run_id>/.

With both set, the trace persists incrementally to <workdir>/runs/<run_id>/trace.json.

Setup¶

We'll trace a small pure-workflow run. A pure workflow (only on_workflow is overridden) is deterministic and needs no LLM — which makes it the cleanest thing to actually run here while still producing a fully populated trace. (Everything below works identically for an agent / amphiflow run; those just need an LLM.)

First, declare a tool, an OTAContext that carries it, and a bare Context.

In [ ]:

  Copied!     
 
from bridgic.amphibious import OTAContext, Context


async def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 22 C in {city}"


# Small-loop context: declare the tools this run carries.
class WeatherOTAContext(OTAContext):
    pass


WeatherOTAContext.tool(get_weather)


# Big-loop context: free-form knowledge (bare here).
class WeatherContext(Context):
    pass
from bridgic.amphibious import OTAContext, Context async def get_weather(city: str) -> str: """Get the current weather for a city.""" return f"Sunny, 22 C in {city}" # Small-loop context: declare the tools this run carries. class WeatherOTAContext(OTAContext): pass WeatherOTAContext.tool(get_weather) # Big-loop context: free-form knowledge (bare here). class WeatherContext(Context): pass

Now the agent. We override only on_workflow, so RunMode.AUTO resolves to WORKFLOW — no LLM required. Each yield ActionCall(...) runs one tool and hands back a List[ToolResult].

In [ ]:

  Copied!     
 
from bridgic.amphibious import AmphibiousAutoma, ActionCall, RETURN


class WeatherWorkflow(AmphibiousAutoma[WeatherOTAContext, WeatherContext]):
    async def on_workflow(self, ota_context, context=None):
        tokyo = yield ActionCall("get_weather", city="Tokyo")    # List[ToolResult]
        london = yield ActionCall("get_weather", city="London")

        tokyo_val = tokyo[0].result if tokyo else "N/A"
        london_val = london[0].result if london else "N/A"
        yield RETURN(f"Tokyo: {tokyo_val} | London: {london_val}")
from bridgic.amphibious import AmphibiousAutoma, ActionCall, RETURN class WeatherWorkflow(AmphibiousAutoma[WeatherOTAContext, WeatherContext]): async def on_workflow(self, ota_context, context=None): tokyo = yield ActionCall("get_weather", city="Tokyo") # List[ToolResult] london = yield ActionCall("get_weather", city="London") tokyo_val = tokyo[0].result if tokyo else "N/A" london_val = london[0].result if london else "N/A" yield RETURN(f"Tokyo: {tokyo_val} | London: {london_val}")

Enable tracing¶

Pass trace=True to arun. After the run, the AgentTrace lives on agent._agent_trace; call .build() to get a flat dict.

{
    "goal":     str,              # the run's user_input
    "metadata": {...},            # agent_class, mode, run_id, timing, tokens, ...
    "history":  [TraceStep, ...]  # one TraceStep per recorded step, in order
}

In [ ]:

  Copied!     
 
agent = WeatherWorkflow(verbose=True)

result = await agent.arun(
    user_input="Check the weather in Tokyo and London.",
    trace=True,    # activate the in-memory AgentTrace
)
print(result)

trace = agent._agent_trace.build()
print("keys:", list(trace.keys()))        # ["goal", "metadata", "history"]
print("goal:", trace["goal"])             # the run's user_input
print("steps:", len(trace["history"]))
agent = WeatherWorkflow(verbose=True) result = await agent.arun( user_input="Check the weather in Tokyo and London.", trace=True, # activate the in-memory AgentTrace ) print(result) trace = agent._agent_trace.build() print("keys:", list(trace.keys())) # ["goal", "metadata", "history"] print("goal:", trace["goal"]) # the run's user_input print("steps:", len(trace["history"]))

Walk the history¶

trace["history"] is a list of TraceStep objects, one per dispatched primitive, in execution order. Each carries a name, the step_content, an output_type (a StepOutputType), and — for tool-calling steps — a list of RecordedToolCall under tool_calls. Each RecordedToolCall records the tool_name, tool_arguments, tool_result, success, and error.

In [ ]:

  Copied!     
 
from bridgic.amphibious import StepOutputType

for i, step in enumerate(trace["history"]):
    print(f"[{i}] name={step.name!r}  output_type={step.output_type.value}")
    if step.step_content:
        print(f"     step_content: {step.step_content[:80]}")
    for tc in step.tool_calls:        # List[RecordedToolCall]
        mark = "ok" if tc.success else "FAIL"
        print(f"     -> {tc.tool_name}({tc.tool_arguments}) [{mark}] = {tc.tool_result!r}")
        if tc.error:
            print(f"        error: {tc.error}")
from bridgic.amphibious import StepOutputType for i, step in enumerate(trace["history"]): print(f"[{i}] name={step.name!r} output_type={step.output_type.value}") if step.step_content: print(f" step_content: {step.step_content[:80]}") for tc in step.tool_calls: # List[RecordedToolCall] mark = "ok" if tc.success else "FAIL" print(f" -> {tc.tool_name}({tc.tool_arguments}) [{mark}] = {tc.tool_result!r}") if tc.error: print(f" error: {tc.error}")

StepOutputType is the discriminator for what kind of step a TraceStep is. Here every step is TOOL_CALLS (a deterministic ActionCall). You can branch on it to render each step kind differently.

In [ ]:

  Copied!     
 
tool_steps = [s for s in trace["history"] if s.output_type == StepOutputType.TOOL_CALLS]
print(f"{len(tool_steps)} tool-call step(s)")

# Flatten every recorded tool call across the whole run.
for s in tool_steps:
    for tc in s.tool_calls:
        print(f"{tc.tool_name}: {tc.tool_result}")
tool_steps = [s for s in trace["history"] if s.output_type == StepOutputType.TOOL_CALLS] print(f"{len(tool_steps)} tool-call step(s)") # Flatten every recorded tool call across the whole run. for s in tool_steps: for tc in s.tool_calls: print(f"{tc.tool_name}: {tc.tool_result}")

Persisting the trace¶

There are two ways to write a trace to disk.

Automatic — via `workdir`¶

When you set both trace=True and workdir, the AgentTrace persists itself incrementally to <workdir>/runs/<run_id>/trace.json. trace and workdir are orthogonal: workdir alone creates the run directory but writes nothing; trace alone keeps the trace in memory only.

In [ ]:

  Copied!     
 
from pathlib import Path

agent2 = WeatherWorkflow(verbose=False)
await agent2.arun(
    user_input="Check the weather in Paris.",
    trace=True,
    workdir="./.bridgic",   # -> ./.bridgic/runs/<run_id>/trace.json
)

run_id = agent2._agent_trace.build()["metadata"]["run_id"]
persisted = Path("./.bridgic/runs") / run_id / "trace.json"
print("persisted:", persisted, "->", persisted.exists())
from pathlib import Path agent2 = WeatherWorkflow(verbose=False) await agent2.arun( user_input="Check the weather in Paris.", trace=True, workdir="./.bridgic", # -> ./.bridgic/runs//trace.json ) run_id = agent2._agent_trace.build()["metadata"]["run_id"] persisted = Path("./.bridgic/runs") / run_id / "trace.json" print("persisted:", persisted, "->", persisted.exists())

Explicit — `save()` / `load()`¶

agent._agent_trace.save(path) writes the trace dict to any path. AgentTrace.load(path) reads it back as a plain dict (its history entries are plain dicts, not TraceStep objects).

In [ ]:

  Copied!     
 
from bridgic.amphibious import AgentTrace

agent._agent_trace.save("trace.json")

loaded = AgentTrace.load("trace.json")   # plain dict
print("keys:", list(loaded.keys()))
print("goal:", loaded["goal"])
print("first step:", loaded["history"][0]["name"], "/", loaded["history"][0]["output_type"])
from bridgic.amphibious import AgentTrace agent._agent_trace.save("trace.json") loaded = AgentTrace.load("trace.json") # plain dict print("keys:", list(loaded.keys())) print("goal:", loaded["goal"]) print("first step:", loaded["history"][0]["name"], "/", loaded["history"][0]["output_type"])

Inspecting metadata¶

trace["metadata"] holds run-level information: the agent identity, the resolved RunMode, the run_id, start/end timing, and resource spend. Note metadata["mode"] is the resolved run mode (here "workflow", since only on_workflow was overridden).

In [ ]:

  Copied!     
 
meta = trace["metadata"]
print("agent_class:  ", meta["agent_class"])
print("mode:         ", meta["mode"])          # resolved RunMode value
print("run_id:       ", meta["run_id"])
print("start_time:   ", meta.get("start_time_iso"))
print("end_time:     ", meta.get("end_time_iso"))
print("cost_time (s):", meta.get("cost_time"))
print("spent_tokens: ", meta.get("spent_tokens"))
meta = trace["metadata"] print("agent_class: ", meta["agent_class"]) print("mode: ", meta["mode"]) # resolved RunMode value print("run_id: ", meta["run_id"]) print("start_time: ", meta.get("start_time_iso")) print("end_time: ", meta.get("end_time_iso")) print("cost_time (s):", meta.get("cost_time")) print("spent_tokens: ", meta.get("spent_tokens"))

Reference¶

`TraceStep` fields¶

Field	Type	Description
`name`	`str`	Step source — the worker class name, or `workflow` / `human_call` / `llm_call` / `think_agent` / `enter_agent`
`step_content`	`str`	Think text, final answer, or a short step descriptor
`tool_calls`	`List[RecordedToolCall]`	Tool invocations made by this step
`observation`	`Optional[str]`	The observation / prompt feeding the step
`observation_hash`	`Optional[str]`	Fingerprint of the observation
`output_type`	`StepOutputType`	What kind of step this is
`structured_output`	`Optional[Dict]`	Serialized structured result (LLM / human / agent steps)
`structured_output_class`	`Optional[str]`	Qualified class name of the structured result
`llm_call_protocol`	`Optional[str]`	Set on `LLMCall` steps (the protocol used)
`think_agent_name`	`Optional[str]`	Set on `ThinkAgent` steps (the descriptor name)

`RecordedToolCall` fields¶

tool_name, tool_arguments, tool_result, success (bool), error (Optional[str]).

`StepOutputType` values¶

Value	Produced by
`TOOL_CALLS`	a step that executed tool calls (`ActionCall`, a tool-calling think)
`CONTENT_ONLY`	a content-only finish (no tool calls)
`LLM_CALL`	a `LLMCall` in `on_workflow`
`THINK_AGENT`	a `ThinkAgent` delegation
`HUMAN_CALL`	a `HumanCall`
`ENTER_AGENT`	an `EnterAgent` mode switch

Next Steps¶

Dual-Mode Orchestration — on_agent vs on_workflow, and switching with EnterAgent.
CognitiveWorker & think_unit — the atomic think unit in depth.
Built-in Tools — declare the shipped tools on your OTA context.

Execution Tracing¶

Setup¶

Enable tracing¶

Walk the history¶

Persisting the trace¶

Automatic — via workdir¶

Explicit — save() / load()¶

Inspecting metadata¶

Reference¶

TraceStep fields¶

RecordedToolCall fields¶

StepOutputType values¶

Next Steps¶

Automatic — via `workdir`¶

Explicit — `save()` / `load()`¶

`TraceStep` fields¶

`RecordedToolCall` fields¶

`StepOutputType` values¶