Module 6: Agent Design Patterns
Design Patterns Lab
Hands-on: refactor a naive agent using registry, middleware, and checkpoint patterns
~15 min readIncludes code
Overview
In Module 3, you built a research assistant agent that searched the web, read pages, and produced structured reports. It worked. It also had every problem we diagnosed in lesson 01 of this module: tools were hardcoded in a dictionary, logging was scattered print() statements, there was no cost tracking, no context management, and no way to resume after a crash. The architecture was right. The engineering was not there yet.
In this lab, you will progressively refactor that same research agent using the four design patterns you learned in lessons 02 through 06. Each refactoring step is independent -- you add one pattern at a time, and after each step the agent still runs correctly but is more maintainable, observable, and resilient.
By the end, you will have transformed a monolithic script into a well-structured agent system. And you will see exactly where the next limitation lies -- which sets the stage for Module 7.
Before and After: The Architecture
Let's start by looking at the transformation we are going to make. The left side is the Module 3 agent. The right side is what we will have at the end of this lab.
graph TD
subgraph Before["Module 3: Monolithic Agent"]
direction TB
B1["Tools defined as dicts<br/>in a global list"]
B2["Tool dispatch via<br/>if/elif or dict lookup"]
B3["print statements<br/>for logging"]
B4["No cost tracking"]
B5["Messages list grows<br/>unbounded"]
B6["Crash = start over"]
B1 --> B2 --> B3 --> B4 --> B5 --> B6
end
subgraph After["Module 6: Pattern-Based Agent"]
direction TB
A1["ToolRegistry<br/>auto-generates schemas"]
A2["Middleware pipeline<br/>logging + cost tracking"]
A3["ContextManager<br/>summarizes old messages"]
A4["CheckpointStore<br/>saves state each iteration"]
A1 --> A2 --> A3 --> A4
end
Before -->|"This lab"| After
style Before fill:#fed7d7,stroke:#fc8181
style After fill:#f0fff4,stroke:#68d391
style B1 fill:#fff5f5,stroke:#feb2b2
style B2 fill:#fff5f5,stroke:#feb2b2
style B3 fill:#fff5f5,stroke:#feb2b2
style B4 fill:#fff5f5,stroke:#feb2b2
style B5 fill:#fff5f5,stroke:#feb2b2
style B6 fill:#fff5f5,stroke:#feb2b2
style A1 fill:#f0fff4,stroke:#9ae6b4
style A2 fill:#f0fff4,stroke:#9ae6b4
style A3 fill:#f0fff4,stroke:#9ae6b4
style A4 fill:#f0fff4,stroke:#9ae6b4
We will apply each pattern in order, testing at each step. This is how real refactoring works -- small, verified changes, never a big-bang rewrite.
The Starting Point
Here is the Module 3 research agent, condensed to the parts we will refactor. If you completed the Module 3 lab, this is your code. If not, use this as the starting point.
"""Module 3 Research Agent -- the monolithic version."""
import json
import datetime
from typing import Any
import anthropic
# --- Tools defined inline ---
def search_web(query: str, max_results: int = 3) -> str:
"""Search the web for information on a topic."""
simulated = [
{
"title": f"Result for: {query}",
"url": f"https://example.com/{query.replace(' ', '-')[:40]}",
"snippet": f"Overview of {query}.",
}
for i in range(max_results)
]
return json.dumps({"results": simulated})
def read_webpage(url: str) -> str:
"""Fetch text content of a webpage."""
content = f"Content from {url}. Key findings about the topic..."
return json.dumps({"url": url, "content": content[:2000]})
def get_current_date() -> str:
"""Return today's date."""
today = datetime.date.today()
return json.dumps({"date": today.isoformat()})
def save_report(title: str, summary: str,
sections: list[dict], sources: list[str]) -> str:
"""Save a structured research report."""
filename = title.lower().replace(" ", "_")[:50] + ".md"
return json.dumps({"filename": filename, "char_count": 1200})
# --- Dispatch table ---
tool_functions: dict[str, Any] = {
"search_web": search_web,
"read_webpage": read_webpage,
"get_current_date": get_current_date,
"save_report": save_report,
}
# --- Tool schemas (hand-written, 80+ lines omitted for brevity) ---
tools = [...] # The full list from Module 3
# --- Agent loop ---
client = anthropic.Anthropic()
def run_agent(question: str) -> str:
messages = [{"role": "user", "content": question}]
for iteration in range(15):
print(f"--- Iteration {iteration + 1} ---") # Ad-hoc logging
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
system="You are a research assistant...",
tools=tools,
messages=messages,
)
if response.stop_reason == "end_turn":
for block in response.content:
if block.type == "text":
return block.text
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" Calling: {block.name}") # Ad-hoc logging
try:
result = tool_functions[block.name](**block.input)
except Exception as e:
result = json.dumps({"error": str(e)})
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({"role": "user", "content": tool_results})
return "Max iterations reached."
`} language="python" filename="research_agent_v0.py" />
Count the problems: tool schemas are hand-written and can drift from implementations, logging is `print()` sprinkled everywhere, there is no cost tracking, the `messages` list grows without bound, and if the process crashes at iteration 10 you lose everything.
## Step 1: Add the Tool Registry
The first refactoring replaces the hand-written tool schemas and dispatch table with a **Tool Registry**. From lesson 02, you know the registry uses Python's `inspect` module to auto-generate JSON schemas from function signatures and docstrings.
<CodeBlock code={`"""Step 1: Replace hand-written schemas with a ToolRegistry."""
import inspect
import json
from typing import Any, Callable, get_type_hints
class ToolRegistry:
"""Central registry for agent tools.
Register a function once. The registry auto-generates the
JSON schema, handles dispatch, and manages tool lifecycle.
"""
def __init__(self):
self._tools: dict[str, Callable] = {}
self._schemas: dict[str, dict] = {}
def register(self, func: Callable) -> Callable:
"""Register a tool function. Can be used as a decorator."""
name = func.__name__
self._tools[name] = func
self._schemas[name] = self._generate_schema(func)
return func
def execute(self, name: str, args: dict) -> str:
"""Execute a registered tool by name."""
if name not in self._tools:
return json.dumps({
"error": f"Unknown tool: {name}",
"available": list(self._tools.keys()),
})
try:
return self._tools[name](**args)
except TypeError as e:
return json.dumps({"error": f"Invalid arguments: {e}"})
except Exception as e:
return json.dumps({"error": f"Tool failed: {e}"})
def get_definitions(self) -> list[dict]:
"""Return tool definitions for the API call."""
return list(self._schemas.values())
def _generate_schema(self, func: Callable) -> dict:
"""Auto-generate a tool schema from function metadata."""
sig = inspect.signature(func)
hints = get_type_hints(func)
doc = inspect.getdoc(func) or ""
properties = {}
required = []
for param_name, param in sig.parameters.items():
prop: dict[str, Any] = {}
hint = hints.get(param_name)
# Map Python types to JSON Schema types
if hint == str:
prop["type"] = "string"
elif hint == int:
prop["type"] = "integer"
elif hint == bool:
prop["type"] = "boolean"
elif hint == float:
prop["type"] = "number"
else:
prop["type"] = "string" # safe fallback
if param.default is inspect.Parameter.empty:
required.append(param_name)
else:
prop["default"] = param.default
properties[param_name] = prop
return {
"name": func.__name__,
"description": doc,
"input_schema": {
"type": "object",
"properties": properties,
"required": required,
},
}
`} language="python" filename="tool_registry.py" />
Now registering tools is a one-line operation. No more hand-written schemas that can drift from the implementation.
<CodeBlock code={`"""Register the research agent's tools with the registry."""
import json
import datetime
registry = ToolRegistry()
@registry.register
def search_web(query: str, max_results: int = 3) -> str:
"""Search the web for information on a topic. Use as the
first step when researching a new question."""
simulated = [
{
"title": f"Result for: {query}",
"url": f"https://example.com/{query.replace(' ', '-')[:40]}",
"snippet": f"Overview of {query}.",
}
for i in range(max_results)
]
return json.dumps({"results": simulated})
@registry.register
def read_webpage(url: str) -> str:
"""Fetch and return the text content of a webpage. Use after
search_web to read promising results."""
content = f"Content from {url}. Key findings about the topic..."
return json.dumps({"url": url, "content": content[:2000]})
@registry.register
def get_current_date() -> str:
"""Get today's date for time-sensitive research queries."""
today = datetime.date.today()
return json.dumps({"date": today.isoformat()})
@registry.register
def save_report(title: str, summary: str,
sections: str, sources: str) -> str:
"""Save a structured research report when you have gathered
enough information to answer the question."""
filename = title.lower().replace(" ", "_")[:50] + ".md"
return json.dumps({"filename": filename, "char_count": 1200})
# The agent loop now uses the registry:
# tools=registry.get_definitions()
# result = registry.execute(block.name, block.input)
`} language="python" filename="register_tools.py" />
What changed: the 80+ lines of hand-written tool schemas are gone. The docstring *is* the description. The function signature *is* the schema. Adding a new tool means writing a function and adding `@registry.register`. Removing a tool means deleting the function. The dispatch table and schema list are always in sync because they are generated from the same source.
## Step 2: Add Middleware for Logging and Cost Tracking
The second refactoring replaces scattered `print()` statements with a **middleware pipeline**. From lesson 03, you know that middleware intercepts every tool call, letting you add cross-cutting behavior in one place.
<CodeBlock code={`"""Step 2: Add middleware pipeline for logging and cost tracking."""
import time
from dataclasses import dataclass, field
@dataclass
class AgentMetrics:
"""Tracks cost and usage across an agent run."""
total_input_tokens: int = 0
total_output_tokens: int = 0
tool_calls: int = 0
tool_errors: int = 0
tool_durations: list[float] = field(default_factory=list)
estimated_cost_usd: float = 0.0
def record_api_call(self, usage):
"""Record token usage from an API response."""
self.total_input_tokens += usage.input_tokens
self.total_output_tokens += usage.output_tokens
# Pricing for claude-sonnet-4-6 (per million tokens)
self.estimated_cost_usd += (
usage.input_tokens * 3.0 / 1_000_000
+ usage.output_tokens * 15.0 / 1_000_000
)
def summary(self) -> str:
avg_ms = (
sum(self.tool_durations) / len(self.tool_durations) * 1000
if self.tool_durations else 0
)
return (
f"Tokens: {self.total_input_tokens:,} in / "
f"{self.total_output_tokens:,} out | "
f"Tools: {self.tool_calls} calls, "
f"{self.tool_errors} errors, "
f"{avg_ms:.0f}ms avg | "
f"Cost: ${self.estimated_cost_usd:.4f}"
)
def logging_middleware(registry: ToolRegistry,
metrics: AgentMetrics):
"""Wrap the registry's execute method with logging."""
original_execute = registry.execute
def wrapped_execute(name: str, args: dict) -> str:
metrics.tool_calls += 1
print(f" [tool] {name}("
f"{json.dumps(args, default=str)[:80]}...)")
start = time.time()
result = original_execute(name, args)
duration = time.time() - start
metrics.tool_durations.append(duration)
# Check for errors in the result
try:
parsed = json.loads(result)
if "error" in parsed:
metrics.tool_errors += 1
print(f" [error] {parsed['error']}")
except (json.JSONDecodeError, TypeError):
pass
print(f" [done] {duration*1000:.0f}ms | "
f"result: {result[:100]}...")
return result
registry.execute = wrapped_execute
`} language="python" filename="middleware.py" />
Notice the pattern: the middleware wraps `registry.execute` without modifying the registry class itself. This is the **decorator pattern** applied at the method level. You can stack multiple middleware -- logging, cost tracking, rate limiting -- and each one is a standalone function.
Here is how the agent loop changes with middleware applied:
<CodeBlock code={`"""Agent loop with registry + middleware."""
import anthropic
client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
SYSTEM_PROMPT = """You are a research assistant. Search the web,
read sources, and produce a structured report.
Process: (1) get today's date, (2) search, (3) read 2-3 pages,
(4) refine search if needed, (5) save report, (6) summarize.
Rules: max 5 pages, max 12 tool calls, do not fabricate."""
def run_agent_v2(question: str) -> str:
"""Agent loop with ToolRegistry and middleware."""
# Set up registry and middleware
metrics = AgentMetrics()
logging_middleware(registry, metrics)
messages = [{"role": "user", "content": question}]
print(f"\\nResearch: {question}")
print("=" * 60)
for iteration in range(15):
print(f"\\n--- Iteration {iteration + 1} ---")
response = client.messages.create(
model=MODEL,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=registry.get_definitions(), # From registry
messages=messages,
)
metrics.record_api_call(response.usage)
if response.stop_reason == "end_turn":
print(f"\\n{metrics.summary()}") # Final metrics
for block in response.content:
if block.type == "text":
return block.text
if response.stop_reason == "tool_use":
messages.append({
"role": "assistant",
"content": response.content,
})
tool_results = []
for block in response.content:
if block.type == "tool_use":
# Middleware handles logging + metrics
result = registry.execute(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({"role": "user", "content": tool_results})
print(f"\\n{metrics.summary()}")
return "Max iterations reached."
`} language="python" filename="agent_v2.py" />
What changed: all logging now flows through one middleware function. The agent loop itself has zero `print()` statements for tool tracking. Adding cost tracking required zero changes to the agent loop -- we just added `metrics.record_api_call()` where we already had the response object. To add rate limiting, you would write another middleware function and apply it -- one line of setup, no loop changes.
## Step 3: Add Context Management
The third refactoring addresses a problem you cannot see in a short demo but that kills long-running agents: **unbounded context growth**. From lesson 04, you know that every iteration adds both an assistant message and tool results to the `messages` list. After enough iterations, you exceed the model's context window and the API call fails.
<CodeBlock code={`"""Step 3: Add context management to prevent window overflow."""
import anthropic
class ContextManager:
"""Manages the conversation history to stay within token limits.
Strategy: when the message list exceeds a threshold, summarize
older messages and replace them with a compact summary. Recent
messages are kept intact so the agent has full detail for its
current step.
"""
def __init__(self, client: anthropic.Anthropic,
model: str,
max_messages: int = 20,
keep_recent: int = 6):
self.client = client
self.model = model
self.max_messages = max_messages
self.keep_recent = keep_recent
def maybe_compress(self, messages: list[dict]) -> list[dict]:
"""Compress the message history if it exceeds the limit.
Returns a new message list with old messages summarized.
"""
if len(messages) <= self.max_messages:
return messages # No compression needed
# Split into old messages and recent messages
boundary = len(messages) - self.keep_recent
old_messages = messages[:boundary]
recent_messages = messages[boundary:]
# Summarize the old messages
summary = self._summarize(old_messages)
# Build new message list: summary + recent
compressed = [
{
"role": "user",
"content": (
f"[Context summary of previous {len(old_messages)} "
f"messages]\\n\\n{summary}\\n\\n"
"[End of summary. Continue from here.]"
),
},
{
"role": "assistant",
"content": "Understood. I will continue the research "
"using the context above.",
},
] + recent_messages
print(f" [context] Compressed {len(messages)} messages "
f"-> {len(compressed)} messages")
return compressed
def _summarize(self, messages: list[dict]) -> str:
"""Use the LLM to summarize older conversation history."""
# Build a text representation of the old messages
parts = []
for msg in messages:
role = msg["role"]
content = msg.get("content", "")
if isinstance(content, str):
parts.append(f"{role}: {content[:300]}")
elif isinstance(content, list):
for block in content:
if isinstance(block, dict):
if block.get("type") == "tool_result":
parts.append(
f"tool_result: "
f"{str(block.get('content', ''))[:200]}"
)
elif hasattr(block, "type"):
if block.type == "text":
parts.append(f"assistant: {block.text[:200]}")
elif block.type == "tool_use":
parts.append(
f"tool_call: {block.name}"
f"({json.dumps(block.input)[:100]})"
)
history_text = "\\n".join(parts)
response = self.client.messages.create(
model=self.model,
max_tokens=500,
messages=[{
"role": "user",
"content": (
"Summarize this agent conversation history. "
"Include: the research question, tools called, "
"key findings so far, and sources found. Be "
"concise but preserve all factual details.\\n\\n"
f"{history_text}"
),
}],
)
return response.content[0].text
`} language="python" filename="context_manager.py" />
Integrating this into the agent loop requires exactly two lines:
<CodeBlock code={`# In the agent loop, after creating the client:
context_mgr = ContextManager(client, MODEL, max_messages=20)
# Inside the loop, before each API call:
messages = context_mgr.maybe_compress(messages)
`} language="python" filename="integration_context.py" />
What changed: the `messages` list can no longer grow without bound. When it hits 20 messages, the older ones are summarized and replaced. The agent retains full detail for its recent actions and a compact summary of everything before. The summarization costs one extra API call, but that is far cheaper than failing because you exceeded the context window -- or worse, degrading response quality because the model is drowning in old, irrelevant tool results.
## Step 4: Add Checkpointing
The fourth and final refactoring makes the agent **resumable**. From lesson 06, you know that checkpointing saves the agent's state after each meaningful step, so a crash at iteration 10 does not mean starting over from iteration 1.
<CodeBlock code={`"""Step 4: Add checkpointing for crash recovery."""
import json
import os
import hashlib
from dataclasses import dataclass, asdict
@dataclass
class Checkpoint:
"""Snapshot of agent state at a given iteration."""
question: str
iteration: int
messages_json: str # Serialized message history
metrics_snapshot: dict
status: str # "running", "completed", "failed"
def save(self, directory: str = ".checkpoints") -> str:
"""Save checkpoint to disk."""
os.makedirs(directory, exist_ok=True)
# Deterministic filename from the question
q_hash = hashlib.sha256(
self.question.encode()
).hexdigest()[:12]
path = os.path.join(
directory,
f"research_{q_hash}_iter{self.iteration:03d}.json"
)
with open(path, "w") as f:
json.dump(asdict(self), f, indent=2, default=str)
print(f" [checkpoint] Saved iteration {self.iteration} "
f"-> {path}")
return path
@classmethod
def load_latest(cls, question: str,
directory: str = ".checkpoints"
) -> "Checkpoint | None":
"""Load the most recent checkpoint for a question."""
if not os.path.exists(directory):
return None
q_hash = hashlib.sha256(
question.encode()
).hexdigest()[:12]
prefix = f"research_{q_hash}_iter"
# Find all checkpoints for this question
candidates = sorted([
f for f in os.listdir(directory)
if f.startswith(prefix) and f.endswith(".json")
])
if not candidates:
return None
# Load the latest one
path = os.path.join(directory, candidates[-1])
with open(path) as f:
data = json.load(f)
print(f" [checkpoint] Resuming from iteration "
f"{data['iteration']} ({path})")
return cls(**data)
def serialize_messages(messages: list[dict]) -> str:
"""Serialize message history to JSON for checkpointing.
Handles Anthropic content blocks by converting them to dicts.
"""
serializable = []
for msg in messages:
content = msg.get("content", "")
if isinstance(content, list):
# Convert content blocks to serializable dicts
blocks = []
for block in content:
if hasattr(block, "type"):
# Anthropic ContentBlock object
if block.type == "text":
blocks.append({
"type": "text", "text": block.text
})
elif block.type == "tool_use":
blocks.append({
"type": "tool_use",
"id": block.id,
"name": block.name,
"input": block.input,
})
else:
blocks.append(block)
serializable.append({
"role": msg["role"], "content": blocks
})
else:
serializable.append(msg)
return json.dumps(serializable, default=str)
`} language="python" filename="checkpointing.py" />
## The Complete Refactored Agent
Here is the full agent loop with all four patterns applied. Compare its structure to the Module 3 version -- every cross-cutting concern has its own home.
<CodeBlock code={`"""Research Agent v3 -- fully refactored with design patterns.
Patterns applied:
1. ToolRegistry -- dynamic tool management
2. Middleware -- logging and cost tracking
3. ContextManager -- prevents context overflow
4. Checkpointing -- crash recovery
Usage:
python research_agent_v3.py
# If it crashes, run again -- it resumes from checkpoint.
"""
import json
import anthropic
# --- Configuration ---
MODEL = "claude-sonnet-4-6"
MAX_ITERATIONS = 15
SYSTEM_PROMPT = """You are a research assistant. Search the web,
read sources, and produce a structured report.
Process: (1) get date, (2) search, (3) read 2-3 pages,
(4) refine if needed, (5) save report, (6) summarize.
Rules: max 5 pages, max 12 tool calls, never fabricate."""
client = anthropic.Anthropic()
def run_agent_v3(question: str) -> str:
"""Production-ready agent loop with all four patterns."""
# --- Pattern 1: Tool Registry ---
# (tools registered via @registry.register above)
tool_defs = registry.get_definitions()
# --- Pattern 2: Middleware ---
metrics = AgentMetrics()
logging_middleware(registry, metrics)
# --- Pattern 3: Context Manager ---
context_mgr = ContextManager(
client, MODEL, max_messages=20, keep_recent=6
)
# --- Pattern 4: Checkpoint Recovery ---
checkpoint = Checkpoint.load_latest(question)
if checkpoint and checkpoint.status == "running":
messages = json.loads(checkpoint.messages_json)
start_iteration = checkpoint.iteration + 1
print(f"Resuming from iteration {start_iteration}")
else:
messages = [{"role": "user", "content": question}]
start_iteration = 1
print(f"\\nResearch: {question}")
print("=" * 60)
for iteration in range(start_iteration, MAX_ITERATIONS + 1):
print(f"\\n--- Iteration {iteration} ---")
# Context management: compress if needed
messages = context_mgr.maybe_compress(messages)
# LLM call
response = client.messages.create(
model=MODEL,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=tool_defs,
messages=messages,
)
metrics.record_api_call(response.usage)
# Terminal condition
if response.stop_reason == "end_turn":
Checkpoint(
question=question,
iteration=iteration,
messages_json=serialize_messages(messages),
metrics_snapshot={"summary": metrics.summary()},
status="completed",
).save()
print(f"\\n{metrics.summary()}")
for block in response.content:
if block.type == "text":
return block.text
# Tool execution through registry + middleware
if response.stop_reason == "tool_use":
messages.append({
"role": "assistant",
"content": response.content,
})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = registry.execute(
block.name, block.input
)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({
"role": "user", "content": tool_results
})
# Checkpoint after each iteration
Checkpoint(
question=question,
iteration=iteration,
messages_json=serialize_messages(messages),
metrics_snapshot={"summary": metrics.summary()},
status="running",
).save()
print(f"\\n{metrics.summary()}")
return "Max iterations reached."
if __name__ == "__main__":
answer = run_agent_v3(
"What are the main approaches to reducing "
"hallucinations in large language models?"
)
print(f"\\nFinal answer: {answer[:300]}...")
`} language="python" filename="research_agent_v3.py" />
## What Each Pattern Bought You
Let's trace how the refactored agent handles the same change requests that broke the monolithic version in lesson 01:
**"Add a new tool."** Write a function, add `@registry.register`. The schema is auto-generated. The middleware already intercepts it. The checkpoint serializer already handles it. One line of code, zero changes elsewhere.
**"Add logging to every tool call."** Already done. The logging middleware intercepts every call through `registry.execute`. To change the log format, edit one function.
**"Track costs."** Already done. The `AgentMetrics` dataclass accumulates token usage from every API call. The `metrics.summary()` prints totals at the end.
**"Handle long conversations."** Already done. The `ContextManager` compresses old messages before they overflow the context window. The agent never crashes from token limits.
**"Resume after a crash."** Already done. Run the same command again. The `Checkpoint.load_latest()` finds the last saved state and continues from there. No work is lost.
Each pattern solved exactly one concern. Together, they transformed a fragile script into a system that can evolve.
## The Interaction Between Patterns
These four patterns are not just independent additions -- they **reinforce each other**. Here is how they interact at runtime:
<MermaidDiagram chart={`sequenceDiagram
participant Loop as Agent Loop
participant Ctx as ContextManager
participant LLM as LLM API
participant MW as Middleware
participant Reg as ToolRegistry
participant Chk as Checkpoint
Note over Loop: Start iteration N
Loop->>Ctx: maybe_compress(messages)
Ctx-->>Loop: compressed messages
Loop->>LLM: messages.create(tools, messages)
LLM-->>Loop: response (tool_use)
Note over Loop: Record token usage in metrics
loop For each tool call
Loop->>MW: execute(name, args)
MW->>MW: Log call, start timer
MW->>Reg: original_execute(name, args)
Reg-->>MW: result
MW->>MW: Log result, record duration
MW-->>Loop: result
end
Loop->>Chk: save(iteration, messages, metrics)
Note over Loop: Next iteration
The sequence matters: context management happens before the LLM call (so the model sees a clean, right-sized history), middleware wraps around tool execution (so every call is logged and timed), and checkpointing happens after tool results are collected (so the saved state includes the latest results). This ordering is not accidental -- it follows the data flow of the agent loop.
What Is Still Missing
The refactored agent is well-structured, observable, and resilient. But run it twice with the same question and notice something: the second run starts from scratch. It does not remember that it already researched this topic yesterday. It cannot build on previous findings. It cannot learn from past mistakes.
This is not a pattern problem -- it is a memory problem. Our agent has:
- Working memory (the
messageslist) -- but it is scoped to a single run and gets summarized away - Checkpoint state (the
.checkpoints/directory) -- but that is for crash recovery, not knowledge retention - No long-term memory -- no way to store and retrieve knowledge across sessions
Consider what a memory-equipped research agent could do:
- "I researched LLM hallucinations last week. Let me check what I already know before searching again."
- "The last time I used that URL, the content was outdated. Let me try a different source."
- "The user asked about RAG three times this month. They probably care about this topic -- I should track new developments."
This is the gap between a well-engineered tool and a truly intelligent assistant. Design patterns gave us the structure. Memory gives us the continuity.
Key insight: Design patterns make your agent's code maintainable and resilient. Memory makes your agent's behavior intelligent across sessions. You need both.
Summary
You refactored the Module 3 research agent in four steps, each applying one design pattern from this module:
- Tool Registry replaced hand-written schemas and dispatch tables with auto-generated definitions from function metadata. Adding or removing a tool is now a one-line change.
- Middleware replaced scattered
print()statements with a centralized interception layer that handles logging and cost tracking. Cross-cutting concerns live in one place. - Context Management prevents the message list from growing unbounded by summarizing older messages when the conversation exceeds a threshold. Long-running agents no longer crash from context overflow.
- Checkpointing saves agent state after each iteration, enabling crash recovery by resuming from the last saved checkpoint. No work is lost.
These patterns are composable -- each one solves a single concern, and they reinforce each other when combined. The Tool Registry gives middleware a single interception point. The Context Manager keeps checkpoints small. The middleware metrics tell you whether context compression is being triggered too often.
The refactored agent is maintainable, observable, and resilient. But it forgets everything between sessions. In Module 7: Memory and Knowledge Management, you will give your agent the ability to store, retrieve, and reason over information across runs -- turning a well-structured tool into a system that learns.