Build a Resilient AI Agent with LangGraph

Build a production-patterned AI agent with circuit breaker, caching, and human-in-the-loop.

Introduction

30 Second Summary

AI agents are powerful when they work. When they break, they burn through API credits, return garbage, or take dangerous actions without anyone watching. Production-grade agents need safety nets built in from day one.

In this project, you will build a multi-tool research assistant agent with production resilience patterns baked in. You will implement a circuit breaker, response caching, and a human-in-the-loop approval gate so your agent handles failures gracefully and stays safe under pressure.

What You'll Build

A research assistant agent powered by Groq (Llama 3.3 70B) that can search for information, perform calculations, and publish reports, orchestrated through a LangGraph StateGraph with conditional routing and tool calling.

By the end of this project, you'll have:

  • A LangGraph StateGraph with conditional routing and a ReAct-style tool-calling loop that decides what to do next based on model output.
  • A circuit breaker state machine (Closed/Open/Half-Open) paired with exponential backoff with jitter that protects against cascading failures and controls retry behavior.
  • A response cache with TTL that eliminates redundant LLM calls, plus a human-in-the-loop interrupt pattern that pauses execution for approval on dangerous actions.
  • Secret Mission: Add a model fallback chain that automatically routes to a backup model when the primary trips the circuit breaker.

Are there any prerequisites?

You should be comfortable with Python (functions, classes, decorators) and have a basic understanding of APIs and HTTP. No prior experience with LangGraph or AI agents is required.

Set Up Your Project

Before building the agent, you need a working Python environment with the right packages and a connection to Groq's LLM API.

In this step, you'll create your project folder, install LangGraph and the Groq integration, obtain a free API key, and verify everything connects.

In this step, get ready to:
  • Create a project directory with a virtual environment and dependencies.
  • Obtain a free Groq API key.
  • Verify the connection to the LLM works.
Create your project directory and install packages
  • Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir resilient-agent
cd resilient-agent
  • Confirm the folder was created by running:
ls ~/Desktop

You should see resilient-agent listed in the output.

  • Create a virtual environment and activate it by running:
python3 -m venv venv
source venv/bin/activate

You should see (venv) appear at the start of your terminal prompt.

What is a virtual environment?

A virtual environment is an isolated Python installation that keeps this project's packages separate from your system Python. This prevents version conflicts between projects.

  • Install the project's dependencies by running:
pip install langgraph==1.2.7 langchain-groq==1.1.3 python-dotenv==1.1.0

You should see output ending with Successfully installed followed by the package names.

Installation failing?

Make sure you see (venv) in your prompt. If not, re-run source venv/bin/activate and try the install again.

If you see a network error, check your internet connection and try once more.

Why is pip failing to install my packages?

  • Open your code editor. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the resilient-agent folder, then click Open.
  • Create a new file called requirements.txt in the resilient-agent folder. In VS Code, click the new file icon in the Explorer sidebar (or press Cmd+N).
  • Paste the following into requirements.txt:
langgraph==1.2.7
langchain-groq==1.1.3
python-dotenv==1.1.0

What does this file do?

A requirements.txt file documents every package (and exact version) your project needs. Anyone who clones your project can run pip install -r requirements.txt to install the same dependencies.

  • Create another new file called .env in the same folder.
  • Paste the following into .env:
GROQ_API_KEY=your-api-key-here

What is a .env file?

A .env file stores environment variables (like API keys) outside your code. The python-dotenv package loads these variables at runtime so your secrets never get committed to version control.

You'll replace your-api-key-here with a real key in the next substep.

  • Save both files (Cmd+S on macOS, Ctrl+S on Windows).
Get a Groq API key

Groq provides free access to powerful LLMs including Llama 3.3 70B. You'll never be charged on the free tier. If you exceed rate limits, Groq returns an error instead of a bill.

  • Open your browser and go to console.groq.com.
  • Click Sign Up to create a free account (or sign in if you already have one).
  • Once logged in, navigate to the API Keys section in the left sidebar, or go directly to console.groq.com/keys.
  • Click Create API Key.
  • Give it a name (e.g. resilient-agent) and click Submit.
  • Copy the key immediately. You will only see it once.

Why is Groq free?

Groq's free tier gives you access to all models with rate limits of 30 requests per minute for llama-3.3-70b-versatile. You're never charged. If you hit the limit, you'll get a 429 (rate limit) error, which is exactly what your retry logic will handle later in this project.

  • Switch back to VS Code and open your .env file.
  • Replace your-api-key-here with the API key you just copied.
  • Save the file (Cmd+S).

Keep your API key private

Never commit your .env file to Git or share your key publicly. If you suspect it's been exposed, revoke it immediately at console.groq.com/keys and generate a new one.

Can't create an API key?

Some users have reported permission issues when signing up with non-Gmail accounts. If you see a message about needing a developer role, try signing up with a standard Gmail address instead.

Help me troubleshoot Groq API key creation

Verify the connection works

With your key saved, you'll write a short script to confirm your environment can reach Groq's API and get a response from the llama-3.3-70b-versatile model.

  • In VS Code, create a new file called test_connection.py in your resilient-agent folder.
  • Paste the following code into test_connection.py:
import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq

# Load environment variables from .env file
load_dotenv()

# Create a ChatGroq instance pointing at the Llama 3.3 70B model
llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)

# Send a simple test prompt and print the response
response = llm.invoke("Say hello in one sentence.")
print(response.content)

What does this code do?

  • load_dotenv() reads your .env file and makes GROQ_API_KEY available via os.getenv().
  • ChatGroq is the LangChain integration for Groq. It handles authentication and formatting for you.
  • llm.invoke() sends a prompt to the model and returns the response.
  • Save test_connection.py.
  • Switch back to your terminal (make sure you still see (venv) in the prompt) and run the script:
python test_connection.py

You should see a short greeting printed to the terminal, something like "Hello, how can I help you today?" The exact wording will vary.

Seeing an authentication error?

Double-check that your .env file is saved in the same resilient-agent folder as test_connection.py. The key in your .env must not have quotes around it.

If you see a module not found error, confirm your virtual environment is active (you should see (venv) in the prompt).

Help me fix my Groq connection test

✔️ Awesome, I've got everything!

Great. Double check you've saved all three files (requirements.txt, .env, and test_connection.py) before moving on.

ⓧ I'd like to double check the full code

Here are all three files you should have in your resilient-agent folder:

GROQ_API_KEY=your-api-key-here
langgraph==1.2.7
langchain-groq==1.1.3
python-dotenv==1.1.0
import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq

load_dotenv()

llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)

response = llm.invoke("Say hello in one sentence.")
print(response.content)

Your environment is ready and confirmed working. Next up, you'll build the agent graph with tool calling so your LLM can take actions in the world.

Build the Agent Graph with Tool Calling

Your connection to Groq is confirmed. Now it's time to build the actual agent.

An AI agent isn't just a model that answers questions. It's a loop: the model decides what to do, calls a tool, reads the result, and decides again. In this step, you'll build that loop using LangGraph's StateGraph, giving your agent three tools and the wiring to use them autonomously.

In this step, get ready to:
  • Create three agent tools (search, calculate, publish).
  • Build the agent graph with conditional routing.
  • Test the agent with a calculation query.
Create the agent tools

Every agent needs tools. Tools are Python functions the model can call to interact with the outside world. You'll create three: one that searches the web (with simulated failures), one that evaluates math, and one that publishes reports.

  • Create a new file called tools.py in your resilient-agent folder.
  • Add the imports and the search_web tool by pasting this code:
import hashlib
from langchain.tools import tool

# Track how many times search has been called
_search_call_count = 0
FAILURE_RATE = 5


@tool
def search_web(query: str) -> str:
    """Search the web for information on a given query."""
    global _search_call_count
    _search_call_count += 1

    # Simulate unreliable external service: fail every 5th call
    if _search_call_count % FAILURE_RATE == 0:
        raise ConnectionError("Search service temporarily unavailable")

    query_hash = hashlib.md5(query.encode()).hexdigest()[:6]
    return (
        f"Search results for '{query}':\n"
        f"1. [Result {query_hash}a] Relevant information about {query}.\n"
        f"2. [Result {query_hash}b] Additional context on {query}.\n"
        f"3. [Result {query_hash}c] Expert analysis of {query}."
    )

What does this code do?

The @tool decorator from LangChain turns a regular function into something an LLM can call. The docstring becomes the tool's description that the model reads to decide when to use it.

The _search_call_count and FAILURE_RATE simulate an unreliable external service. Every 5th call raises a ConnectionError. This gives the circuit breaker (coming in Step 3) something real to protect against.

  • Add the calculate and publish_report tools below the search_web function:
@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression. Only supports basic arithmetic."""
    # Only allow safe characters (digits, operators, parentheses)
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        return f"Error: Invalid characters in expression '{expression}'"

    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"


@tool
def publish_report(title: str, content: str) -> str:
    """Publish a report with the given title and content. This is a sensitive action."""
    return f"Report '{title}' published successfully with content: {content[:100]}..."

What does this code do?

  • calculate validates the input against a whitelist of safe characters before using eval(). This prevents arbitrary code execution.
  • publish_report is deliberately marked as a "sensitive action" in its docstring. In Step 5, you'll add a human-in-the-loop check that intercepts this tool before it runs.
  • Save tools.py.

Seeing an import error?

Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt. If not, run source venv/bin/activate from inside your resilient-agent folder.

help me fix an import error with langchain.tools

✔️ Awesome, I've got everything!

Great. Double check you've saved tools.py.

ⓧ I'd like to double check the full code

import hashlib
from langchain.tools import tool

_search_call_count = 0
FAILURE_RATE = 5


@tool
def search_web(query: str) -> str:
    """Search the web for information on a given query."""
    global _search_call_count
    _search_call_count += 1

    if _search_call_count % FAILURE_RATE == 0:
        raise ConnectionError("Search service temporarily unavailable")

    query_hash = hashlib.md5(query.encode()).hexdigest()[:6]
    return (
        f"Search results for '{query}':\n"
        f"1. [Result {query_hash}a] Relevant information about {query}.\n"
        f"2. [Result {query_hash}b] Additional context on {query}.\n"
        f"3. [Result {query_hash}c] Expert analysis of {query}."
    )


@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression. Only supports basic arithmetic."""
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        return f"Error: Invalid characters in expression '{expression}'"

    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"


@tool
def publish_report(title: str, content: str) -> str:
    """Publish a report with the given title and content. This is a sensitive action."""
    return f"Report '{title}' published successfully with content: {content[:100]}..."
Build the agent graph

A StateGraph is how LangGraph organizes your agent. You define nodes (functions that do work) and edges (rules for what happens next). The model calls a tool, the tool returns a result, and the model decides whether to call another tool or give a final answer.

  • Create a new file called agent.py in your resilient-agent folder.
  • Add the imports and module-level configuration by pasting this code:
import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq
from langchain.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver

from tools import search_web, calculate, publish_report

load_dotenv()

# System prompt tells the model what tools it has and how to behave
SYSTEM_PROMPT = """You are a research assistant that can search the web, perform calculations, and publish reports.
Use the appropriate tool for each task. Be concise in your responses."""

# Register all tools and create the prebuilt ToolNode executor
tools = [search_web, calculate, publish_report]
tool_node = ToolNode(tools)

# Initialize the LLM and bind tools so it knows how to call them
llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)
llm_with_tools = llm.bind_tools(tools)

What does this code do?

  • ToolNode is a prebuilt LangGraph component. When the model returns a message with tool_calls, ToolNode automatically executes the matching functions and passes the results back.
  • .bind_tools(tools) attaches the tool schemas to the LLM so it knows what tools exist, what arguments they take, and when to call them.
  • MessagesState is LangGraph's built-in state schema. It holds the conversation as a list of messages that grows with each node execution.
  • Add the call_model node function below the configuration code:
def call_model(state: MessagesState):
    """Node that sends messages to the LLM and returns its response."""
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

What does this code do?

This function is the "brain" of your agent. It prepends the system prompt to the conversation, sends everything to the LLM, and returns the response. If the model decides it needs a tool, the response will contain tool_calls. If not, it's a final answer.

  • Add the routing function that decides where to go after the model responds:
def check_dangerous_tools(state: MessagesState):
    """Route based on whether the model wants to call a tool."""
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return "end"
    return "tools"

What does this code do?

This is the agent's decision point. After the model responds, this function checks: did the model ask to call a tool? If yes, route to the tools node. If no, route to END (the conversation is done). You'll expand this function in Step 5 to intercept dangerous tool calls.

  • Wire the nodes into a graph and compile it by adding this code:
# Build the state graph
workflow = StateGraph(MessagesState)

# Add nodes (the functions that do work)
workflow.add_node("call_model", call_model)
workflow.add_node("tools", tool_node)

# Wire the edges (rules for what happens next)
workflow.add_edge(START, "call_model")
workflow.add_conditional_edges(
    "call_model",
    check_dangerous_tools,
    {"tools": "tools", "end": END},
)
workflow.add_edge("tools", "call_model")

# Compile with a checkpointer for state persistence
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

What does this code do?

This is where the agent loop takes shape. The graph starts at call_model. If the model returns tool calls, it routes to tools (which executes them). After tools run, the edge goes back to call_model so the model can process the results. This loop continues until the model gives a final answer with no tool calls.

The InMemorySaver checkpointer lets LangGraph save the graph's state between invocations. You'll use this for human-in-the-loop in Step 5.

  • Save agent.py. You won't run it yet because it still needs an entry point.
Add the entry point and test the agent

The graph is wired. Now you need a function that sends a query into the graph and prints the result.

  • Add the run_agent function and __main__ block at the bottom of agent.py:
def run_agent(query: str, thread_id: str = "default"):
    """Send a query to the agent and print the response."""
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print(f"{'='*60}")

    config = {"configurable": {"thread_id": thread_id}}
    input_messages = {"messages": [HumanMessage(content=query)]}

    result = graph.invoke(input_messages, config=config)

    # Find the last AI message with content
    last_ai = None
    for msg in reversed(result["messages"]):
        if isinstance(msg, AIMessage) and msg.content:
            last_ai = msg
            break

    if last_ai:
        print(f"\nAgent: {last_ai.content}")

    return result


if __name__ == "__main__":
    run_agent("What is 25 * 17?")

What does this code do?

  • run_agent wraps the graph invocation with a formatted query and response printout.
  • thread_id identifies the conversation thread for the checkpointer. Each unique thread ID tracks a separate conversation history.
  • The __main__ block runs a test query asking "What is 25 * 17?" which should trigger the calculate tool.
  • Save agent.py.
  • Run the agent by executing this command:
python agent.py

You should see output showing the query, and the agent's response containing the answer 425.

Not seeing the correct answer?

  • If you see a ModuleNotFoundError, make sure your virtual environment is active and tools.py is saved in the same folder as agent.py.
  • If you see a rate limit error, wait 60 seconds and try again. Groq's free tier has a 30 RPM limit on llama-3.3-70b-versatile.
  • If the agent responds without using the calculator (just gives a text answer), check that .bind_tools(tools) is called and you're using llm_with_tools in call_model, not the plain llm.

help me debug my LangGraph agent not calling tools

✔️ Awesome, I've got everything!

Great. Double check you've saved both tools.py and agent.py.

ⓧ I'd like to double check the full code

import hashlib
from langchain.tools import tool

_search_call_count = 0
FAILURE_RATE = 5


@tool
def search_web(query: str) -> str:
    """Search the web for information on a given query."""
    global _search_call_count
    _search_call_count += 1

    if _search_call_count % FAILURE_RATE == 0:
        raise ConnectionError("Search service temporarily unavailable")

    query_hash = hashlib.md5(query.encode()).hexdigest()[:6]
    return (
        f"Search results for '{query}':\n"
        f"1. [Result {query_hash}a] Relevant information about {query}.\n"
        f"2. [Result {query_hash}b] Additional context on {query}.\n"
        f"3. [Result {query_hash}c] Expert analysis of {query}."
    )


@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression. Only supports basic arithmetic."""
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        return f"Error: Invalid characters in expression '{expression}'"

    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"


@tool
def publish_report(title: str, content: str) -> str:
    """Publish a report with the given title and content. This is a sensitive action."""
    return f"Report '{title}' published successfully with content: {content[:100]}..."
import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq
from langchain.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver

from tools import search_web, calculate, publish_report

load_dotenv()

SYSTEM_PROMPT = """You are a research assistant that can search the web, perform calculations, and publish reports.
Use the appropriate tool for each task. Be concise in your responses."""

tools = [search_web, calculate, publish_report]
tool_node = ToolNode(tools)

llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)
llm_with_tools = llm.bind_tools(tools)


def call_model(state: MessagesState):
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}


def check_dangerous_tools(state: MessagesState):
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return "end"
    return "tools"


workflow = StateGraph(MessagesState)

workflow.add_node("call_model", call_model)
workflow.add_node("tools", tool_node)

workflow.add_edge(START, "call_model")
workflow.add_conditional_edges(
    "call_model",
    check_dangerous_tools,
    {"tools": "tools", "end": END},
)
workflow.add_edge("tools", "call_model")

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)


def run_agent(query: str, thread_id: str = "default"):
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print(f"{'='*60}")

    config = {"configurable": {"thread_id": thread_id}}
    input_messages = {"messages": [HumanMessage(content=query)]}

    result = graph.invoke(input_messages, config=config)

    last_ai = None
    for msg in reversed(result["messages"]):
        if isinstance(msg, AIMessage) and msg.content:
            last_ai = msg
            break

    if last_ai:
        print(f"\nAgent: {last_ai.content}")

    return result


if __name__ == "__main__":
    run_agent("What is 25 * 17?")

Why conditional routing instead of a fixed path?

A fixed graph (model then tools then done) would only allow one tool call per query. Conditional routing creates a loop: the model can call a tool, read the result, call another tool, and keep going until it has enough information to answer. This is the ReAct pattern that powers most production agents.

Your agent is alive. It takes a question, picks the right tool, uses it, and gives you a final answer. Next up, you'll make it production-grade by adding a circuit breaker that protects against tool failures and retry logic that handles rate limits gracefully.

Add Circuit Breaker and Retry Logic

Your agent can call tools and route decisions, but it has zero protection against failures. In production, API providers go down, rate limits get hit, and network connections drop.

Without resilience patterns, a single failure cascades into retries that hammer an already-struggling service. In this step, you will build a circuit breaker state machine and an exponential backoff retry decorator that protect your agent from exactly this scenario.

In this step, get ready to:
  • Build a CircuitBreaker class with three states (Closed, Open, Half-Open) that stops calling a failing service.
  • Add a retry decorator with exponential backoff and jitter for transient errors.
  • Wire the circuit breaker and retry logic into your agent's LLM call path.
Create the CircuitBreaker class

A circuit breaker works like a physical electrical circuit breaker. When too many failures happen in a row, it "trips open" and immediately rejects new calls instead of wasting time on a dead service.

  • Create a new file called resilience.py in your resilient-agent folder.
  • Add the following imports and the CircuitBreaker class:
import time
import random
import hashlib
import json
from functools import wraps


class CircuitBreaker:
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

    def __init__(self, failure_threshold=3, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = self.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.success_count = 0

    def call(self, func, *args, **kwargs):
        if self.state == self.OPEN:
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = self.HALF_OPEN
                print(f"  [Circuit] Entering HALF_OPEN - testing recovery...")
            else:
                raise CircuitBreakerOpenError(
                    f"Circuit is OPEN. Retry after {self.recovery_timeout}s."
                )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e

    def _on_success(self):
        if self.state == self.HALF_OPEN:
            print(f"  [Circuit] Recovery successful - closing circuit.")
            self.state = self.CLOSED
        self.failure_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = self.OPEN
            print(
                f"  [Circuit] OPENED after {self.failure_count} failures. "
                f"Will retry in {self.recovery_timeout}s."
            )


class CircuitBreakerOpenError(Exception):
    pass

What does this code do?

The CircuitBreaker class implements a three-state machine that protects against cascading failures:

  • CLOSED (normal): calls pass through. Each failure increments failure_count. Once it hits failure_threshold, the circuit trips open.
  • OPEN (failing fast): every call immediately raises CircuitBreakerOpenError without touching the service. No wasted time or network calls.
  • HALF_OPEN (testing recovery): after recovery_timeout seconds elapse, the next call is allowed through as a probe. If it succeeds, the circuit closes. If it fails, the circuit reopens.
Add the retry_with_backoff decorator

Retrying a failed call immediately is pointless if the service is rate-limiting you. Exponential backoff waits longer between each retry, and adding jitter (randomness) prevents a stampede of retries from multiple clients hitting at the exact same moment.

  • In resilience.py, add the retry_with_backoff decorator below the CircuitBreakerOpenError class:
def retry_with_backoff(max_retries=3, base_delay=1.0, cap=16.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    error_str = str(e).lower()
                    is_retryable = "429" in error_str or "rate" in error_str or "503" in error_str
                    if not is_retryable or attempt == max_retries:
                        raise
                    delay = random.uniform(0, min(cap, base_delay * (2 ** attempt)))
                    print(f"  [Retry] Attempt {attempt + 1} failed. Retrying in {delay:.2f}s...")
                    time.sleep(delay)
            return func(*args, **kwargs)
        return wrapper
    return decorator

What does this code do?

  • The decorator only retries on retryable errors: HTTP 429 (rate limit), rate-related messages, and 503 (service unavailable). Auth errors or bad requests fail immediately.
  • The delay formula random.uniform(0, min(cap, base_delay * 2^attempt)) implements full jitter. Each retry waits a random amount up to an exponentially growing ceiling, preventing synchronized retry storms.
  • The cap parameter ensures the maximum wait never exceeds a fixed bound, no matter how many retries occur.

Why exponential backoff with jitter instead of fixed delays?

Fixed-interval retries cause the "thundering herd" problem. If 100 clients all get rate-limited at the same second, they all retry at the same second, creating another spike. Exponential backoff spreads retries over increasing windows. Jitter randomizes within each window so no two clients retry at the same instant.

  • Save resilience.py.

Seeing a syntax error?

  • Check that retry_with_backoff is indented at the module level (no extra indentation). It is a standalone function, not a method inside the CircuitBreaker class.
  • Make sure the @wraps(func) line uses wraps from the functools import at the top of the file.
  • Verify you have three levels of nested def statements: retry_with_backoff returns decorator which returns wrapper.

Help me debug a syntax error in my retry_with_backoff decorator in resilience.py.

Integrate the circuit breaker into agent.py

With the resilience module ready, you need to wire it into your agent's LLM call path. The circuit breaker wraps the function that calls Groq, and the retry decorator handles transient failures before they count against the circuit.

  • Switch back to agent.py and find the import line near the top that says:
from resilience import CircuitBreaker, CircuitBreakerOpenError, retry_with_backoff, ResponseCache

This import brings in both the classes you just built. Now find the circuit breaker instance and the decorated LLM function below the tool setup:

circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
cache = ResponseCache()


@retry_with_backoff(max_retries=2, base_delay=1.0, cap=8.0)
def _invoke_llm(messages):
    return llm_with_tools.invoke(messages)

How do the retry and circuit breaker work together?

  • The @retry_with_backoff decorator wraps _invoke_llm. If Groq returns a 429 or 503, the decorator retries up to 2 times with jitter before giving up.
  • The circuit_breaker.call(_invoke_llm, messages) in the call_model node wraps the already-retrying function. If the retries are exhausted and the call still fails, that failure counts against the circuit breaker.
  • After 3 consecutive failures (failure_threshold=3), the circuit opens and immediately returns a "service unavailable" message for the next 30 seconds without calling Groq at all.

Now look at the call_model function to see how errors are caught:

    try:
        response = circuit_breaker.call(_invoke_llm, messages)
        cache.set(state["messages"], response)
        return {"messages": [response]}
    except CircuitBreakerOpenError as e:
        error_msg = AIMessage(content=f"Service temporarily unavailable: {e}")
        return {"messages": [error_msg]}
    except Exception as e:
        error_msg = AIMessage(content=f"Error calling model: {e}")
        return {"messages": [error_msg]}

What does this error handling do?

  • When the circuit is open, CircuitBreakerOpenError is caught specifically. The agent returns a graceful "service unavailable" message instead of crashing.
  • Any other exception (network timeout, unexpected error) is caught by the generic except Exception handler. The agent stays alive either way.

Why build the circuit breaker from scratch?

Libraries like pybreaker exist, but the interview value is in understanding every line. When someone asks "how does your agent handle provider outages?" you can walk through the three states, explain why you chose a 30-second recovery timeout, and describe how half-open probing avoids premature recovery.

Your resilience module is ready and wired in. You will test the full flow once the response cache is added in the next step, since agent.py depends on both the circuit breaker and the cache working together.

✔️ Awesome, I've got everything!

Great. Make sure you have saved resilience.py.

ⓧ I'd like to double check the full code

Here is the full resilience.py file up to this point:

import time
import random
import hashlib
import json
from functools import wraps


class CircuitBreaker:
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

    def __init__(self, failure_threshold=3, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = self.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.success_count = 0

    def call(self, func, *args, **kwargs):
        if self.state == self.OPEN:
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = self.HALF_OPEN
                print(f"  [Circuit] Entering HALF_OPEN - testing recovery...")
            else:
                raise CircuitBreakerOpenError(
                    f"Circuit is OPEN. Retry after {self.recovery_timeout}s."
                )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e

    def _on_success(self):
        if self.state == self.HALF_OPEN:
            print(f"  [Circuit] Recovery successful - closing circuit.")
            self.state = self.CLOSED
        self.failure_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = self.OPEN
            print(
                f"  [Circuit] OPENED after {self.failure_count} failures. "
                f"Will retry in {self.recovery_timeout}s."
            )


class CircuitBreakerOpenError(Exception):
    pass


def retry_with_backoff(max_retries=3, base_delay=1.0, cap=16.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    error_str = str(e).lower()
                    is_retryable = "429" in error_str or "rate" in error_str or "503" in error_str
                    if not is_retryable or attempt == max_retries:
                        raise
                    delay = random.uniform(0, min(cap, base_delay * (2 ** attempt)))
                    print(f"  [Retry] Attempt {attempt + 1} failed. Retrying in {delay:.2f}s...")
                    time.sleep(delay)
            return func(*args, **kwargs)
        return wrapper
    return decorator

Your agent now has two layers of failure protection: retries with backoff for transient hiccups, and a circuit breaker that stops calling a dead service entirely. Next up, you will add a response cache that eliminates redundant LLM calls and saves you from hitting those rate limits in the first place.

Add Response Caching

Every time your agent processes the same question, it makes a fresh LLM call. That costs tokens, adds latency, and counts against your rate limit.

In this step, you will build a response cache that stores previous LLM responses with a time-to-live (TTL). When the same query arrives again, the cache returns the stored response instantly, skipping the LLM call entirely.

In this step, get ready to:
  • Build a ResponseCache class with TTL-based expiry in resilience.py.
  • Walk through the cache integration in agent.py.
  • Test the cache by running the same query twice and observing the hit.
Add the ResponseCache class to resilience.py

The cache needs to do three things: generate a stable key from the conversation messages, store responses with an expiry timestamp, and track hit/miss counts for observability.

  • In resilience.py, scroll to the bottom of the file (after the retry_with_backoff function) and add the following class:
class ResponseCache:
    def __init__(self):
        self._cache = {}
        self.hits = 0
        self.misses = 0

    def _make_key(self, messages) -> str:
        content = json.dumps(
            [{"role": getattr(m, "type", "unknown"), "content": getattr(m, "content", str(m))}
             for m in messages if getattr(m, "type", "") != "system"],
            sort_keys=True,
        )
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, messages):
        key = self._make_key(messages)
        entry = self._cache.get(key)
        if entry is None:
            self.misses += 1
            return None
        if time.time() > entry["expires"]:
            del self._cache[key]
            self.misses += 1
            return None
        self.hits += 1
        print(f"  [Cache] HIT - returning cached response")
        return entry["value"]

    def set(self, messages, value, ttl=300):
        key = self._make_key(messages)
        self._cache[key] = {"value": value, "expires": time.time() + ttl}

    def stats(self) -> str:
        total = self.hits + self.misses
        rate = (self.hits / total * 100) if total > 0 else 0
        return f"Cache stats: {self.hits} hits, {self.misses} misses ({rate:.0f}% hit rate)"

What does this code do?

The ResponseCache stores LLM responses in a Python dictionary and checks whether they are still fresh before returning them.

  • _make_key() serializes the message list (excluding system messages) into JSON and hashes it with SHA-256. This gives a stable, collision-resistant cache key regardless of message object identity.
  • get() looks up the key. If the entry exists and has not expired, it returns the cached value and increments hits. Otherwise it increments misses and returns None.
  • set() stores a response with a TTL of 300 seconds (5 minutes) by default.
  • stats() returns a human-readable summary of cache performance for debugging.
  • Save resilience.py.

Seeing a NameError for json or hashlib?

The ResponseCache class uses json, hashlib, and time from the standard library. These imports should already be at the top of resilience.py from the previous step. Confirm all three appear in your import lines.

help me debug a NameError in my ResponseCache class

Review the cache integration in agent.py

Your agent.py already imports and uses ResponseCache from the previous step. Now that the class exists, the integration is live.

  • Switch back to agent.py and find the import line near the top:
from resilience import CircuitBreaker, CircuitBreakerOpenError, retry_with_backoff, ResponseCache
  • Find the call_model function. The cache logic sits here:
def call_model(state: MessagesState):
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]

    cached = cache.get(state["messages"])
    if cached is not None:
        return {"messages": [cached]}

    try:
        response = circuit_breaker.call(_invoke_llm, messages)
        cache.set(state["messages"], response)
        return {"messages": [response]}
    except CircuitBreakerOpenError as e:
        error_msg = AIMessage(content=f"Service temporarily unavailable: {e}")
        return {"messages": [error_msg]}
    except Exception as e:
        error_msg = AIMessage(content=f"Error calling model: {e}")
        return {"messages": [error_msg]}

How does the cache fit into the agent loop?

Every time call_model runs, it first asks the cache whether it has already answered this exact set of messages. If the cache returns a hit, the function returns immediately without touching the LLM or the circuit breaker.

On a miss, the function calls the LLM as usual and stores the successful response in the cache for next time. This means the very first run pays the full latency cost, but every repeat within the TTL window is essentially free.

Test the cache with a repeated query

The __main__ block in agent.py already sends the same calculation query twice. This is the perfect test for cache behaviour.

  • Run the agent by executing this command in your terminal:
python agent.py
  • Watch the terminal output. On the first "What is 25 * 17?" query, you will see a cache miss and the LLM responds with 425.
  • On the third invocation (same query repeated), you should see [Cache] HIT - returning cached response printed, and the response arrives noticeably faster because no LLM call was made.
  • At the end of each run, the cache stats line shows cumulative hits and misses.

What should I see?

Your output will include three agent runs. The first and second runs show 0 hits because those are unique queries. The third run (repeating the calculation) shows 1 hits and prints the [Cache] HIT message before returning the cached answer.

Not seeing a cache hit on the repeated query?

Make sure both calls use the exact same query string. If one has a trailing space or different casing, the SHA-256 hash will differ and the cache treats them as separate entries.

Also confirm you saved resilience.py after adding the ResponseCache class. If the class is missing, Python will throw an ImportError on startup.

help me debug why my response cache is not returning a hit

✔️ Awesome, I've got everything!

Great. Double check you have saved resilience.py with the ResponseCache class at the bottom.

ⓧ I'd like to double check the full code

import time
import random
import hashlib
import json
from functools import wraps


class CircuitBreaker:
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

    def __init__(self, failure_threshold=3, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = self.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.success_count = 0

    def call(self, func, *args, **kwargs):
        if self.state == self.OPEN:
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = self.HALF_OPEN
                print(f"  [Circuit] Entering HALF_OPEN - testing recovery...")
            else:
                raise CircuitBreakerOpenError(
                    f"Circuit is OPEN. Retry after {self.recovery_timeout}s."
                )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e

    def _on_success(self):
        if self.state == self.HALF_OPEN:
            print(f"  [Circuit] Recovery successful - closing circuit.")
            self.state = self.CLOSED
        self.failure_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = self.OPEN
            print(
                f"  [Circuit] OPENED after {self.failure_count} failures. "
                f"Will retry in {self.recovery_timeout}s."
            )


class CircuitBreakerOpenError(Exception):
    pass


def retry_with_backoff(max_retries=3, base_delay=1.0, cap=16.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    error_str = str(e).lower()
                    is_retryable = "429" in error_str or "rate" in error_str or "503" in error_str
                    if not is_retryable or attempt == max_retries:
                        raise
                    delay = random.uniform(0, min(cap, base_delay * (2 ** attempt)))
                    print(f"  [Retry] Attempt {attempt + 1} failed. Retrying in {delay:.2f}s...")
                    time.sleep(delay)
            return func(*args, **kwargs)
        return wrapper
    return decorator


class ResponseCache:
    def __init__(self):
        self._cache = {}
        self.hits = 0
        self.misses = 0

    def _make_key(self, messages) -> str:
        content = json.dumps(
            [{"role": getattr(m, "type", "unknown"), "content": getattr(m, "content", str(m))}
             for m in messages if getattr(m, "type", "") != "system"],
            sort_keys=True,
        )
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, messages):
        key = self._make_key(messages)
        entry = self._cache.get(key)
        if entry is None:
            self.misses += 1
            return None
        if time.time() > entry["expires"]:
            del self._cache[key]
            self.misses += 1
            return None
        self.hits += 1
        print(f"  [Cache] HIT - returning cached response")
        return entry["value"]

    def set(self, messages, value, ttl=300):
        key = self._make_key(messages)
        self._cache[key] = {"value": value, "expires": time.time() + ttl}

    def stats(self) -> str:
        total = self.hits + self.misses
        rate = (self.hits / total * 100) if total > 0 else 0
        return f"Cache stats: {self.hits} hits, {self.misses} misses ({rate:.0f}% hit rate)"

Your agent now avoids paying the latency and token cost for repeated queries. Next up, you will add a human-in-the-loop pattern that pauses the agent before it executes dangerous actions.

Add Human-in-the-Loop

Your agent can now handle failures gracefully and avoid redundant calls. But there is a critical gap: nothing stops the agent from executing dangerous actions autonomously.

In production systems, certain actions (publishing reports, sending emails, deleting data) must pause for human approval. This is the human-in-the-loop pattern. In this step, you will walk through how your agent already uses LangGraph's interrupt() to pause execution, surface the pending action to a human, and resume only when approved.

In this step, get ready to:
  • Understand how the checkpointer enables interrupt-based pausing.
  • Trace the conditional routing that detects dangerous tool calls.
  • Test the full approve/reject flow with a publish report query.
Understand the checkpointer and interrupt imports

For the agent to pause mid-execution and resume later, LangGraph needs a way to save the graph's state at the point of interruption. That is what the checkpointer does.

  • In agent.py, find these import lines near the top:
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
  • Now scroll down to where the graph is compiled, and find these lines:
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

What does the checkpointer do?

The InMemorySaver stores the graph's state (messages, pending tool calls, node position) in memory every time a node finishes.

When interrupt() is called, the graph freezes at that exact point. Later, Command(resume=...) picks up from the saved checkpoint and continues. Without a checkpointer, there is no saved state to resume from.

Trace the human review routing logic

The human-in-the-loop pattern has three moving parts: a routing function that detects dangerous tools, a review node that pauses execution, and a post-review router that decides whether to continue or stop.

  • In agent.py, find the check_dangerous_tools function:
def check_dangerous_tools(state: MessagesState):
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return "end"

    for tool_call in last_message.tool_calls:
        if tool_call["name"] == "publish_report":
            return "human_review"
    return "tools"

What does this routing function do?

  • If the model response has no tool calls, route to END (the agent is done talking).
  • If any tool call targets publish_report, route to human_review for approval.
  • Otherwise, route to tools for normal execution (search, calculate).
  • Now find the human_review node function:
def human_review(state: MessagesState):
    last_message = state["messages"][-1]
    tool_call = next(
        tc for tc in last_message.tool_calls if tc["name"] == "publish_report"
    )

    decision = interrupt({
        "action": "publish_report",
        "title": tool_call["args"].get("title", ""),
        "content": tool_call["args"].get("content", ""),
        "message": "Do you approve this report publication?",
    })

    if decision.get("action") == "approve":
        return {"messages": []}
    else:
        rejection = AIMessage(content="Report publication was rejected by human reviewer.")
        return {"messages": [rejection]}

How does interrupt() pause the agent?

The interrupt() call freezes the graph at this exact line. The dictionary passed to it becomes the payload that the calling code can inspect to show the human what is about to happen.

When the human resumes with Command(resume={"action": "approve"}), the graph picks up right where it stopped. The resumed value becomes the return value of interrupt(), which is stored in decision. If the human approves, execution continues to the tool node. If they reject, the agent returns a rejection message and ends.

  • Finally, find route_after_review and the conditional edges that wire this together:
def route_after_review(state: MessagesState):
    last_message = state["messages"][-1]
    if isinstance(last_message, AIMessage) and "rejected" in getattr(last_message, "content", ""):
        return "end"
    return "tools"
workflow.add_conditional_edges(
    "call_model",
    check_dangerous_tools,
    {"tools": "tools", "human_review": "human_review", "end": END},
)
workflow.add_conditional_edges(
    "human_review",
    route_after_review,
    {"tools": "tools", "end": END},
)

Why two sets of conditional edges?

The first set routes from call_model to either safe tools, human review, or end. The second set routes from human_review to either the tool node (approved) or end (rejected). This creates a gate that only opens when a human says so.

Test the full approve and reject flow

The run_with_approval() function handles the full interrupt cycle: invoke the graph, detect the pause via graph.get_state(), show the pending action, and resume based on user input.

  • Run the agent to test all four queries (including the approval flow) by running:
python agent.py

The first three queries run as before (calculate, search, and cache hit). Then you will see a human review prompt.

  • When you see Approve? (y/n): in the terminal, type y and press Enter to approve the report.

What should I see?

Before the approval prompt, the terminal prints the pending action details:

  • Action: publish_report
  • Title: Q3 Results
  • Content: Revenue increased by 15% year over year
  • The message: Do you approve this report publication?

After you type y, the agent executes publish_report and prints a success confirmation.

  • Run the agent a second time to test the rejection path:
python agent.py
  • This time, when you see Approve? (y/n):, type n and press Enter.

The agent prints a rejection message and stops without publishing. This confirms the safety gate works in both directions.

Agent hangs or crashes at the approval step?

  • Make sure checkpointer=InMemorySaver() is passed to workflow.compile(). Without it, interrupt() has no state to save and the graph cannot pause.
  • Confirm that from langgraph.types import interrupt, Command is at the top of agent.py. A missing import causes a NameError.
  • If you see a rate limit error, wait 60 seconds and try again. The free tier has a 30 RPM limit.

Still stuck? Help me debug why my agent crashes at the human review step.

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py and that both the approve and reject paths worked as expected.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq
from langchain.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

from tools import search_web, calculate, publish_report
from resilience import CircuitBreaker, CircuitBreakerOpenError, retry_with_backoff, ResponseCache

load_dotenv()

SYSTEM_PROMPT = """You are a research assistant that can search the web, perform calculations, and publish reports.
Use the appropriate tool for each task. Be concise in your responses."""

tools = [search_web, calculate, publish_report]
tool_node = ToolNode(tools)

llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)
llm_with_tools = llm.bind_tools(tools)

circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
cache = ResponseCache()


@retry_with_backoff(max_retries=2, base_delay=1.0, cap=8.0)
def _invoke_llm(messages):
    return llm_with_tools.invoke(messages)


def call_model(state: MessagesState):
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]

    cached = cache.get(state["messages"])
    if cached is not None:
        return {"messages": [cached]}

    try:
        response = circuit_breaker.call(_invoke_llm, messages)
        cache.set(state["messages"], response)
        return {"messages": [response]}
    except CircuitBreakerOpenError as e:
        error_msg = AIMessage(content=f"Service temporarily unavailable: {e}")
        return {"messages": [error_msg]}
    except Exception as e:
        error_msg = AIMessage(content=f"Error calling model: {e}")
        return {"messages": [error_msg]}


def check_dangerous_tools(state: MessagesState):
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return "end"

    for tool_call in last_message.tool_calls:
        if tool_call["name"] == "publish_report":
            return "human_review"
    return "tools"


def human_review(state: MessagesState):
    last_message = state["messages"][-1]
    tool_call = next(
        tc for tc in last_message.tool_calls if tc["name"] == "publish_report"
    )

    decision = interrupt({
        "action": "publish_report",
        "title": tool_call["args"].get("title", ""),
        "content": tool_call["args"].get("content", ""),
        "message": "Do you approve this report publication?",
    })

    if decision.get("action") == "approve":
        return {"messages": []}
    else:
        rejection = AIMessage(content="Report publication was rejected by human reviewer.")
        return {"messages": [rejection]}


def route_after_review(state: MessagesState):
    last_message = state["messages"][-1]
    if isinstance(last_message, AIMessage) and "rejected" in getattr(last_message, "content", ""):
        return "end"
    return "tools"


workflow = StateGraph(MessagesState)

workflow.add_node("call_model", call_model)
workflow.add_node("tools", tool_node)
workflow.add_node("human_review", human_review)

workflow.add_edge(START, "call_model")
workflow.add_conditional_edges(
    "call_model",
    check_dangerous_tools,
    {"tools": "tools", "human_review": "human_review", "end": END},
)
workflow.add_conditional_edges(
    "human_review",
    route_after_review,
    {"tools": "tools", "end": END},
)
workflow.add_edge("tools", "call_model")

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)


def run_agent(query: str, thread_id: str = "default"):
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print(f"{'='*60}")

    config = {"configurable": {"thread_id": thread_id}}
    input_messages = {"messages": [HumanMessage(content=query)]}

    result = graph.invoke(input_messages, config=config)

    last_ai = None
    for msg in reversed(result["messages"]):
        if isinstance(msg, AIMessage) and msg.content:
            last_ai = msg
            break

    if last_ai:
        print(f"\nAgent: {last_ai.content}")

    print(f"\n{cache.stats()}")
    print(f"Circuit state: {circuit_breaker.state}")
    return result


def run_with_approval(query: str, thread_id: str = "approval_thread"):
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print(f"{'='*60}")

    config = {"configurable": {"thread_id": thread_id}}
    input_messages = {"messages": [HumanMessage(content=query)]}

    result = graph.invoke(input_messages, config=config)

    snapshot = graph.get_state(config)
    if snapshot.next:
        print("\n--- HUMAN REVIEW REQUIRED ---")
        for task in snapshot.tasks:
            if hasattr(task, "interrupts") and task.interrupts:
                for intr in task.interrupts:
                    print(f"  Action: {intr.value.get('action', 'unknown')}")
                    print(f"  Title: {intr.value.get('title', '')}")
                    print(f"  Content: {intr.value.get('content', '')[:100]}")
                    print(f"  {intr.value.get('message', '')}")

        user_input = input("\nApprove? (y/n): ").strip().lower()
        action = "approve" if user_input == "y" else "reject"

        result = graph.invoke(Command(resume={"action": action}), config=config)

        last_ai = None
        for msg in reversed(result["messages"]):
            if isinstance(msg, AIMessage) and msg.content:
                last_ai = msg
                break
        if last_ai:
            print(f"\nAgent: {last_ai.content}")

    print(f"\n{cache.stats()}")
    return result


if __name__ == "__main__":
    run_agent("What is 25 * 17?")
    run_agent("Search for information about circuit breaker patterns")
    run_agent("What is 25 * 17?")
    print("\n--- Testing Human-in-the-Loop ---")
    run_with_approval("Publish a report titled 'Q3 Results' with content 'Revenue increased by 15% year over year'")

Your agent now has a full safety gate. It autonomously handles safe tools (search, calculate) but stops and asks a human before executing anything dangerous. Next up, you will push this further with a model fallback chain in the Secret Mission.

Secret mission

Add a Model Fallback Chain

Your agent currently has one model. If it goes down, everything stops. In production, resilient systems route traffic to a backup model automatically. Can you wire up a fallback chain that kicks in when the primary trips the circuit breaker?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. The Groq free tier never charges you, and all files live on your machine.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The resilient-agent project folder on your Desktop (includes all Python files, .env, and the venv virtual environment).

Keep everything running

No action needed. Choose this if you're still actively building or want to revisit the agent patterns later.

  • Your project folder stays on your Desktop at ~/Desktop/resilient-agent.
  • Your Groq API key remains active in the .env file. Since the free tier never charges, there is no cost risk.
  • The virtual environment and installed packages stay inside the project folder.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files and packages stay intact for when you return.

  • If your virtual environment is currently active, deactivate it by running:
deactivate

When you come back, reactivate by running:

cd ~/Desktop/resilient-agent
source venv/bin/activate

Everything will be exactly as you left it.

Delete - I don't want to use this again

Remove the entire project folder. This deletes all code files, the virtual environment, and your .env file.

  • Deactivate the virtual environment if it is still active:
deactivate
  • Delete the project folder by running:

macOS

rm -rf ~/Desktop/resilient-agent

Windows

rmdir /s /q %USERPROFILE%\Desktop\resilient-agent
  • Optionally, revoke your Groq API key by going to console.groq.com, navigating to the API Keys section, and deleting the key you created for this project.

Nice Work!

Nice Work!

You just built a production-patterned AI agent with real resilience infrastructure. This is not a toy demo. Every component you wrote maps directly to an agentic system design interview answer.

Here's what you learned:

  • Built a multi-tool LangGraph StateGraph with conditional routing and a ReAct-style tool-calling loop that decides when to search, calculate, or publish.
  • Implemented a circuit breaker state machine (Closed/Open/Half-Open) with exponential backoff and a response cache to handle failures gracefully and eliminate redundant LLM calls.
  • Added human-in-the-loop using LangGraph's interrupt pattern to pause execution and require approval before safety-critical actions.
  • Secret Mission: Added a model fallback chain that automatically routes traffic to a backup model when the primary circuit breaker trips.

Ready to quiz yourself?