Retry If: API Integration Agent That Retries on Transient Failures

Why this matters

Transient failures are a fact of life in production systems – API rate limits, network timeouts, temporary service outages. loop_while() retries an agent when a condition indicates a transient failure, with bounded iteration to prevent infinite retry storms. This is the agent equivalent of an HTTP retry policy, built directly into the pipeline topology.

Warning

Without this Without conditional retry, a transient API timeout crashes the entire pipeline. You end up wrapping every external call in try/except with manual retry loops, scattering resilience logic across the codebase. loop_while() keeps retry logic declarative and visible in the pipeline expression.

How to retry agent execution on transient failures with loop_while().

Source: 38_retry_if.py

# Native ADK has no built-in conditional retry. You'd need to:
#   1. Wrap the agent in a LoopAgent
#   2. Create a custom checkpoint BaseAgent that evaluates a predicate
#   3. Yield Event(actions=EventActions(escalate=True)) to exit when satisfied
# For a payment gateway integration, this means 30+ lines of boilerplate
# just to handle transient 503 errors.

from adk_fluent import Agent, Loop

# Scenario: A payment processing agent that calls an external gateway.
# Transient failures (timeouts, rate limits) should trigger automatic retries,
# but permanent failures (invalid card) should stop immediately.

# .loop_while(): keep retrying while the predicate returns True
payment_processor = (
    Agent("payment_processor")
    .model("gemini-2.5-flash")
    .instruct(
        "Process the payment through the gateway. Report status as 'success', 'transient_error', or 'permanent_error'."
    )
    .writes("payment_status")
    .loop_while(lambda s: s.get("payment_status") == "transient_error", max_iterations=3)
)

# loop_while on a pipeline -- retry the entire charge-then-verify flow
charge_and_verify = (
    Agent("charge_agent")
    .model("gemini-2.5-flash")
    .instruct("Submit charge to payment gateway.")
    .writes("charge_result")
    >> Agent("verification_agent")
    .model("gemini-2.5-flash")
    .instruct("Verify the charge was recorded by the bank.")
    .writes("verified")
).loop_while(lambda s: s.get("verified") != "confirmed", max_iterations=5)

# Equivalence: loop_while(p) == loop_until(not p)
# These produce identical behavior for an inventory sync agent:
via_retry = (
    Agent("inventory_sync")
    .model("gemini-2.5-flash")
    .loop_while(lambda s: s.get("sync_status") != "complete", max_iterations=4)
)
via_loop = (
    Agent("inventory_sync")
    .model("gemini-2.5-flash")
    .loop_until(lambda s: s.get("sync_status") == "complete", max_iterations=4)
)

Equivalence

from adk_fluent.workflow import Loop as LoopBuilder

# loop_while creates a Loop builder
assert isinstance(payment_processor, LoopBuilder)

# max_retries maps to max_iterations
assert payment_processor._config["max_iterations"] == 3

# Pipeline retry also creates a Loop
assert isinstance(charge_and_verify, LoopBuilder)
assert charge_and_verify._config["max_iterations"] == 5

# The predicate is inverted: loop_while(p) stores not-p as until_predicate
until_pred = payment_processor._config["_until_predicate"]
assert until_pred({"payment_status": "success"}) is True  # exit: stop retrying
assert until_pred({"payment_status": "transient_error"}) is False  # continue retrying

# Both loop_while and loop_until produce Loop builders with identical structure
assert isinstance(via_retry, LoopBuilder)
assert isinstance(via_loop, LoopBuilder)
assert via_retry._config["max_iterations"] == via_loop._config["max_iterations"]

# Build verifies checkpoint agent is injected
built = payment_processor.build()
assert len(built.sub_agents) >= 2
assert built.sub_agents[-1].name == "_until_check"