Module 2: Tool Use Fundamentals
4. The Complete Tool Execution Loop
Capsule overview
In the previous capsule (03) you learned to connect tools to the model with bind_tools() and to inspect the tool_calls it returns. But inspecting isn't enough — someone has to execute those tools, return the results to the model, and let the model decide whether it needs more information or can already answer. That "someone" is the tool execution loop: the complete cycle that turns an instruction from the model into a real action and a final answer.
This loop is the most important concept in the module. If in Module 1 (capsule 03) you implemented a basic ReAct loop with 2 tools and logging, here you'll go deeper: handling multiple tool calls in parallel, robust error handling inside the loop, the exact ToolMessage contract, and the comparison between implementing the loop by hand vs using create_react_agent. When you finish this capsule, you'll be able to draw the user → model → tool_call → execute → ToolMessage → model → response cycle from memory — and you'll understand exactly what happens at each step.
The difference between "knowing how to use" an agent and "understanding" an agent lives here. Frameworks like LangGraph abstract this loop away. That's useful in production — but if you don't understand what's underneath the abstraction, you can't debug it when it fails, you can't optimize it when it's slow, and you can't design custom loops when the framework doesn't cover your case.
The Full Cycle, Step by Step
Cycle diagram
┌─────────────────────────────────────────────────────────────────────┐
│ TOOL EXECUTION LOOP │
│ │
│ ┌──────────────────┐ │
│ │ 1. USER INPUT │ The user sends their message │
│ │ │ → A HumanMessage is created │
│ └────────┬──────────┘ │
│ ↓ │
│ ┌──────────────────┐ │
│ │ 2. MODEL INVOKE │ The model receives every message │
│ │ │ → Returns an AIMessage │
│ └────────┬──────────┘ │
│ ↓ │
│ ┌──────────────────┐ │
│ │ 3. CHECK │ Does the AIMessage have tool_calls? │
│ │ TOOL_CALLS │ │
│ └────────┬──────────┘ │
│ / \ │
│ Yes No → Return response.content (END) │
│ ↓ │
│ ┌──────────────────┐ │
│ │ 4. EXECUTE TOOLS │ For EACH tool_call: │
│ │ │ - Look up the tool by name │
│ │ │ - Run it with the args │
│ │ │ - Capture the result or the error │
│ └────────┬──────────┘ │
│ ↓ │
│ ┌──────────────────┐ │
│ │ 5. CREATE │ For each result: │
│ │ TOOLMESSAGES │ - ToolMessage(content, tool_call_id) │
│ │ │ - Append it to messages │
│ └────────┬──────────┘ │
│ ↓ │
│ Back to step 2 │
│ │
└─────────────────────────────────────────────────────────────────────┘
Each step in detail
Step 1 — User Input: Everything starts with a HumanMessage. This is the only external input to the loop. From here on, everything is a conversation between the model and the tools.
Step 2 — Model Invoke: The model receives the complete messages list and produces an AIMessage. This AIMessage can contain two things: content (text for the user) and/or tool_calls (instructions to run tools). It never executes tools directly — it only emits instructions.
Step 3 — Check tool_calls: This is the loop's central fork. If response.tool_calls is empty, the model decided it has enough information — you return response.content and you're done. If it has tool_calls, the loop continues.
Step 4 — Execute Tools: For each tool_call in the response, you look up the matching tool in your tools_by_name dictionary, run it with the arguments the model asked for, and capture the result. If the tool fails, you capture the error as a string — never let an exception break the loop.
Step 5 — Create ToolMessages: Each result is packaged as a ToolMessage with two mandatory fields: content (the result as a string) and tool_call_id (the exact ID of the tool_call that produced this response). The model uses this ID to know which result belongs to which call. After appending all the ToolMessages, you go back to step 2.
The message types in the loop
| Type | Who creates it | What it contains | When it appears |
|---|---|---|---|
HumanMessage | The user | The original input | Start of the loop |
AIMessage | The model | content and/or tool_calls | Every iteration |
ToolMessage | Your code | A tool's result + tool_call_id | After running tools |
SystemMessage | Your code | General instructions | Optional, at the start |
Manual Implementation
Setup: 3 tools for the loop
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage, SystemMessage
from datetime import datetime
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Use for: temperature and weather conditions.
Example: get_weather('Madrid')"""
climates = {
"Madrid": "22°C, sunny",
"Barcelona": "20°C, cloudy",
"Mexico City": "25°C, partly cloudy",
"Buenos Aires": "18°C, rainy",
"Bogotá": "14°C, cloudy",
}
return f"Weather in {city}: {climates.get(city, '18°C, partly cloudy')}"
@tool
def calculator(expression: str) -> str:
"""Evaluate a safe math expression.
Only accepts: numbers, +, -, *, /, (), .
Example: calculator('25 * 4 + 10')"""
allowed = set("0123456789+-*/(). ")
if not all(c in allowed for c in expression):
return "Error: expression contains disallowed characters"
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"Calculation error: {e}"
@tool
def get_date() -> str:
"""Get the current date, time and day of the week.
Takes no arguments."""
now = datetime.now()
return now.strftime("%Y-%m-%d %H:%M — %A")
The complete loop
model = init_chat_model("openai:gpt-4.1-mini")
tools = [get_weather, calculator, get_date]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)
def agent_loop(user_input: str, max_iterations: int = 5) -> str:
"""The complete tool execution loop, with error handling."""
messages = [HumanMessage(content=user_input)]
for i in range(max_iterations):
response = model_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
return response.content or "No answer"
for tc in response.tool_calls:
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except KeyError:
result = f"Error: tool '{tc['name']}' does not exist. Available: {list(tools_by_name.keys())}"
except Exception as e:
result = f"Error running {tc['name']}: {e}"
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
return "Iteration limit reached without a final answer"
Running it and the expected output
print(agent_loop("What day is it, how much is 150 * 3, and what's the weather in Bogotá?"))
# The model calls 3 tools in parallel (1 iteration):
# get_date() → "2026-03-08 14:30 — Sunday"
# calculator("150 * 3") → "450"
# get_weather("Bogotá") → "Weather in Bogotá: 14°C, cloudy"
#
# On iteration 2, the model has every result and answers:
# "Today is Sunday, March 8, 2026. 150 × 3 = 450.
# The weather in Bogotá is 14°C and cloudy."
A version with step-by-step logging
So you can see exactly what happens on each iteration:
def agent_loop_verbose(user_input: str, max_iterations: int = 5) -> str:
"""The loop with detailed logging of every step."""
messages = [HumanMessage(content=user_input)]
print(f"INPUT: {user_input}\n{'=' * 70}")
for i in range(max_iterations):
print(f"\n--- Iteration {i + 1} (messages: {len(messages)}) ---")
response = model_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
print(f" FINAL ANSWER: {response.content}")
return response.content or "No answer"
print(f" TOOL CALLS: {len(response.tool_calls)}")
for tc in response.tool_calls:
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except Exception as e:
result = f"Error: {e}"
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
print(f" {tc['name']}({tc['args']}) → {result}")
return "Iteration limit reached"
agent_loop_verbose("How much is 99 * 77 and what's the weather in Madrid?")
# Output:
# INPUT: How much is 99 * 77 and what's the weather in Madrid?
# ======================================================================
#
# --- Iteration 1 (messages: 1) ---
# TOOL CALLS: 2
# calculator({'expression': '99 * 77'}) → 7623
# get_weather({'city': 'Madrid'}) → Weather in Madrid: 22°C, sunny
#
# --- Iteration 2 (messages: 4) ---
# FINAL ANSWER: 99 × 77 = 7,623. The weather in Madrid is 22°C and sunny.
Notice two things in that output:
- Iteration 1 has 1 message (the HumanMessage). Iteration 2 has 4 (HumanMessage + AIMessage with tool_calls + 2 ToolMessages). The context grows with every pass.
- The model called 2 tools in parallel in a single iteration. Modern models (GPT-4.1, Claude) detect independent sub-tasks and call them simultaneously.
ToolMessage: The Return Contract
The two mandatory fields
Every ToolMessage needs exactly two things:
ToolMessage(
content=str(result), # The result as a string
tool_call_id=tc["id"] # The ID of the tool_call that produced it
)
The tool_call_id is the glue connecting the answer to the question. Without that ID, the model doesn't know which result belongs to which tool call — and the API will raise an error.
Where the tool_call_id comes from
response = model_with_tools.invoke(messages)
for tc in response.tool_calls:
print(tc)
# {
# "name": "get_weather",
# "args": {"city": "Madrid"},
# "id": "call_abc123" ← This is the tool_call_id
# }
The id is generated by the provider (OpenAI, Anthropic). It's unique per call. Your job is to use it exactly as-is when you create the ToolMessage — don't modify it, don't make one up.
Content is always a string
ToolMessage.content must be a string. If your tool returns a dict or a number, convert it with str(). For complex data (lists, nested dicts), use json.dumps():
import json
@tool
def search_products(query: str) -> str:
"""Search for products by name."""
products = [{"name": "Laptop", "price": 999}, {"name": "Mouse", "price": 25}]
return json.dumps(products, ensure_ascii=False)
A ToolMessage carrying an error
When a tool fails, don't raise the exception — send the error as the ToolMessage's content. The model can interpret the error and decide what to do (retry with different args, use another tool, or tell the user):
for tc in response.tool_calls:
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except Exception as e:
result = f"Error running {tc['name']}: {str(e)}"
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
The model receives "Error running calculator: invalid syntax" and can decide: "OK, the expression was wrong, let me try another syntax" or "I can't compute this, I'll tell the user."
One ToolMessage per tool_call
If the model asked for 3 tool calls, you need 3 ToolMessages — one for each tool_call with its own tool_call_id. Not one with everything bundled together. Each ToolMessage is paired with its original tool_call through the ID.
Handling Multiple Tool Calls
When the model makes multiple calls
Modern models do parallel function calling when they detect independent sub-tasks in the same message. If the user asks "What's the weather in Madrid, how much is 5*5, and what day is it?", the model emits 3 tool_calls in a single response because none depends on the others.
But if one task depends on another's result ("How much is the temperature in Madrid multiplied by 2?"), the model makes sequential calls: first it asks for the weather (iteration 1), receives the result, and then asks for the calculation using that data (iteration 2).
Processing parallel calls
for tc in response.tool_calls:
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except KeyError:
result = f"Error: tool '{tc['name']}' does not exist"
except Exception as e:
result = f"Error: {e}"
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
This loop processes each tool_call sequentially in your code (one after another), but from the model's perspective they were emitted "in parallel" in a single AIMessage. Execution order doesn't matter because the sub-tasks are independent.
Genuinely parallel execution (optional)
If your tools call external APIs with significant latency (100ms+), you can run them in parallel with ThreadPoolExecutor:
from concurrent.futures import ThreadPoolExecutor
# Inside the loop, replace the sequential for with:
with ThreadPoolExecutor() as executor:
futures = {
tc["id"]: (executor.submit(tools_by_name[tc["name"]].invoke, tc["args"]), tc)
for tc in response.tool_calls
}
for tool_call_id, (future, tc) in futures.items():
try:
result = str(future.result(timeout=30))
except Exception as e:
result = f"Error: {e}"
messages.append(ToolMessage(content=result, tool_call_id=tool_call_id))
For tools that do instant local computation, the sequential version is simpler and good enough.
Dependent calls: the model handles it
You don't need to detect dependencies between calls — the model does it for you. If one task depends on another's result, the model makes sequential calls across multiple iterations:
print(agent_loop("Get the temperature in Madrid and multiply it by 3"))
# Iteration 1: tool_calls: [get_weather("Madrid")] → "Weather in Madrid: 22°C, sunny"
# Iteration 2: tool_calls: [calculator("22 * 3")] → "66"
# Iteration 3: answers "The temperature in Madrid is 22°C. Multiplied by 3: 66."
The model manages the dependencies because it sees the entire message history. After each iteration, the context includes the ToolMessages with previous results.
Manual Loop vs create_react_agent
create_react_agent in 5 lines
from langgraph.prebuilt import create_react_agent
model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, tools)
result = agent.invoke({"messages": [("user", "Weather in Madrid and 25*4?")]})
print(result["messages"][-1].content)
It does exactly the same thing as your ~25-line manual loop. So why implement it by hand?
When to use the manual loop
-
Learning: You're here, learning how an agent works. The manual loop shows you every step. When you use
create_react_agentin production, you'll know exactly what's underneath. -
Custom stop conditions: You need to stop the loop for reasons the framework doesn't cover — a token budget, a maximum time, specific content in a tool's result. Example: checking
response.usage_metadata["total_tokens"]after each iteration and cutting off if it exceeds the budget. -
Custom middleware: You need to inject logic between steps: logging to a database, latency metrics, content filtering, an audit trail.
-
Deep debugging: Something fails in production and you need to see exactly which messages are being sent to the model on each iteration. With the manual loop, drop in a breakpoint or a print and you're done.
When to use create_react_agent
- Stable production: Your agent already works, you want less code and better maintainability.
- Streaming:
create_react_agentsupports token streaming out of the box. Implementing it by hand takes more code. - Checkpointing and memory: With LangGraph, you add
MemorySaverand the agent persists state across conversations. In the manual loop, you'd have to implement persistence yourself. - The LangGraph ecosystem: You want LangGraph Studio to visualize the graph, LangSmith for tracing, deployment with LangGraph Cloud. All of it integrates natively with
create_react_agent.
Comparison
| Aspect | Manual loop | create_react_agent | AgentExecutor (legacy) |
|---|---|---|---|
| Lines of code | ~25-30 | ~5 | ~5 |
| Loop control | Total — every step is your code | Via configuration and callbacks | Via callbacks |
| Custom stop conditions | Direct — edit the if/for | recursion_limit + custom functions | max_iterations + early stopping |
| Error handling | You design it | Default + customizable | Default |
| Streaming | Implement by hand | Built-in | Built-in |
| Checkpointing | Implement by hand | MemorySaver / PostgresSaver | Not supported |
| Debugging | Print/breakpoint wherever you like | LangSmith tracing | LangSmith tracing |
| Parallel tool calls | ThreadPoolExecutor by hand | Automatic | Automatic |
| Visual tooling | No | LangGraph Studio | No |
| Current status | Always available | Recommended (LangGraph) | Deprecated — don't use |
AgentExecutor is deprecated. If you see tutorials using it, they're legacy. The official recommendation is LangGraph's create_react_agent for production and the manual loop for learning.
A progressive recommendation:
- Now (M2): The manual loop, to understand the cycle
- M3: The manual loop with advanced patterns (parallel, routing, retry)
- M4+:
create_react_agentand custom StateGraph for production
Connection to the Project
In this module's project (capsule 08), you'll build an agent with 5 real external tools. The tool execution loop you implemented here will be that agent's heart. Specifically:
- You'll use
agent_loopas the base, extended with 5 real tools (Tavily search, weather API, calculator, file reader, datetime) - You'll add robust error handling to cope with external API failures (timeouts, rate limits, 404s)
- You'll implement logging to see every tool call and its result
- You'll compare the manual version against the
create_react_agentversion
In the evolving project (Modules 4-10):
- M4: The loop becomes a
StateGraph— each step (invoke, check, execute) turns into a node of the graph - M5: The "check tool_calls" step expands: the agent doesn't just decide "do I call a tool or answer?" but "what is my plan for this research?"
- M6: The messages get persisted with
MemorySaver— if the agent stops, it resumes where it left off - M7: Tools are loaded dynamically from MCP servers — the
tools_by_namedictionary is built at runtime
Troubleshooting
Problem 1: "ToolMessage has no matching tool_call_id"
Cause: The ToolMessage's tool_call_id doesn't match any id in the previous AIMessage's tool_calls. This happens when you invent an ID, hardcode it, or copy it wrong.
Fix: Always use tc["id"] straight from the tool_call:
for tc in response.tool_calls:
result = tools_by_name[tc["name"]].invoke(tc["args"])
messages.append(ToolMessage(
content=str(result),
tool_call_id=tc["id"] # Always from the original tool_call
))
Problem 2: An infinite loop — the model never stops calling tools
Cause: The model has no clear instructions on when to stop, or the information it receives from the tools isn't enough to answer.
Fix: Add a system prompt that tells it when to finish, and make sure max_iterations is reasonable:
messages = [
SystemMessage(content=(
"You are a helpful assistant. Use tools only when you need data "
"you don't have. Once you have all the information, answer directly "
"without calling more tools."
)),
HumanMessage(content=user_input)
]
Problem 3: KeyError — the model asks for a tool that doesn't exist
Cause: The model "hallucinates" a tool name that isn't in the list. Rare with modern models, but possible.
Fix: Validate the name before running it and send a ToolMessage with the error:
for tc in response.tool_calls:
if tc["name"] not in tools_by_name:
messages.append(ToolMessage(
content=f"Error: '{tc['name']}' does not exist. Available tools: {list(tools_by_name.keys())}",
tool_call_id=tc["id"]
))
continue
result = tools_by_name[tc["name"]].invoke(tc["args"])
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
On receiving the error as a ToolMessage, the model self-corrects on the next iteration and calls the right tool.
Problem 4: The model repeats the same tool call with the same args
Cause: The model isn't "seeing" the previous ToolMessages, or the tool returns ambiguous results that the model doesn't interpret as a sufficient answer.
Fix: Verify that you're appending the ToolMessages correctly to the messages array. If the problem persists, add duplicate detection:
seen_calls = set()
for tc in response.tool_calls:
signature = f"{tc['name']}:{json.dumps(tc['args'], sort_keys=True)}"
if signature in seen_calls:
messages.append(ToolMessage(
content=f"I already ran {tc['name']} with these arguments. The previous result is in the history.",
tool_call_id=tc["id"]
))
continue
seen_calls.add(signature)
result = tools_by_name[tc["name"]].invoke(tc["args"])
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
Problem 5: ToolMessage.content cannot be None
Cause: The tool returns None and you pass it straight through as content.
Fix: Always convert the result to a string with a fallback:
result = tools_by_name[tc["name"]].invoke(tc["args"])
content = str(result) if result is not None else "Tool ran with no result"
messages.append(ToolMessage(content=content, tool_call_id=tc["id"]))
Exercises
Exercise 1: Trace the messages state (Easy)
For the input "What day is it and how much is 200/4?", trace the state of the messages array after each step of the loop. Assume the model calls both tools in parallel.
See solution
Start:
messages = [HumanMessage("What day is it and how much is 200/4?")]
# len: 1
Iteration 1 — after model.invoke:
messages = [
HumanMessage("What day is it and how much is 200/4?"),
AIMessage(tool_calls=[
{"name": "get_date", "args": {}, "id": "call_001"},
{"name": "calculator", "args": {"expression": "200/4"}, "id": "call_002"}
])
]
# len: 2
Iteration 1 — after running the tools:
messages = [
HumanMessage("What day is it and how much is 200/4?"),
AIMessage(tool_calls=[...]),
ToolMessage(content="2026-03-08 14:30 — Saturday", tool_call_id="call_001"),
ToolMessage(content="50.0", tool_call_id="call_002")
]
# len: 4
Iteration 2 — after model.invoke (no tool_calls):
messages = [
HumanMessage("What day is it and how much is 200/4?"),
AIMessage(tool_calls=[...]),
ToolMessage(content="2026-03-08...", tool_call_id="call_001"),
ToolMessage(content="50.0", tool_call_id="call_002"),
AIMessage(content="Today is Saturday, March 8, 2026. 200 ÷ 4 = 50.")
]
# len: 5
Return: "Today is Saturday, March 8, 2026. 200 ÷ 4 = 50."
Total: 2 iterations, 2 tool calls (parallel), 2 LLM calls.
Exercise 2: Add automatic retries (Medium)
Modify agent_loop so that if a tool fails, it retries up to 2 times before sending the error as a ToolMessage.
See solution
def agent_loop_with_retry(user_input: str, max_iterations: int = 5, max_retries: int = 2) -> str:
messages = [HumanMessage(content=user_input)]
for i in range(max_iterations):
response = model_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
return response.content or "No answer"
for tc in response.tool_calls:
result = None
for attempt in range(max_retries + 1):
try:
result = str(tools_by_name[tc["name"]].invoke(tc["args"]))
break
except Exception as e:
if attempt == max_retries:
result = f"Error after {max_retries + 1} attempts: {e}"
else:
print(f" Retry {attempt + 1} for {tc['name']}...")
messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
return "Iteration limit reached"
The for attempt in range(max_retries + 1) runs the tool up to 3 times (1 original attempt + 2 retries). If it fails all 3 times, it sends the error as a ToolMessage so the model can decide what to do.
Exercise 3: A loop with a system prompt and validation (Medium)
Implement a loop where the system prompt says "Only answer in Spanish. If a tool returns data in English, translate it." Verify it works with get_date(), which returns the day in English (e.g. "Saturday").
See solution
def agent_spanish(user_input: str, max_iterations: int = 5) -> str:
messages = [
SystemMessage(content=(
"You are an assistant that answers ONLY in Spanish. "
"If a tool returns data in English (such as day or month "
"names), translate it into Spanish in your final answer."
)),
HumanMessage(content=user_input)
]
for i in range(max_iterations):
response = model_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
return response.content or "No answer"
for tc in response.tool_calls:
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except Exception as e:
result = f"Error: {e}"
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
return "Iteration limit reached"
print(agent_spanish("¿Qué día es hoy?"))
# get_date() returns "2026-03-08 14:30 — Saturday"
# The model translates: "Hoy es sábado 8 de marzo de 2026."
The model receives the ToolMessage with "Saturday" and, following the system prompt's instruction, translates it to "sábado" in the final answer. The system prompt controls the model's post-tool behavior.
Exercise 4: Loop metrics (Medium)
Modify the loop so it returns a dict with: the answer, the number of iterations used, the total number of tool calls, and the names of every tool called.
See solution
def agent_loop_metrics(user_input: str, max_iterations: int = 5) -> dict:
messages = [HumanMessage(content=user_input)]
total_tool_calls = 0
tools_used = []
for i in range(max_iterations):
response = model_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
return {
"response": response.content or "No answer",
"iterations": i + 1,
"total_tool_calls": total_tool_calls,
"tools_used": tools_used,
"total_messages": len(messages),
}
total_tool_calls += len(response.tool_calls)
for tc in response.tool_calls:
tools_used.append(tc["name"])
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except Exception as e:
result = f"Error: {e}"
messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
return {
"response": "Iteration limit reached",
"iterations": max_iterations,
"total_tool_calls": total_tool_calls,
"tools_used": tools_used,
"total_messages": len(messages),
}
metrics = agent_loop_metrics("Weather in Barcelona, how much is 15*23, and what day is it?")
print(f"Answer: {metrics['response']}")
print(f"Iterations: {metrics['iterations']}")
print(f"Tool calls: {metrics['total_tool_calls']}")
print(f"Tools: {metrics['tools_used']}")
print(f"Messages: {metrics['total_messages']}")
# Expected output:
# Answer: The weather in Barcelona is 20°C and cloudy. 15 × 23 = 345. Today is...
# Iterations: 2
# Tool calls: 3
# Tools: ['get_weather', 'calculator', 'get_date']
# Messages: 6
These metrics are the first step toward agent observability. In Module 10 you'll use LangSmith for automatic metrics (tokens, latency, cost). Here you implement them by hand to understand what's worth measuring.
Exercise 5: A loop with a circuit breaker (Hard)
Implement a "circuit breaker": if a specific tool fails 3 times in a row during the session, disable it for the rest of the conversation. The model should receive a message telling it the tool was disabled.
See solution
def agent_loop_circuit_breaker(
user_input: str,
max_iterations: int = 5,
failure_threshold: int = 3
) -> str:
messages = [HumanMessage(content=user_input)]
failure_counts = {name: 0 for name in tools_by_name}
disabled_tools = set()
for i in range(max_iterations):
response = model_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
return response.content or "No answer"
for tc in response.tool_calls:
name = tc["name"]
if name in disabled_tools:
messages.append(ToolMessage(
content=f"Tool '{name}' disabled after repeated failures. Use another tool or answer without it.",
tool_call_id=tc["id"]
))
continue
if name not in tools_by_name:
messages.append(ToolMessage(
content=f"Error: tool '{name}' does not exist",
tool_call_id=tc["id"]
))
continue
try:
result = str(tools_by_name[name].invoke(tc["args"]))
failure_counts[name] = 0
except Exception as e:
failure_counts[name] += 1
result = f"Error ({failure_counts[name]}/{failure_threshold}): {e}"
if failure_counts[name] >= failure_threshold:
disabled_tools.add(name)
result += f" — Tool '{name}' DISABLED after {failure_threshold} consecutive failures."
messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
return "Iteration limit reached"
The circuit breaker is a production pattern that stops an agent from retrying a broken tool indefinitely. Each success resets the counter (failure_counts[name] = 0). Each failure increments it. When it hits the threshold, the tool "opens" (is disabled) and the model receives a message explaining why. The model then has to adapt and use a different strategy.
Summary
In this capsule you learned:
- The complete cycle of tool execution:
user → model → tool_call → execute → ToolMessage → model → response - Every iteration of the loop has 5 steps: input → invoke → check → execute → create ToolMessages
- ToolMessage requires two mandatory fields:
content(a string) andtool_call_id(from the original tool_call) - Modern models do parallel function calling: multiple tools in a single iteration
- Error handling inside the loop is critical: catch exceptions and send them as ToolMessages so the model can self-correct
- The manual loop gives total control (custom stop conditions, logging, middleware)
- create_react_agent gives productivity (less code, streaming, checkpointing)
- AgentExecutor is deprecated — don't use it in new code
Next capsule: Built-in Tools and External Tools — you'll connect your agent to real APIs (Tavily, DuckDuckGo, weather) instead of local mocks.
Additional Resources
- LangGraph create_react_agent Reference — The prebuilt agent API that abstracts the tool execution loop
- LangChain ToolMessage API — The complete ToolMessage reference
- OpenAI Function Calling — Parallel Calls — How OpenAI implements parallel function calling
- Anthropic Tool Use Documentation — Tool use implementation in Claude (same concept, different provider)
- LangGraph How-To: Tool Calling — The official tool calling guide with LangGraph
- Circuit Breaker Pattern — The production pattern you implemented in Exercise 5