Phidata / Agno agent circuit breaker: loop detection and cost cap
Phidata rebranded to Agno in late 2024, but the framework’s core architecture — composable Agent classes, typed tool functions, multi-agent Team pipelines, and an LLM-agnostic model abstraction — stayed intact. That architecture is elegant for building reasoning agents quickly. It also creates a predictable failure mode: when a tool returns a “not found” result or an error string, the agent’s reasoning loop re-tries the same tool with slightly rephrased arguments, converges on the same failure, and does it again until max_tool_response_length is exhausted or the context window fills. Each iteration calls your LLM. At $3/million tokens on Claude Sonnet or GPT-4.1, a 40-iteration loop on a 10k-token context is a $1.20 accident waiting to happen — per agent invocation. This page shows how to add a circuit breaker that trips the moment a loop is detected, and a dollar cap that fires before you see the bill.
How Phidata / Agno agents loop
Agno’s Agent class runs a standard Reason + Act loop: the LLM generates text, Agno parses tool calls, executes them, appends results to the message history, and feeds the updated history back to the LLM. The loop ends when the LLM generates a response with no tool calls — which it interprets as “I have enough information to answer.”
The loop continues indefinitely in two scenarios:
- Tool returns unhelpful results. A search tool that returns empty arrays, a database tool that returns “no rows found,” or an API tool that returns a rate-limit error message (as a string, not an exception) gives the LLM no signal to stop reasoning. The LLM generates a new tool call with different arguments hoping for a better result. If none comes, it loops. Agno’s
max_tool_response_lengthtruncates the response content but does not terminate the loop — the truncated content just looks like another partial result. - Multi-agent Team loops. Agno’s
Teamprimitive lets a coordinator agent route subtasks to member agents. If a member agent fails and returns a fallback message, the coordinator may route the same task to another member, which also fails, creating a cross-agent loop that multiplies LLM calls by the number of team members tried.
What Agno provides natively and what it doesn’t
Agno includes a handful of useful defaults:
max_retrieson theAgent— retries the model call on transient errors like timeouts. This is not a loop detector; it makes the agent more resilient to flaky API calls, not to flaky tool results.max_stepsparameter — caps the total number of Reason+Act iterations. This prevents runaway looping in theory. In practice, the default is permissive enough that even a 20-step loop generates substantial cost, and there is no distinction between “making useful progress at step 18” and “repeating the same tool call for the 18th time.”- Built-in observability hooks — Agno can log each tool call. Logs are useful for post-incident analysis; they do not interrupt the loop.
None of these are circuit breakers. A circuit breaker must detect a pattern (repeated identical tool-call fingerprints) and interrupt the agent before the next LLM call is made. Agno’s built-ins detect counts, not patterns. RunGuard adds the missing layer.
Adding RunGuard to an Agno agent
RunGuard integrates with Agno at the tool level. Wrap your tool functions with guard() before passing them to the Agent. The guard tracks call signatures across the agent’s run and raises LoopDetectedError on the second identical pattern.
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from runguard import guard, BudgetTracker, LoopDetectedError, BudgetExceededError
# Your existing tool functions
def search_knowledge_base(query: str) -> str:
return kb.search(query)
def fetch_document(doc_id: str) -> str:
return docs.fetch(doc_id)
# Wrap tools with RunGuard — loop_threshold=2 trips on first repeat
guarded_search = guard(
search_knowledge_base,
loop_window=8, # watch last 8 tool calls
loop_threshold=2, # trip on 2nd identical (args, result) pair
)
guarded_fetch = guard(fetch_document, loop_window=8, loop_threshold=2)
# Budget tracker — shared across all tools in this run
budget = BudgetTracker(max_usd=2.00)
agent = Agent(
model=OpenAIChat(id="gpt-4.1"),
tools=[guarded_search, guarded_fetch],
max_steps=20,
)
try:
agent.print_response("Summarise all documents about Q2 planning.")
except LoopDetectedError as e:
print(f"Loop detected: {e.tool_name} repeated {e.count}x — aborting")
except BudgetExceededError as e:
print(f"Budget cap reached: ${e.accumulated_usd:.2f} of ${e.max_usd:.2f}")
Circuit breaker for Agno Team multi-agent pipelines
For Team pipelines, the loop risk is different: the coordinator agent may route the same task to multiple failing members, amplifying cost. RunGuard’s team guard wraps the entire coordination call and detects cross-agent repetition.
from agno.team import Team
from runguard import guard, TeamLoopGuard
# Guard individual member agents' tools as above
research_agent = Agent(
name="researcher",
tools=[guarded_search],
model=OpenAIChat(id="gpt-4.1-mini"),
)
writer_agent = Agent(
name="writer",
model=OpenAIChat(id="gpt-4.1-mini"),
)
# TeamLoopGuard watches coordinator routing decisions
# and trips if the same subtask is delegated >2 times
team = Team(
name="research_team",
mode="coordinate",
members=[research_agent, writer_agent],
team_loop_guard=TeamLoopGuard(
max_rerouts=2, # trip after 2 rerouts of the same subtask
budget_usd=5.00, # total budget across all member agents
),
)
result = team.run("Write a competitive analysis for our Q3 product launch.")
Agno / Phidata built-ins vs RunGuard
| Scenario | Agno built-in | RunGuard |
|---|---|---|
| Single agent tool loop | max_steps cap (count only) | LoopDetectedError on 2nd identical fingerprint |
| Team coordinator re-routing loop | No detection | TeamLoopGuard trips at max_rerouts |
| Per-run dollar cap | No | BudgetTracker raises BudgetExceededError |
| Tool error masking (error returned as string) | No detection | Loop detector catches repeated error-string pattern |
| Observability / trace logs | Built-in logging hooks | Structured error with tool name, args, repeat count |
| Context window fill | Truncates response content | ContextOverflowError at configurable threshold |
The alert pattern: Slack notification on trip
A circuit breaker that raises silently is better than none, but the real value comes from routing the alert to wherever your on-call team watches. RunGuard fires a Slack webhook whenever a breaker trips, with the tool name, the repeated fingerprint, and the run cost at time of trip.
from runguard import guard, SlackAlerter
alerter = SlackAlerter(webhook_url="https://hooks.slack.com/services/…")
guarded_search = guard(
search_knowledge_base,
loop_window=8,
loop_threshold=2,
on_trip=alerter.notify, # called synchronously before raising
)
# Slack message format sent on trip:
# [RunGuard] Loop detected in Agno agent
# Tool: search_knowledge_base
# Repeated call: {"query": "Q2 planning documents"} → "" (2nd time)
# Cost at trip: $0.47
# Agent run ID: agt_20260601_abc123
Migrating from Phidata 2.x to Agno 1.x: what changes for RunGuard
If you’re still running Phidata 2.x imports (from phi.agent import Agent) and planning to migrate to Agno (from agno.agent import Agent), the RunGuard integration is identical in both versions. The guard() wrapper operates at the Python function level, not at the agent framework level, so there is no import swap needed on the RunGuard side. The key migration checkpoint is that Agno renamed phi.tools.Toolkit to agno.tools.Toolkit — any toolkit class you wrap with guard stays guarded after the rename.
Add a circuit breaker to your Agno agent today
RunGuard’s Python SDK installs with pip install runguard. Wrap your tool functions with guard(), attach a BudgetTracker, and catch LoopDetectedError and BudgetExceededError in your agent’s run loop. No changes to your Agent definition, no new framework overhead.
Get started with RunGuard — or see the same pattern for PydanticAI, Haystack, and AutoGen.