Retries, Fallback Models and Circuit Breaker for a Python AI Agent in 10xGraph

"""Keep an agent answering when its LLM provider fails.

Primary model is Anthropic. RetryConfig retries transient errors with exponential
backoff and adds a circuit breaker. fallback_models switches to OpenAI, then Google.
Run: python examples/resilience/retry_and_fallback.py
"""

import asyncio

from dotenv import load_dotenv

from tenxgraph.core.exceptions import GraphError
from tenxgraph.core.graph import RetryConfig
from tenxgraph.core.state import Message
from tenxgraph.prebuilt.agent import ReactAgent
from tenxgraph.storage.checkpointer import InMemoryCheckpointer


load_dotenv()


def get_ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return f"Ticket {ticket_id} is open and assigned to the billing team."


retry_config = RetryConfig(
    max_retries=2,  # up to 3 calls per model: 1 try + 2 retries
    initial_delay=1.0,  # wait 1s, then 2s (delay * backoff_factor ** n)
    backoff_factor=2.0,
    max_delay=20.0,  # cap on any single wait
    retryable_status_codes=frozenset({429, 500, 502, 503, 529}),  # the default set
    # Circuit breaker, tracked per (provider, model) on this agent. After 3 failed
    # attempt cycles in a row the model is skipped for 60s, so a dead provider is not
    # retried on every request. After 60s one trial call is allowed.
    circuit_breaker_enabled=True,
    circuit_breaker_threshold=3,
    circuit_breaker_reset_timeout=60.0,
)


async def main() -> None:
    agent = ReactAgent(
        model="claude-sonnet-5-5",
        system_prompt=[{"role": "system", "content": "You are a support assistant."}],
        tools=[get_ticket_status],
        retry_config=retry_config,
        # A bare string inherits the PRIMARY model's provider, so "gpt-5-mini" alone would be
        # sent to Anthropic. To cross companies, give a (model, provider) tuple.
        fallback_models=[
            ("gpt-5-mini", "openai"),
            ("gemini-2.5-flash", "google"),
        ],
    )

    graph = agent.compile(checkpointer=InMemoryCheckpointer())
    config = {"thread_id": "ticket-7", "user_id": "demo-user"}
    try:
        result = await graph.ainvoke(
            {"messages": [Message.text_message("What is the status of ticket T-884?")]},
            config,
        )
        print("assistant:", result["messages"][-1].text())
    except GraphError as exc:
        # Library errors carry a code and context you can log or return from an API.
        print("graph error:", exc.to_dict())
    except Exception as exc:
        # When the primary and every fallback fail, the LAST provider's exception is
        # re-raised unchanged (an SDK error such as openai.APIStatusError or a
        # google.genai error). Handle it here, or let the caller retry later.
        print("all models failed:", type(exc).__name__, exc)
    finally:
        await graph.aclose()


if __name__ == "__main__":
    asyncio.run(main())

"""Stop a runaway graph with recursion_limit and catch GraphRecursionError.

No LLM is called: the graph is two plain nodes that route back and forth forever,
which is what a confused agent loop looks like. Also shows the library exception
hierarchy. Run: python examples/resilience/recursion_limit.py
"""

import asyncio

from tenxgraph.core.exceptions import GraphError, GraphRecursionError
from tenxgraph.core.graph import StateGraph
from tenxgraph.core.state import AgentState, Message
from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from tenxgraph.utils import END


class LoopState(AgentState):
    attempts: int = 0


def think(state: LoopState) -> LoopState:
    state.attempts += 1
    return state


def act(state: LoopState) -> Message:
    return Message.text_message(f"attempt {state.attempts}: not done yet", role="assistant")


def route(state: LoopState) -> str:
    # A real agent would stop when the task is done. This one never is.
    return "THINK"


def build_graph():
    graph = StateGraph(LoopState())
    graph.add_node("THINK", think)
    graph.add_node("ACT", act)
    graph.add_edge("THINK", "ACT")
    graph.add_conditional_edges("ACT", route, {"THINK": "THINK", END: END})
    graph.set_entry_point("THINK")
    return graph.compile(checkpointer=InMemoryCheckpointer())


async def main() -> None:
    app = build_graph()
    inputs = {"messages": [Message.text_message("solve it")]}

    # recursion_limit counts node steps (default 25). Lower it for cheap, fast failure.
    config = {"thread_id": "loop-1", "recursion_limit": 6}
    try:
        await app.ainvoke(inputs, config)
    except GraphRecursionError as exc:
        print("stopped by recursion limit")
        print("  code:   ", exc.error_code)  # RECURSION_001
        print("  context:", exc.context)  # max_steps, current_step, current_node
        print("  dict:   ", exc.to_dict())  # structured form, safe to return from an API

    # Every library error derives from GraphError, so one handler can catch them all.
    try:
        await app.ainvoke(inputs, {"thread_id": "loop-2", "recursion_limit": 4})
    except GraphError as exc:
        print("caught as GraphError:", type(exc).__name__, "-", exc.message)

    # The checkpointer still holds the work done before the stop.
    saved = await app._checkpointer.aget_state({"thread_id": "loop-1", "user_id": "anonymous"})
    print("attempts recorded before the stop:", saved.attempts if saved else None)

    await app.aclose()


if __name__ == "__main__":
    asyncio.run(main())