AI Agent Architecture in Practice: Memory, Tool Use, and Failure Modes
There's no shortage of "AI agents will reshape everything" essays. This isn't one of them. Instead, let's look at the concrete architecture decisions you make when building a real agent system, and the failure modes that burn you if you skip them.
The Core Loop: ReAct in Practice
Most agent systems use some variation of the ReAct loop (Reason + Act). The LLM emits a chain of thought → action → observation until it reaches a final answer. Here's a minimal implementation using the Anthropic API:
import anthropic
import json
from typing import Any
client = anthropic.Anthropic()
TOOLS = [
{
"name": "search_codebase",
"description": "Search for a pattern in the codebase. Returns matching file paths and line numbers.",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query or regex pattern"},
"file_pattern": {"type": "string", "description": "Glob pattern like '**/*.py'", "default": "**/*"}
},
"required": ["query"]
}
},
{
"name": "read_file",
"description": "Read a file's contents.",
"input_schema": {
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"]
}
},
{
"name": "run_tests",
"description": "Run the test suite. Returns exit code and stdout.",
"input_schema": {
"type": "object",
"properties": {"test_path": {"type": "string", "default": "tests/"}},
"required": []
}
}
]
def run_agent(task: str, max_iterations: int = 10) -> str:
messages = [{"role": "user", "content": task}]
for i in range(max_iterations):
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=4096,
tools=TOOLS,
messages=messages
)
# Append assistant turn
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason == "end_turn":
# Extract final text
for block in response.content:
if hasattr(block, "text"):
return block.text
return "No result"
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = dispatch_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": str(result)
})
messages.append({"role": "user", "content": tool_results})
return "Max iterations reached"
def dispatch_tool(name: str, inputs: dict) -> Any:
match name:
case "search_codebase":
return search_codebase(inputs["query"], inputs.get("file_pattern", "**/*"))
case "read_file":
return read_file(inputs["path"])
case "run_tests":
return run_tests(inputs.get("test_path", "tests/"))
case _:
return f"Unknown tool: {name}"
This is the skeleton. The real work is in what you put inside dispatch_tool and how you handle failures.
Memory Architecture: What Actually Works
The memory problem is genuinely hard. You need the agent to "remember" relevant context from past runs without stuffing 50k tokens into every request.
Working Memory (In-Context)
Anything in the current context window. Fast, no retrieval cost, but limited and ephemeral. Use structured state tracking:
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class AgentState:
task: str
completed_steps: list[str] = field(default_factory=list)
discovered_facts: list[str] = field(default_factory=list)
current_hypothesis: Optional[str] = None
open_questions: list[str] = field(default_factory=list)
iterations: int = 0
def to_context_block(self) -> str:
lines = [
f"Task: {self.task}",
f"Completed steps: {', '.join(self.completed_steps) or 'none'}",
f"Key facts: {'; '.join(self.discovered_facts) or 'none'}",
]
if self.current_hypothesis:
lines.append(f"Current hypothesis: {self.current_hypothesis}")
if self.open_questions:
lines.append(f"Still need to check: {', '.join(self.open_questions)}")
return "\n".join(lines)
Inject state.to_context_block() into the system prompt on each iteration. This keeps the agent from "forgetting" what it already verified.
Long-Term Memory (Vector Store)
Use for retrieving relevant past interactions or knowledge. With pgvector:
import psycopg2
import numpy as np
from anthropic import Anthropic
client = Anthropic()
def embed(text: str) -> list[float]:
# Use a dedicated embedding model
response = client.embeddings.create(
model="voyage-3",
input=text
)
return response.embeddings[0]
def store_memory(conn, content: str, metadata: dict):
vector = embed(content)
with conn.cursor() as cur:
cur.execute(
"INSERT INTO agent_memory (content, embedding, metadata) VALUES (%s, %s, %s)",
(content, np.array(vector), json.dumps(metadata))
)
conn.commit()
def recall(conn, query: str, limit: int = 5) -> list[str]:
vector = embed(query)
with conn.cursor() as cur:
cur.execute(
"""
SELECT content, 1 - (embedding <=> %s::vector) AS similarity
FROM agent_memory
ORDER BY embedding <=> %s::vector
LIMIT %s
""",
(np.array(vector), np.array(vector), limit)
)
return [row[0] for row in cur.fetchall() if row[1] > 0.75]
Call recall() before each agent run and inject the top results into the system prompt. This is cheap - typically 1-3 database queries.
Tool Design: Where Most Agents Break
The biggest source of agent failures isn't the LLM - it's poorly designed tools. Three patterns that cause failures:
1. Tools That Return Too Much
# Bad: returns 10,000 lines of log
def get_logs() -> str:
return open("/var/log/app.log").read()
# Good: returns relevant slice
def get_logs(since_minutes: int = 10, filter_pattern: str = "ERROR") -> str:
import subprocess
result = subprocess.run(
["journalctl", "-u", "app", f"--since={since_minutes} min ago",
"--grep", filter_pattern, "-n", "50"],
capture_output=True, text=True
)
return result.stdout or "No matching log lines"
The agent will try to reason over everything you give it. Huge tool outputs either cause context overflow or distract it from the actual signal.
2. Tools Without Error Structure
# Bad: raises exception → agent sees a stack trace and panics
def query_database(sql: str) -> list[dict]:
conn = get_conn()
return conn.execute(sql).fetchall() # might throw
# Good: structured error returns
def query_database(sql: str) -> dict:
try:
conn = get_conn()
rows = conn.execute(sql).fetchall()
return {"ok": True, "rows": rows, "count": len(rows)}
except Exception as e:
return {"ok": False, "error": str(e), "hint": "Check SQL syntax and table names"}
Structured errors let the agent retry with a corrected query. Raw exceptions cause it to loop or give up.
3. Tools Without Side-Effect Labels
Mark destructive tools explicitly:
TOOLS = [
{
"name": "run_query",
"description": "Execute a READ-ONLY SQL query. SELECT statements only. Will reject writes.",
...
},
{
"name": "execute_migration",
"description": "⚠️ DESTRUCTIVE: Runs a database migration. Requires explicit confirmation from the user before calling. Do not call unless user has confirmed they want to proceed.",
...
}
]
This phrasing directly in the description reduces accidental destructive actions significantly.
Multi-Agent Coordination: The Simplest Pattern That Works
Don't reach for a full orchestration framework immediately. A simple supervisor/worker pattern handles most cases:
class TaskQueue:
def __init__(self):
self.pending: list[dict] = []
self.results: dict[str, Any] = {}
def add(self, task_id: str, task: dict):
self.pending.append({"id": task_id, **task})
def complete(self, task_id: str, result: Any):
self.results[task_id] = result
def run_supervisor(objective: str) -> dict:
queue = TaskQueue()
# Supervisor agent decomposes the work
decomposition = client.messages.create(
model="claude-haiku-4-5",
max_tokens=1024,
system="You decompose tasks into independent parallel subtasks. Output JSON: {tasks: [{id, description, requires}]}",
messages=[{"role": "user", "content": objective}]
)
plan = json.loads(decomposition.content[0].text)
# Execute tasks in dependency order
for task in plan["tasks"]:
# Block on dependencies
deps_done = all(d in queue.results for d in task.get("requires", []))
if deps_done:
result = run_agent(task["description"])
queue.complete(task["id"], result)
# Supervisor synthesizes
synthesis = client.messages.create(
model="claude-opus-4-5",
max_tokens=2048,
system="Synthesize the subtask results into a final answer.",
messages=[{"role": "user", "content": json.dumps(queue.results)}]
)
return {"result": synthesis.content[0].text, "subtasks": queue.results}
This handles the 80% case. Add a proper queue (Redis, Celery) when you need retry logic, persistence across restarts, or parallel execution.
Testing Agents: What Actually Catches Bugs
Unit testing individual tools is easy. Testing the agent's decision-making is harder. The most useful test type is trajectory testing - assert that the agent visits expected checkpoints:
import pytest
def test_agent_reads_file_before_editing():
"""Agent should always read a file before proposing edits to it."""
task = "Fix the bug on line 42 of src/parser.py"
tool_calls = []
original_dispatch = dispatch_tool
def tracking_dispatch(name, inputs):
tool_calls.append(name)
return original_dispatch(name, inputs)
with patch("your_module.dispatch_tool", tracking_dispatch):
run_agent(task)
read_index = next((i for i, t in enumerate(tool_calls) if t == "read_file"), -1)
edit_index = next((i for i, t in enumerate(tool_calls) if t == "edit_file"), -1)
assert read_index != -1, "Agent never read the file"
assert read_index < edit_index, "Agent edited without reading first"
def test_agent_terminates_within_budget():
"""Agent shouldn't loop indefinitely on a solvable task."""
result = run_agent("What is the Python version in pyproject.toml?", max_iterations=5)
assert result # Got some answer
assert "3." in result # Looks like a version number
These tests run against the real model and are slow, but they catch the failure modes that matter: tool call ordering, infinite loops, off-topic final answers.
You Might Also Like
- Real-time Data Streaming with Apache Kafka
- System Design: Building a Distributed Rate Limiter
- Advanced GraphQL Federation
- Event-Driven Architecture with Kafka
Test Your Knowledge
In the provided `run_agent` function, what is a critical production tradeoff or potential failure mode related to the `max_iterations` parameter?
Free In-Browser Developer Tools
Clean AI CLI logs, build cron expressions, decode JWTs, and calculate chmod permissions offline.
Related Articles
AI Agents in Software Engineering: Architecture Patterns That Actually Work
Moving beyond copilots to autonomous agents: tool-calling loops, MCP integration, memory architectures, multi-agent coordination, and the safety boundaries every engineering team needs to define before deploying agents.
Read more
LangGraph vs DSPy: Declarative Prompt Optimization vs Graph State Machines for AI Agents
LangGraph vs DSPy: contrasting stateful agentic graph workflows with automated prompt optimization and compiled LLM pipeline architectures.
Read more
Hierarchical Multi-Agent Systems in LangGraph: Supervisors, Subgraphs & State Machines
Multi-agent orchestration with LangGraph: hierarchical supervisor routing, shared state schemas, and subagent delegation patterns.
Read more