A functional Python AI agent combines an LLM, a tool-calling loop, and state management into an autonomous system that decides what to do next at each…
Map the four components of a Python AI agent — LLM, tools, loop, and state — and see how they wire together in a single diagram before writing any code. The running example is a research agent that searches the web, reads pages, and writes a summary.
Maps the four core components of a Python AI agent — LLM, tools, loop, and state — and shows how they wire together before any framework is introduced.
Why this matters: Every agent you build, debug, or evaluate reduces to these four parts; knowing which component owns which responsibility is the fastest way to diagnose misbehaviour.
Your research script calls an LLM once and prints an answer. But what if the answer requires three web searches, two page reads, and a judgment call? A single call can't do that. You need an .
An is a language model in a loop that takes actions, not just answers. On each turn it reasons about the goal. It calls a such as web search or page reader. It reads the result. That shapes the next decision. The cycle repeats until the task is done.
Every Python agent is built from four components: the , the , the , and the . Knowing which component owns which responsibility helps you debug when it misbehaves.
Each component has one job; none of them overlap.
The LLM never calls a tool directly — it emits a structured (a JSON object naming the tool and its arguments). The loop intercepts that, runs the real function, and appends the result to state as an . The LLM sees the observation on the next turn.
Goal: summarise the latest Python packaging changes. Here is how each component fires, in order.
web_search(query="Python packaging 2024").web_search function and appends the result to state as an .read_page(url="...").Notice that the LLM never "knows" it searched the web — it only sees text in its (the model's working memory). The loop is the only part that actually executes code.
def run_agent(goal: str): messages = [{"role": "user", "content": goal}] while True: response = llm.chat(messages) # returns a string if response.tool_call: result = run_tool(response.tool_call) messages.append(result) else: return response.content # done
This looks right — but it has no exit guard. Before reading on, predict: what happens if the LLM keeps emitting tool calls and never returns a final answer?
It runs forever (or until your API budget runs out). You'll see no error — just an infinite stream of tool calls and API charges. The fix is a max_iterations guard: track a counter and raise an error or return a partial answer when it's exceeded. This is the most common silent failure in first-draft agents.
MAX_ITER = 10 def run_agent(goal: str): state = {"messages": [{"role": "user", "content": goal}]} for _ in range(MAX_ITER): # loop with termination guard response = llm.chat(state["messages"]) if not response.tool_call: # termination condition met return response.content obs = run_tool(response.tool_call) # execute tool, get observation state["messages"].append(obs) # update state raise RuntimeError("Max iterations reached")
Each of the four components now has a clear home: llm.chat() is the LLM, run_tool() is the tool layer, the for loop is the , and state["messages"] is the . The MAX_ITER guard is the .
The LLM sees the full message history including the new observation — but it does NOT see the Python variables, the loop counter, or anything outside state['messages']. Its entire world is the context window. If you forget to append the observation, the LLM repeats the same tool call on the next turn.
TOOLS = {
"web_search": web_search,
"read_page": read_page,
}
def run_tool(tool_call: dict) -> dict:
name = tool_call["name"]
args = tool_call["args"]
# TODO: look up `name` in TOOLS, call it with **args,
# and return {"role": "tool", "content": <result>}
...Stop — attempt the TODO before revealing. The missing piece is the crux of the tool layer: dispatching a named call and packaging the result as a message the LLM can read.
TOOLS[name] gives you the function. (2) The return dict must have role: "tool" so the LLM treats it as an observation, not a user message.result = TOOLS[name](**args)
return {"role": "tool", "content": str(result)}
Changed lines vs Stage 2: run_tool() was a black box — now it dispatches by name from a registry and wraps the output in the correct message role. The role key is what changed most: without it, the LLM sees the observation as a user message and its reasoning goes off-track.
Three failure patterns cause most first-agent bugs. Two produce no error message.
state["messages"]. The LLM repeats the identical tool call every turn. The same tool fires in an infinite cycle with no progress.TypeError. The LLM's reasoning looks correct; only dispatch fails.Quick verify checklist: (1) confirm a guard exists and raises; (2) confirm every tool result is appended to state before the next LLM call; (3) confirm each tool's argument names match the Python function signature exactly.
You have the skeleton — loop, state, tool dispatch. But llm.chat() is still a placeholder. The agent can't reason until that slot is filled with a real model call.
The next module wires the OpenAI Python client (v1.x) into that slot. You'll add a that gives the research agent its persona. Define a schema with for type-safe tool calls. Add retry logic so transient API errors don't crash the loop.
Wire the OpenAI Python client (v1.x) into the research agent's LLM slot, adding system prompts, structured output schemas with Pydantic, and retry logic so every call returns a predictable object. You'll complete a partially written `call_llm()` function by adding the missing validation step.
Wire the OpenAI Python client into the research agent's LLM slot, enforce structured Pydantic output, and add retry logic so every call returns a predictable typed object.
Why this matters: A reliable call_llm() function is the foundation every other agent component depends on — without typed, validated output the tool dispatcher and agent loop can't function safely.
Decision this forces: Structured output via response_format / JSON mode vs. plain text + regex parsing — which to use and when.
The four are: the (reasoning core), (actions), the (reason-act-observe cycle), and (persistent memory).
This module fills the LLM slot: calling it reliably, shaping output, and making every response a typed Python object.
Your agent's loop dispatches tool calls based on LLM decisions. Free-form prose breaks this dispatch.
A schema guarantees the response matches — no regex, no surprises.
A validated model gives your dispatcher typed fields to branch on, not strings to guess.
The tradeoff: response_format adds latency and requires gpt-4o or later. For simple tasks it's overkill.
import openai client = openai.OpenAI() # reads OPENAI_API_KEY from env def call_llm(messages: list[dict]) -> str: response = client.chat.completions.create( model="gpt-4o", messages=messages, ) return response.choices[0].message.content result = call_llm([{"role": "user", "content": "Search for: AI agents"}]) print(result) # unpredictable prose — no schema, no system prompt
This is the obvious first attempt: a bare API call with no and no schema.
result is free-form prose, e.g. "Sure! Here are some findings on AI agents…" — the loop has no reliable key to branch on, so any tool dispatch based on parsing this string will fail or silently do the wrong thing when the model changes its phrasing.
from pydantic import BaseModel import openai, json client = openai.OpenAI() class AgentDecision(BaseModel): action: str # e.g. "search" | "read" | "finish" argument: str # the query or URL reasoning: str SYSTEM_PROMPT = ( "You are a research agent. Reply ONLY with valid JSON " "matching the schema provided." ) def call_llm( history: list[dict], ) -> AgentDecision: messages = [{"role": "system", "content": SYSTEM_PROMPT}] + history response = client.chat.completions.create( model="gpt-4o", messages=messages, response_format={"type": "json_object"}, ) raw = response.choices[0].message.content return AgentDecision.model_validate(json.loads(raw))
Three changes from Stage 1: a anchors the model's role, response_format forces JSON output, and AgentDecision.model_validate() turns the raw string into a typed object your loop can branch on.
It returns an AgentDecision instance: AgentDecision(action='search', argument='AI agents 2024', reasoning='Start broad'). The loop can now safely do if decision.action == 'search': ... without any string parsing.
import time, openai from pydantic import ValidationError MAX_RETRIES = 3 def call_llm_with_retry(history: list[dict]) -> AgentDecision: for attempt in range(MAX_RETRIES): try: return call_llm(history) # from Stage 2 except (openai.RateLimitError, openai.APIError, ValidationError) as exc: if attempt == MAX_RETRIES - 1: raise time.sleep(2 ** attempt) # 1s, 2s, 4s raise RuntimeError("unreachable")
The retry wrapper catches three failure classes: rate-limit throttling, transient API errors, and ValidationError (the model returned JSON that didn't match the schema).
Exponential backoff — 1 s, 2 s, 4 s — avoids hammering the API during a rate-limit window while still recovering quickly from a one-off blip.
Attempt 0 fails → sleep(20) = sleep(1). Attempt 1 succeeds → returns immediately. If attempt 1 had also failed → sleep(21) = sleep(2). The delays are 1 s, 2 s, 4 s for attempts 0, 1, 2.
# Variation: research agent needs a *summary* decision with extra `sources` field. class SummaryDecision(BaseModel): action: str argument: str sources: list[str] # NEW — list of URLs to cite def call_summary_llm(history: list[dict]) -> SummaryDecision: messages = [{"role": "system", "content": SYSTEM_PROMPT}] + history response = client.chat.completions.create( model="gpt-4o", messages=messages, response_format={"type": "json_object"}, ) raw = response.choices[0].message.content # TODO: parse `raw` and return a validated SummaryDecision # Hint 1: you need json.loads() first. # Hint 2: use the same Pydantic method as Stage 2.
Stop — attempt the TODO before revealing the answer.
The schema has a new sources field; everything else mirrors Stage 2. The missing line is the crux of this module: turning a raw JSON string into a validated typed object.
return SummaryDecision.model_validate(json.loads(raw))
Changed from Stage 2: only the class name (SummaryDecision instead of AgentDecision) — the pattern is identical. If the model omits sources, Pydantic raises ValidationError immediately, which the retry wrapper in Stage 3 will catch.
| Option | Output reliability | Schema enforcement | Downstream type safety | When to choose | Cost | Complexity |
|---|---|---|---|---|---|---|
| response_format / JSON mode | API-guaranteed valid JSON on every call | Hard — model is constrained to the schema | Pydantic parses directly; fields are typed | When the agent loop must branch on the LLM's decision (tool name, action type) — any path where a missing key would crash the dispatcher. | Slightly higher latency; requires gpt-4o or a model that supports structured outputs | Low — define a Pydantic model, pass its schema once |
| Plain text + regex parsing | Fragile — any phrasing change breaks the parse | None — model can ignore the format instruction | Manual casting; silent wrong-type bugs are common | When output is a single scalar (a yes/no, a score) and the schema would add more complexity than it saves — or when the model doesn't support structured outputs. | No extra latency, but brittle — regex breaks on model wording changes | Low to start; grows fast as edge cases accumulate |
ValidationError: field required every call. Fix: sync schema description with model class.openai.BadRequestError: maximum context length exceeded. Fix: trim or summarise history."sources": "https://example.com" (string, not list). Without validation, this crashes downstream. Fix: use model_validate(), not dict.get()."system"? Missing system messages make models ignore schema instructions.model_validate() (not json.loads())? Raw JSON loads won't catch type mismatches.ValidationError in addition to API errors? Generated code often forgets this.response_format={"type": "json_object"}? Without it, the model may return prose.With reliable, typed call_llm() in place, the next module defines tools and writes the dispatcher that routes each AgentDecision to the right action.
Define two tools for the research agent (a web-search stub and a page-reader), register them as OpenAI function schemas, and write the dispatcher that routes the model's tool-call response to the right Python function. You'll complete the dispatcher by adding the missing result-serialization step.
Define tool functions and their JSON schemas, then build the dispatcher that routes the model's tool-call response to the right Python function.
Why this matters: Your SEO research agent can't search the web or read pages until you wire up tools — this module adds the action layer that turns model intent into real results.
Decision this forces: JSON function-calling schema (OpenAI-style) vs. code-generation style (smolagents CodeAgent) — tradeoffs for safety and flexibility.
The LLM slot returns a structured response object. But the model can only reason — it can't search the web or read pages on its own.
That gap is exactly what closes. You describe capabilities to the model as typed schemas. The model responds with a structured request to invoke one. Your code does the actual work and hands the result back.
This module adds two missing pieces: tool definitions and the dispatcher. Together they turn the model's intent into real actions.
A is the contract between your code and the model. It names a capability, describes parameters with types and required fields, and sets the boundary of what the model may request.
When the model decides a tool is needed, it emits a — a JSON object with the function name and arguments — instead of plain text.
Your dispatcher reads that object, validates the arguments, and routes to the matching Python function. It returns the result as a new message in the .
The schema is an action boundary, not just documentation. The model proposes arguments, but your code decides whether to execute them.
# Two plain Python functions — no framework needed def web_search(query: str) -> str: return f"[stub] top result for: {query}" def read_page(url: str) -> str: return f"[stub] page content from: {url}" # Build the OpenAI function-schema list TOOLS = [ {"type": "function", "function": {"name": "web_search", "description": "Search the web and return a snippet.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}}, {"type": "function", "function": {"name": "read_page", "description": "Fetch and return the text of a web page.", "parameters": {"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]}}}, ]
Each tool is a plain Python function paired with a JSON schema dict that the OpenAI API reads. The schema's "required" array tells the model which arguments it must always supply — omitting it lets the model skip fields silently.
No. When the model answers directly, finish_reason is "stop" and tool_calls is null (or absent). A tool_call field only appears when finish_reason is "tool_calls". Your dispatcher must check finish_reason first, before trying to read tool_calls.
import json # Registry maps name → callable TOOL_REGISTRY = {"web_search": web_search, "read_page": read_page} def dispatch(tool_call) -> dict: name = tool_call.function.name if name not in TOOL_REGISTRY: raise KeyError(f"Unknown tool: {name}") args = json.loads(tool_call.function.arguments) result = TOOL_REGISTRY[name](**args) return { "role": "tool", "tool_call_id": tool_call.id, "content": result, # ← must be a string }
The dispatcher is a lookup table plus three steps: validate the name, parse the JSON arguments, call the function, and return a "role": "tool" message the model can read on the next turn.
The tool_call_id field is mandatory: the API uses it to match each result to the right request when the model calls multiple tools in one turn.
The API raises a validation error: content must be a string, not a dict. You must serialize it first — json.dumps(result) — before assigning it to "content". This is the most common dispatcher bug and it surfaces immediately as a 400 from the API.
import json TOOL_REGISTRY = {"web_search": web_search, "read_page": read_page} def dispatch(tool_call) -> dict: name = tool_call.function.name if name not in TOOL_REGISTRY: raise KeyError(f"Unknown tool: {name}") args = json.loads(tool_call.function.arguments) result = TOOL_REGISTRY[name](**args) return { "role": "tool", "tool_call_id": tool_call.id, "content": # TODO: serialize result so the API accepts it }
Stop — attempt the TODO before revealing. The function already calls the right tool and builds the return dict; your job is the one line that makes content safe for the API regardless of what type result is.
Hint 1: the API requires a string. Hint 2: your tool might return a plain string or a dict — handle both.
result if isinstance(result, str) else json.dumps(result)
Changed line: "content" now guards against non-string returns. A plain string passes through unchanged; anything else is JSON-serialized. This is the crux of the module — the dispatcher is useless if the model receives a Python object it can't read.
Imagine the research agent is asked: "What is the current PageRank of example.com?" The model reasons that it needs fresh data. It emits a tool_call for web_search with {"query": "PageRank example.com"}. Your dispatcher runs the function and returns a "role": "tool" message.
That message is appended to the as an . On the next turn the model reads the observation and decides: is the answer ready, or does it need another tool call?
The dispatcher doesn't decide when to stop — that's the loop's job. It only converts one tool-call request into one tool-result message, cleanly and safely.
With tools defined and the dispatcher wired, the next module builds the . The model reasons, picks a tool, reads the observation, and decides whether to continue or return a final answer.
| Option | Safety / sandboxing | Flexibility for novel tasks | Schema maintenance cost | When to choose | Cost | Complexity |
|---|---|---|---|---|---|---|
| OpenAI-style JSON function schema | High: model can only call declared functions with typed args; arbitrary code execution is impossible | Limited: model is constrained to the declared tool set; can't improvise new operations | Grows with tool count; each new tool needs a hand-written or auto-generated schema | When you have a fixed, known set of tools and need predictable, auditable actions — the right default for most production agents. | Minimal — schema is static JSON; no extra runtime | Low — one dict per tool, validated by the API |
| Code-generation style (smolagents CodeAgent) | Low by default: model writes arbitrary Python; a sandbox is mandatory, not optional | Very high: model can compose any operation expressible in Python, including multi-step logic | Near zero: no per-tool schemas; the model reads docstrings and type hints directly | When tasks are open-ended and the model must compose novel operations at runtime — accept the higher risk in exchange for flexibility. | Higher — sandbox infrastructure adds latency and operational overhead | High — requires a secure sandbox (e.g. E2B) to run generated code safely |
Three failure patterns account for most dispatcher bugs in real agents:
KeyError and the agent loop crashes silently. Fix: always guard with if name not in TOOL_REGISTRY and return a descriptive error message back to the model so it can recover.tool_call.function.arguments. json.loads raises a JSONDecodeError. Wrap the parse in a try/except and return the error string as the tool result — the model can then retry with corrected arguments.content yields a 400 validation error from the API. The symptom is an immediate HTTP error, not a model-level failure — easy to miss if you're not logging raw responses.Verify AI-generated dispatchers by checking: (1) does it guard unknown names? (2) does it wrap json.loads in a try/except? (3) does it serialize content to a string before returning? A dispatcher missing any of these will fail on the first edge case the model hits.
Implement the ReAct loop that drives the research agent: the model reasons, picks a tool, receives an observation, and decides whether to continue or return a final answer. You'll complete the loop by writing the termination condition and the observation-injection step from scratch.
Implements the ReAct loop — the reason-act-observe cycle that drives the research agent across multiple turns until it reaches a final answer.
Why this matters: This is the engine of your agent: without the loop and its termination condition, your tool dispatcher from module 3 can only fire once and stop.
Decision this forces: ReAct (interleaved reason+act) vs. Plan-then-Execute — when each pattern fits the task.
tool_call_response from the model. What two things did it extract from that object to route the call to the right Python function?Answer: the function name (to look up the handler) and the JSON-encoded arguments (to pass as kwargs).
Module 3 left you with a dispatcher that can execute one tool call. But nothing drives it in a loop yet.
This module closes that gap: wiring the dispatcher into a repeating loop where the model reasons, acts, reads the result, and decides whether to keep going or stop.
is a prompting pattern where the model alternates between two moves: a reasoning step (what do I know, what should I do next?) and an action step (call a tool and read the result back).
Each tool result is an — it gets appended to the so the model sees the full chain of thought on the next turn.
The loop exits when the model emits a plain text reply with no tool call — that's the .
A cap is the safety net: it stops an infinite loop if the model never signals done.
web_search("AI agents Python")read_page(url)def run_agent(messages, tools, max_iterations=10): for i in range(max_iterations): response = call_llm(messages, tools) # returns model response choice = response.choices[0] if choice.finish_reason == "stop": # ← termination check return choice.message.content # final answer # tool call path — not wired yet tool_call = choice.message.tool_calls[0] result = dispatch(tool_call) # from module 3 messages.append(...) # ← observation injection missing return "Max iterations reached"
This skeleton shows the two exit paths: a clean "stop" finish reason (model is done) and the iteration cap fallback.
The messages.append(...) line is deliberately empty — that's the observation-injection step you'll complete in Stage 2.
It returns the string "Max iterations reached" after 10 iterations. The model never signals done, so the for-loop exhausts its range and falls through to the final return statement.
def run_agent(messages, tools, max_iterations=10): for i in range(max_iterations): response = call_llm(messages, tools) choice = response.choices[0] if choice.finish_reason == "stop": return choice.message.content tool_call = choice.message.tool_calls[0] messages.append(choice.message) # assistant turn (with tool call) result = dispatch(tool_call) # run the tool messages.append({ # ← observation injected here "role": "tool", "tool_call_id": tool_call.id, "content": str(result), }) return "Max iterations reached"
Two appends keep the message list consistent: first the assistant's tool-call message, then the tool-role observation keyed by tool_call_id. The model sees both on the next turn.
Skipping the first append (the assistant message) causes an API error on the next call — the "tool" role message must follow an assistant message that contains the matching tool call.
7 messages: [system, user, assistant(tool_call_1), tool(result_1), assistant(tool_call_2), tool(result_2), assistant(final_answer)]. Each tool round adds two messages; the final answer adds one.
def run_agent(messages, tools, max_iterations=10): for i in range(max_iterations): response = call_llm(messages, tools) choice = response.choices[0] # TODO: add a second termination check here — # stop if the model returns a tool_call but # the tool name is NOT in the registered tools dict. # Hint 1: tool_call.function.name gives you the name. # Hint 2: raise ValueError with a clear message. if choice.finish_reason == "stop": return choice.message.content tool_call = choice.message.tool_calls[0] messages.append(choice.message) result = dispatch(tool_call) messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": str(result)}) return "Max iterations reached"
Stop — attempt the TODO before revealing the answer. The guard you add is the crux of this exercise: it catches a hallucinated tool name before dispatch() silently fails or raises an opaque KeyError.
Changed lines:
if choice.finish_reason != "stop" and \
tool_call.function.name not in tools:
raise ValueError(f"Unknown tool: {tool_call.function.name}")
Why before finish_reason: if finish_reason is 'tool_calls', you need to validate the name before you try to dispatch it. Placing it after the stop-check means a hallucinated tool name reaches dispatch() unchecked, producing a KeyError with no context about which tool was invented.
| Option | Handles mid-task surprises | Predictable step count | When to choose | Cost | Complexity |
|---|---|---|---|---|---|
| ReAct (interleaved reason + act) | Adapts each turn based on the latest observation | Step count is unknown until the model signals done | Tasks where the next step depends on what the previous tool returned — open-ended research, debugging, multi-hop Q&A. | Pays per turn; more turns = more tokens | Low — one loop, no planner |
| Plan-then-Execute | Plan can go stale if early steps fail or return unexpected data | Steps are fixed by the plan before execution begins | Tasks with a known, stable structure — e.g. 'always fetch three sources then summarise' — where you want a fixed, auditable plan before any tool runs. | One extra LLM call upfront; fewer wasted turns if the plan holds | Higher — needs a planner call + executor loop |
If you append the tool result without first appending the assistant's tool-call message, the API returns an error.
Error: "Invalid parameter: messages with role 'tool' must be a response to a preceding message with 'tool_calls'."
Fix: always append choice.message before the tool result.
A model that keeps calling tools — because the tool always returns partial data — will run forever without a guard.
You won't see an error; you'll see a runaway bill and a hung process.
Each observation grows the . After ~8 tool rounds with large page-read results, you'll hit the limit.
Error: "This model's maximum context length is 128000 tokens. Your messages resulted in N tokens."
The next module's StateManager trims this automatically.
max_iterations is set and the fallback return is reachable, (3) the tool name is validated before dispatch, and (4) the tool_call_id in the tool message matches the one in the assistant message — a mismatch causes a silent wrong-tool attribution.Trace a full run of the research agent against the query "best Python agent frameworks 2024".
web_search.read_page(url) on the top hit.Notice that the at each turn is entirely captured by the message list — there's no hidden memory outside it.
That message list is about to get harder to manage. The next module adds a StateManager class that stores tool results keyed by call ID and trims the list before it overflows the context window.
Add a `StateManager` class to the research agent that maintains the message list, stores tool results keyed by call ID, and trims context when it approaches the model's token limit. You'll revisit the `call_llm()` function from Module 2 and extend it to accept the managed message list.
Builds a StateManager class that appends messages in the correct OpenAI format, stores tool results by call ID, and trims the context window safely without breaking tool-call pairs.
Why this matters: Your research agent will fail or hallucinate on long tasks without bounded, well-formed state — this module makes multi-turn conversations reliable.
Decision this forces: In-memory message list vs. persistent store (Redis/DB) — when to graduate from in-memory state.
The ReAct loop passes a growing message list to the model each turn. Each appends before the next call. That list is the agent's : without it, the model forgets what it tried.
Right now that list lives as a bare Python variable inside the loop. This module wraps it in a StateManager class. The goal: run many turns without exceeding the model's or corrupting message format.
The driving question: how do you keep the message list valid and bounded as it grows?
The OpenAI chat API requires messages in strict sequence: user → assistant → tool → assistant. Every message must pair with the assistant tool_call that requested it.
role: "user" — human input or synthetic prompt from your code.role: "assistant" — model reply, may include tool_calls list.role: "tool" — result keyed by tool_call_id to match the request.If you trim a tool_call but leave its paired tool result — or vice versa — the API returns a 400 error.
This pairing constraint is the core rule your StateManager must enforce when trimming old messages.
Your research agent answers a complex SEO question requiring five tool calls. By turn five, the message list has 30+ messages and approaches the token limit.
Without a StateManager: crash with context-length error or silently truncate and corrupt tool pairs. With one: trim the oldest complete exchange and keep the system prompt and recent N messages.
The StateManager owns three responsibilities: append messages in correct format, store tool results keyed by tool_call_id, and trim context safely when exceeding threshold.
class StateManager: def __init__(self, system_prompt: str, max_messages: int = 20): self.max_messages = max_messages self.messages = [{"role": "system", "content": system_prompt}] def add_user(self, content: str): self.messages.append({"role": "user", "content": content}) def add_assistant(self, message: dict): # message is the raw assistant object from the API response self.messages.append(message)
The constructor pins the system prompt at index 0 so trimming logic can never touch it. add_assistant stores the raw API response object — including any tool_calls list — so the pairing information is preserved exactly as the API returned it.
It receives the full assistant message dict (or object) from the API, which includes the tool_calls list. Storing just the text content would discard the tool_call_id values, making it impossible to pair tool results later — the API would reject the next call with a 400 error.
def add_tool_result(self, tool_call_id: str, content: str): self.messages.append({"role": "tool", "tool_call_id": tool_call_id, "content": content}) if len(self.messages) > self.max_messages: self._trim() def _trim(self): # Keep index 0 (system prompt); drop oldest non-system message # Skip index 1 if assistant with tool_calls — paired tool result must go with it if self.messages[1].get("tool_calls"): del self.messages[1:3] # drop assistant + its tool result together else: del self.messages[1] # safe to drop a lone user/assistant message
Trimming fires after every add_tool_result call, the natural moment when a complete exchange has just landed. The _trim method checks whether the oldest non-system message is an assistant turn with tool calls; if so, it deletes that message and its paired tool result together, keeping the list valid.
The else branch runs: del self.messages[1] removes just that one assistant message. No tool result is paired with it, so deleting it alone leaves the list valid.
# Extends call_llm() from Module 2 to accept a StateManager def call_llm(state: StateManager, tools: list) -> dict: response = client.chat.completions.create( model="gpt-4o", messages=state.messages, # managed list replaces raw variable tools=tools, ) assistant_msg = response.choices[0].message state.add_assistant(assistant_msg.model_dump()) # TODO: also handle tool results return assistant_msg # In the agent loop: # state = StateManager(system_prompt=SYSTEM_PROMPT) # state.add_user(query) # reply = call_llm(state, tools=TOOLS)
This extends the call_llm() signature from Module 2: instead of a raw list, it now accepts a StateManager and reads state.messages directly. The TODO marks where your loop must call state.add_tool_result() after dispatching each tool — completing the pair before the next LLM call.
Replace the TODO with a loop after dispatching:
for tc in assistant_msg.tool_calls:
result = dispatch(tc) # your dispatcher from Module 3
state.add_tool_result(tc.id, result)
Changed lines vs. Stage 2: this is the CALLER side — it feeds tool_call_id and the dispatcher's output into add_tool_result(), completing the pair before the next call_llm() call.
Symptom: openai.BadRequestError: messages[3].tool_call_id does not match any tool_call in messages[2]. Cause: trimming removed the assistant message but left its tool result. Fix: delete the pair together.
If the list exceeds the token limit, the API raises openai.BadRequestError: maximum context length exceeded. Some proxy layers silently truncate the tail, cutting recent tool results. The model then hallucinates.
If you append the raw SDK ChatCompletionMessage object instead of calling .model_dump(): the next API call raises TypeError: Object of type ChatCompletionMessage is not JSON serializable. Always serialize before appending.
| Option | Multi-turn durability | Multi-user / concurrent scale | When to choose | Cost | Complexity |
|---|---|---|---|---|---|
| In-memory list (StateManager) | Lost on restart or crash | One list per process; no sharing | Single-session agents, prototypes, scripts, or batch jobs where the process lives for the duration of one conversation. | Zero infra cost; state lost on process restart. | Low — a plain Python class, no dependencies. |
| Persistent store (Redis / DB) | Survives restarts; resumable | Keyed by session ID; scales horizontally | Production agents that must survive restarts, serve multiple users, or resume interrupted sessions. | Infra cost for the store; adds latency per turn (~1–5 ms for Redis). | Medium — requires serialization, a connection, and a session key strategy. |
Add a max-iterations guard, tool-error recovery, and output validation to the research agent's loop, then audit the full agent against a common-mistakes checklist. You'll build the error-handling wrapper solo and verify it against three failure scenarios: tool crash, infinite loop, and malformed model output.
Adds a max-iterations guard, tool-error recovery, and Pydantic output validation to the research agent loop, then audits the full agent against five common failure modes.
Why this matters: A production agent without these guards will silently spin, crash on tool failures, or return malformed data — this module is the difference between a demo and a deployable agent.
Decision this forces: Fail-fast (raise immediately) vs. error-as-observation (inject error into loop and let the model retry) — which to use per error type.
StateManager that trims the when it approaches the model's token limit. What specific thing does it track to decide when to trim — and what happens to the loop if trimming never fires?Answer: StateManager tracks the running token count against the limit. If trimming never fires, the message list grows until the API rejects the request — the loop crashes with a context-length error.
That crash is exactly the gap this module closes. You now have a working research with state, tools, and a ReAct loop — but no safety net when things go wrong. This module adds three guards: a loop limit, tool-error recovery, and output validation.
Every production needs three : a cap, a tool-error wrapper, and output validation.
MaxStepsExceeded and returns a graceful fallback when the loop hasn't terminated after N steps.The key design question is when to fail fast versus when to recover. Fail fast (raise immediately) for structural problems the model can't fix — like a missing API key. Recover (inject the error as an observation) for transient problems the model can work around — like a tool returning a 429 or an empty result.
class MaxStepsExceeded(Exception): pass FALLBACK = "Agent could not complete the task within the step limit." def run_agent(query: str, max_steps: int = 10) -> str: state.reset(query) for step in range(max_steps): response = call_llm(state.messages()) if response.is_final: return response.content # normal exit state.add_tool_call(response) result = dispatch_tool(response.tool_call) state.add_observation(result) raise MaxStepsExceeded(f"Exceeded {max_steps} steps.") # ← guard fires
The loop counts steps with range(max_steps) instead of while True. If the never fires, MaxStepsExceeded is raised after exactly max_steps iterations — the caller can catch it and serve the fallback string.
A MaxStepsExceeded exception is raised. The caller must catch it — e.g. except MaxStepsExceeded: return FALLBACK — to avoid an unhandled crash. The loop does NOT return the fallback string automatically; raising is the fail-fast signal.
def safe_dispatch(tool_call) -> str: try: return dispatch_tool(tool_call) except Exception as exc: # Inject structured error so the model can retry or reroute return ( f'{{"error": "{type(exc).__name__}", ' f'"message": "{str(exc)[:120]}", ' f'"tool": "{tool_call.name}"}}' ) # In run_agent, replace dispatch_tool(...) with: # result = safe_dispatch(response.tool_call)
Wrapping dispatch_tool in safe_dispatch means a crashing never kills the loop. Instead, the exception becomes a JSON-shaped the model can read and act on — it might retry with different arguments or pick a different tool.
{"error": "TimeoutError", "message": "read timed out", "tool": "<tool_call.name>"} — the model sees this as its next observation and can decide to retry or skip the tool.
from pydantic import BaseModel, ValidationError class AgentResult(BaseModel): answer: str sources: list[str] steps_taken: int def validated_run(query: str, max_steps: int = 10) -> AgentResult: try: raw = run_agent(query, max_steps) # returns dict or str # TODO: parse `raw` into AgentResult and return it # Hint 1: use AgentResult.model_validate(raw) # Hint 2: catch ValidationError and raise ValueError with its str() except MaxStepsExceeded: return AgentResult(answer=FALLBACK, sources=[], steps_taken=max_steps)
Stop — attempt the TODO before revealing. validated_run wraps the full agent: it catches MaxStepsExceeded and returns a graceful fallback, but the output-validation path is yours to complete.
# Changed lines (the TODO block):
result = AgentResult.model_validate(raw) # ← parse + validate
return result
except ValidationError as e:
raise ValueError(f"Agent output invalid: {e}") # ← surface clearly
# Why: model_validate raises ValidationError (not a plain dict error) when
# a required field is missing or the wrong type. Re-raising as ValueError
# gives callers a clean, non-Pydantic exception to handle.
Before trusting any AI-generated version of this agent — or your own — run it against these three failure scenarios and check each guard fires correctly.
Force a tool to raise RuntimeError('API down'). Confirm safe_dispatch returns a JSON error string and the loop continues — it must NOT raise or return None.
Stub call_llm to always return is_final=False. Confirm MaxStepsExceeded is raised after exactly max_steps iterations — count the calls.
Pass a dict missing the sources field to validated_run. Confirm a ValueError is raised with Pydantic's field-level detail — not a generic KeyError.
range(max_steps), not while True? A bare while loop is the most common AI-generated mistake.safe_dispatch catch Exception (broad) or only a narrow subclass? Narrow catches miss unexpected tool failures.MaxStepsExceeded caught at the validated_run boundary and converted to a graceful fallback, not swallowed silently?You've now wired all four components — , , , and — plus three production guards into a single research agent. The capstone challenge asks you to build a variation of this agent from scratch, choosing your own tools and failure strategy: everything you need is now in place.
These are the five failure modes most agents hit in production — and what you actually observe for each.
None for an unrecognised tool name. The model receives an empty observation and often hallucinates a result to fill the gap.context_length_exceeded and the loop crashes.Before scrolling down: from memory, list the six build steps in order and name the one component you'd add first if a working agent started looping forever. Then check your answer against the buildOrder below.
Apply what you learned to SEO/GEO page — the page title is the H1 and the exact search query. OPEN with a 40-60 word self-contained, quotable answer capsule that names the entity explicitly in sentence one (no "in this lesson…" preamble) — this is the sentence an LLM lifts. Name the entity by its full name every time (not pronoun-only). Structure each section as ONE sub-question (How it works · X vs Y · When to use it · Example · Common mistakes), each opening with its own 1-2 sentence direct answer then depth. End with a genuine 3-5 question FAQ phrased as real user questions ending in "?" (these become FAQPage schema + the knowledge check). Prefer quotable specifics — concrete numbers, defaults, version names, one short snippet — over vague prose. Self-contained for a reader who arrived cold from a search engine or an LLM. Intent = tutorial: numbered steps + one short runnable code snippet + a common-mistakes section. Target query: "build ai agents"..
An AI agent built in Python has four core components. Which component is responsible for deciding WHICH tool to call next, based on the current goal and message history?
The LLM is the reasoning engine — it reads the goal and message history and decides what action to take next, including which tool to call. The tool dispatcher only executes the decision the LLM already made; it does not choose the tool. The StateManager stores messages but has no decision-making logic. The termination condition checks whether to stop, not what to do next.
You are building a tool dispatcher. The model returns a tool_calls response naming the tool 'search_web', but your dispatcher's registry only contains 'web_search' and 'calculator'. What is the correct behavior?
A hallucinated tool name is a recoverable error — the right guard is to inject a structured error observation (e.g., 'Tool search_web not found') back into the loop so the model can correct itself on the next turn. Crashing the loop is fail-fast behavior, which is appropriate for unrecoverable errors but not for a hallucinated name the model can self-correct. Fuzzy-matching tool names is unsafe and unpredictable. Silently skipping leaves the model with no feedback and will likely cause a runaway loop.
Read this short Python snippet:
if len(messages) > max_context:
messages = messages[-max_context:]
What critical bug does this trimming strategy introduce in an agent that uses tool calls?
The OpenAI API requires that every tool-call message from the assistant is immediately followed by its paired tool-result message. A naive tail-slice like messages[-max_context:] can cut a tool-call message while keeping its orphaned result, or keep a tool-call with no result — both cause an API error. Losing the system prompt is a separate concern not caused by this slice. Message length and role fields are unaffected by this specific bug.
When would you choose ReAct (interleaved reason + act) over Plan-then-Execute for an agent task?
ReAct shines when each observation (tool result) can change what the agent should do next — the interleaved reason-act-observe cycle lets the agent adapt mid-task. Plan-then-Execute is better when the full plan can be laid out before any tool is called, because it batches reasoning upfront and reduces back-and-forth. Minimizing LLM calls favors Plan-then-Execute, not ReAct. A task with no tools needs neither pattern — a single call_llm() is sufficient.
Explain why using response_format / JSON mode (structured output) is preferred over plain text + regex parsing when the LLM's response needs to trigger a tool call. Name at least two concrete failure modes that regex parsing introduces.
Structured output (response_format or JSON mode) pins the model's output to a schema the dispatcher can parse deterministically. Regex is brittle because LLM outputs are probabilistic — small phrasing changes break patterns. Key failure modes include markdown-fenced JSON that regex skips entirely, and partial or mismatched field names that cause wrong or missing tool dispatches. Structured output eliminates both by making schema conformance a model-level constraint, not a post-hoc parsing problem.