Originally published on tamiz.pro.
Large Language Models (LLMs) have revolutionized software development, but when we stack them into multi-step agents, a fundamental architectural flaw emerges: the Context Window. Unlike human engineers who maintain an immutable memory of requirements and state, AI agents often suffer from "context drift"—losing track of instructions or hallucinating facts as the conversation history grows.
This is the Memory Bottleneck. It is not merely a token limit issue; it is a systemic failure in how agents manage state over time. In this deep dive, we will dissect why standard ReAct loops fail under memory pressure and how Self-Driving Tooling—architectures that autonomously manage tools, memory, and execution without constant human intervention—solves this problem.
1. The Anatomy of Agent Failure
To understand the fix, we must first diagnose the disease. An AI agent typically operates in a loop:
- Observe: Receive user input and conversation history.
- Think: The LLM analyzes the context to decide the next action.
-
Act: The agent calls a tool (e.g.,
search_web,execute_code). - Observe: The tool's output is appended to the context window.
As the agent executes more steps, the context window fills up. Modern LLMs (like GPT-4o or Claude 3.5 Sonnet) have large windows (128k-200k tokens), but larger windows do not equal better memory. They equal attention dilution.
The Attention Sink Problem
LLMs are probabilistic engines. As context grows, the probability mass spreads thinner across irrelevant tokens. This leads to two common failure modes:
- Lost in the Middle: Key instructions provided at the start of a long conversation are forgotten or deprioritized as new tool outputs arrive.
- Tool Call Collapse: The model starts repeating previous tool calls because it "forgot" the result or the decision logic, leading to infinite loops or redundant work.
A classic example is a data analysis agent. After fetching five datasets and performing three aggregations, the model might forget the initial definition of "revenue" provided in step one, leading to incorrect final conclusions. The agent has the data, but it has lost the state.
2. From Reactive to Autonomous: The Self-Driving Shift
The term "self-driving tooling" draws an analogy from autonomous vehicles. In a reactive agent, the "driver" (the LLM) looks at the current frame (context) and decides whether to steer left (call tool A) or right (call tool B). If the frame is cluttered (memory bottleneck), the driver crashes.
In a self-driving architecture, the system includes its own sensors and navigation systems that function independently of the driver’s immediate perception. This translates to:
- Externalized Memory: A persistent storage layer (vector DB, SQL, or JSON file) that survives beyond the context window.
- Autonomous Orchestration: A control plane that manages tool execution, error recovery, and state transitions without relying solely on the LLM’s working memory.
- Tooling as Infrastructure: Tools are not just functions to be called; they are managed resources with lifecycle states.
3. Core Pillars of Self-Driving Tooling
How do we build this? We need three technical components.
A. Semantic Memory (RAG for State)
Instead of stuffing the entire conversation history into the context window, we extract critical state into a semantic memory store.
- Working Memory: The active context (last N turns).
- Episodic Memory: A vector database storing past interactions, tool results, and decisions, indexed by embeddings.
When the agent needs to recall a decision made 50 steps ago, it doesn’t read the history. It queries the vector DB for the relevant embedding and injects only the summary back into the context.
B. The Memory Manager Middleware
This is the "self-driving" brain. It sits between the LLM and the tools. Its responsibilities include:
- Summarization: Automatically summarizing older parts of the conversation and replacing them with concise summaries.
- Relevance Filtering: Deciding which past tool outputs are still relevant to the current query.
- Context Packing: Constructing the optimal prompt by combining current input, working memory, and retrieved episodic memory.
C. Tool State Machines
Tools in self-driving systems are not stateless. A deploy_to_prod tool, for example, should maintain a state (e.g., pending, deploying, success, failed). The agent can query the state of a tool execution without needing to re-run it or remember the full log in its context.
4. Architecture: The Self-Driving Agent Loop
Let’s look at a conceptual implementation of a self-driving agent using Python-like pseudocode. We’ll use a pattern that separates the Controller (orchestrator) from the Actor (LLM).
class SelfDrivingAgent:
def __init__(self, llm, memory_store, tool_registry):
self.llm = llm
self.memory = memory_store # Vector DB or SQL
self.tools = tool_registry
self.context_buffer = []
def run(self, user_query):
# 1. Retrieve Relevant History
# Instead of sending full history, we query memory for relevant past events
relevant_context = self.memory.query(user_query, top_k=3)
# 2. Build Prompt
prompt = self._construct_prompt(user_query, relevant_context)
# 3. LLM Decision
decision = self.llm.generate(prompt)
# 4. Tool Execution & State Tracking
if decision.action == "call_tool":
tool_result = self.tools.execute(decision.tool_name, decision.params)
# 5. Memory Commitment
# Store the outcome semantically, not just textually
self.memory.commit(
event_type="tool_execution",
tool=decision.tool_name,
result_summary=summary(tool_result),
embedding=generate_embedding(f"{decision.tool_name}: {tool_result}")
)
# 6. Recursive Step or Final Answer
return self._handle_result(decision, tool_result)
return decision.final_answer
Key Differences from Standard Agents:
- Query, Don’t Dump: We query memory for relevance rather than dumping the whole history.
- Structured Commitment: Tool results are summarized and embedded before storage, keeping the context window lean.
- State Awareness: The agent knows the state of previous tool executions via the memory store.
5. Practical Implementation with LangGraph and Memory
For production-grade self-driving agents, frameworks like LangGraph (by LangChain) provide the infrastructure to manage stateful, multi-agent workflows. LangGraph allows you to define nodes (tools/LLMs) and edges (transitions) with a central state object.
Here’s how you might implement a memory-augmented tool call in LangGraph:
from langgraph.graph import StateGraph, END
from typing import TypedDict
import uuid
class AgentState(TypedDict):
messages: list # Current conversation
memory: list # Retrieved relevant history
tool_results: dict # Cached tool results
# Define a memory retrieval node
def retrieve_memory(state: AgentState) -> AgentState:
query = state['messages'][-1].content
# Query vector store
relevant_docs = vector_store.similarity_search(query, k=2)
state['memory'] = relevant_docs
return state
# Define a tool-calling node
def call_tool(state: AgentState) -> AgentState:
# LLM decides to call a tool
# ... tool execution logic ...
# Store result in tool_results for future reference
state['tool_results'][tool_name] = result
return state
# Build the graph with memory-aware transitions
workflow = StateGraph(AgentState)
workflow.add_node("retrieve_memory", retrieve_memory)
workflow.add_node("agent", "llm_node)
workflow.add_node("tool", call_tool)
workflow.set_entry_point("retrieve_memory")
workflow.add_edge("retrieve_memory", "agent")
workflow.add_conditional_edges(
"agent",
lambda x: "tool" if x.get("needs_tool") else END,
{"tool": "tool", "END": END}
)
workflow.add_edge("tool", "agent")
app = workflow.compile()
Why This Fixes the Bottleneck:
- Memory is Explicit: The
memorykey in the state is populated from an external source, not just accumulated history. - Tool Results are Cached: The
tool_resultsdict acts as a short-term cache, preventing the LLM from needing to remember raw outputs from previous turns. - Control Flow is Deterministic: The graph structure ensures that memory retrieval happens before the LLM makes decisions, reducing the chance of context drift.
6. Advanced: Self-Reflective Tooling
The most robust self-driving agents incorporate self-reflection. After a tool execution, the agent should evaluate:
- Was the result successful?
- Did I misunderstand the tool’s purpose?
- Do I need to store this outcome for future reference?
This meta-cognitive step can be implemented as a separate node in the graph that critiques the tool’s output and updates the memory store accordingly.
def self_reflect(state: AgentState) -> AgentState:
tool_output = state['tool_results']
reflection_prompt = f"Evaluate the success of this tool call: {tool_output}. Summarize key findings for future memory."
reflection = llm.generate(reflection_prompt)
state['memory'].append({
"type": "reflection",
"content": reflection,
"timestamp": datetime.now()
})
return state
This reflection becomes part of the semantic memory, allowing the agent to "learn" from past tool interactions.
7. Best Practices for Production
- Tiered Memory: Use hot memory (Redis) for recent interactions and cold memory (PostgreSQL + Vector Embeddings) for long-term storage.
- Summarization Policies: Implement aggressive summarization for older messages. Replace every 10th message with a summary after a certain token threshold.
- Tool Versioning: As your toolset evolves, ensure the agent’s memory of tool schemas is updated. Stale tool definitions lead to execution failures.
- Failure Recovery: Self-driving systems must handle tool failures gracefully. If a tool fails, the agent should retry with modified parameters or escalate to human intervention, rather than looping indefinitely.
8. Conclusion
The memory bottleneck is the primary reason AI agents fail in complex, multi-step tasks. By shifting from a reactive, history-dumping model to a self-driving tooling architecture—where memory is externalized, tool states are managed, and orchestration is autonomous—we can build agents that are not just smart, but reliable.
This approach mirrors how senior engineers work: they don’t memorize every line of code they’ve ever written; they use documentation (memory), standardized processes (tooling), and clear architectural patterns (orchestration) to solve problems regardless of complexity.
For more insights on building production-grade AI agents, check out Tamiz's Insights on AI system architecture.
Frequently Asked Questions
Q: What is the difference between RAG and Semantic Memory in agents?
A: RAG (Retrieval-Augmented Generation) typically retrieves external knowledge (documents, web pages) to answer questions. Semantic Memory in self-driving agents retrieves internal state (past tool calls, decisions, outcomes) to maintain continuity across a multi-step task.
Q: How much context do I really need?
A: Aim for the minimum viable context. For a 200k token model, you might think you don’t need optimization. However, attention dilution is real. Keeping the active context under 10k tokens by offloading the rest to memory often yields better accuracy than feeding the entire history.
Q: Can I use this with any LLM?
A: Yes. The self-driving architecture is framework-agnostic. Whether you’re using OpenAI, Anthropic, or open-source models like Llama 3, the pattern of external memory and autonomous orchestration applies equally."
Let's wrap this up with a few more questions, then move into the practical implementation.
Q: Won't external memory be slow?
A: Modern vector databases like Chroma, Milvus, or Weaviate return results in single-digit milliseconds for queries under 100K vectors. The latency penalty is negligible compared to the seconds your LLM spends generating each turn. If you're hitting slowness, it's usually an indexing problem, not a retrieval one.
Q: How do I prevent the agent from looping forever?
A: Implement three safeguards: (1) a maximum step budget per task, (2) a deduplication check on tool calls so the same action isn't repeated, and (3) a reflection step where the agent evaluates whether its last action made progress toward the goal. If no progress is detected, the orchestrator triggers a re-planning pass with updated context.
Q: What about cost?
A: Externalizing memory shifts cost from repeated context inflation to one-time embedding and indexing. For a typical agent session, you'll spend more on LLM calls than on memory operations. The key optimization is selective recall—only fetching the memories relevant to the current sub-goal, not dumping the entire knowledge base into every prompt.
Building the Self-Driving Agent
Enough theory. Let's build it.
We'll construct a minimal but complete implementation using Python, with three layers: tool registry, memory layer, and orchestration loop.
Layer 1: Tool Registry
Tools are the agent's hands. Every capability must be declaratively registered so the orchestrator can reason about them.
# tools.py
from dataclasses import dataclass
from typing import Any, Callable
@dataclass
class ToolSpec:
name: str
description: "str"
parameters: dict # JSON Schema
fn: Callable[..., Any]
def to_openai_format(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
class ToolRegistry:
def __init__(self):
self._tools: dict[str, ToolSpec] = {}
def register(self, tool: ToolSpec):
self._tools[tool.name] = tool
def get(self, name: str) -> ToolSpec:
if name not in self._tools:
raise KeyError(f"Tool '{name}' not found")
return self._tools[name]
def list(self) -> list[ToolSpec]:
return list(self._tools.values())
# Example tools
def read_file(path: str) -> str:
with open(path) as f:
return f.read()
def write_file(path: str, content: str) -> str:
with open(path, "w") as f:
f.write(content)
return f"Wrote {len(content)} chars to {path}"
def search_web(query: str, max_results: int = 5) -> list[dict]:
# In production, integrate with a search API
return [{"title": query, "snippet": f"Result for {query}"}] * max_results
registry = ToolRegistry()
registry.register(ToolSpec(
name="read_file",
description="Read the contents of a file from disk",
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "Absolute or relative file path"},
},
"required": ["path"],
},
fn=read_file,
))
registry.register(ToolSpec(
name="write_file",
description="Write content to a file on disk",
parameters={
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"},
},
"required": ["path", "content"],
},
fn=write_file,
))
registry.register(ToolSpec(
name="search_web",
description="Search the web for information",
parameters={
"type": "object",
"properties": {
"query": {"type": "string"},
"max_results": {"type": "integer", "default": 5},
},
"required": ["query"],
},
fn=search_web,
))
Layer 2: Episodic Memory
This is where we solve the memory bottleneck. Every observation, tool result, and decision becomes a structured memory with semantic embedding.
# memory.py
import hashlib
import json
import numpy as np
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional
@dataclass
class Memory:
id: str
type: str # "observation" | "decision" | "tool_result" | "reflection"
content: str
context: Optional[str]
timestamp: str
embedding: Optional[list[float]] = None
importance: float = 1.0
def to_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, d: dict) -> "Memory":
d = d.copy()
return cls(**d)
class VectorMemoryStore:
"""Simple in-memory vector store using cosine similarity."""
def __init__(self, embed_fn=None):
self.memories: list[Memory] = []
self.embed_fn = embed_fn or self._noop_embed
def _noop_embed(self, text: str) -> list[float]:
"""Deterministic placeholder embedding. Replace with a real model."""
h = int(hashlib.md5(text.encode()).hexdigest(), 16)
return [(h >> (i * 8)) & 0xFF for i in range(16)]
def add(self, memory: Memory):
if self.embed_fn and not memory.embedding:
memory.embedding = self.embed_fn(memory.content)
self.memories.append(memory)
def recall(self, query: str, k: int = 5) -> list[Memory]:
query_emb = self.embed_fn(query)
scored = []
for m in self.memories:
if not m.embedding:
continue
sim = self._cosine(query_emb, m.embedding) * m.importance
scored.append((sim, m))
scored.sort(reverse=True, key=lambda x: x[0])
return [m for _, m in scored[:k]]
def _cosine(self, a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
na = (sum(x * x for x in a)) ** 0.5
nb = (sum(x * x for x in b)) ** 0.5
return dot / (na * nb) if na and nb else 0.0
def clear(self):
self.memories = []
def stats(self) -> dict:
types = {}
for m in self.memories:
types[m.type] = types.get(m.type, 0) + 1
return {
"total_memories": len(self.memories),
"by_type": types,
}
Layer 3: The Orchestration Loop
This is the core—where autonomous decision-making happens. The orchestrator runs a loop: observe → plan → act → reflect → store.
# orchestrator.py
import json
from typing import Optional
from tools import ToolRegistry
from memory import VectorMemoryStore, Memory
class AgentOrchestrator:
MAX_STEPS = 20
PROGRESS_THRESHOLD = 0.1 # minimum semantic similarity to prior state
def __init__(
self,
llm_client,
model: str,
registry: ToolRegistry,
memory: VectorMemoryStore,
system_prompt: str = "",
):
self.llm = llm_client
self.model = model
self.registry = registry
self.memory = memory
self.system_prompt = system_prompt or self._default_system_prompt()
self.step_count = 0
self.task_history: list[dict] = []
def _default_system_prompt(self) -> str:
return """You are an autonomous AI agent. Your goal is to accomplish tasks by reasoning,
planning, and using tools. Think carefully before acting. Learn from observations and
build on past experiences stored in your memory. When unsure, search before guessing.
Keep your responses concise and action-oriented."""
def run(self, goal: str, context: str = "") -> dict:
"""Execute a goal autonomously. Returns execution trace."""
self.step_count = 0
self.task_history = []
# Store the initial goal as a memory
self.memory.add(Memory(
id=self._mkid("goal"),
type="observation",
content=goal,
context=context,
timestamp=datetime.now().isoformat(),
importance=2.0,
))
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"Goal: {goal}\n{f'Context: {context}' if context else ''}"},
]
trace = {"goal": goal, "steps": [], "final_output": None}
while self.step_count < self.MAX_STEPS:
self.step_count += 1
step = self._execute_step(messages, trace)
trace["steps"].append(step)
if step["type"] == "success":
trace["final_output"] = step["content"]
break
if step["type"] == "blocked":
trace["final_output"] = step.get("reason", "Agent could not complete the task.")
break
return trace
def _execute_step(self, messages: list, trace: dict) -> dict:
"""Single orchestration step: recall → decide → act → reflect."""
# 1. Recall relevant memories
relevant = self.memory.recall(messages[-1]["content"], k=3)
memory_context = ""
if relevant:
recalled = "\n".join(f"[{m.type}] {m.content}" for m in relevant)
memory_context = f"\nRelevant past experience:\n{recalled}"
# Add recalled memories as system context for this step
messages.append({
"role": "system",
"content": f"Recalled context:{memory_context}",
})
# 2. Get LLM decision
response = self.llm.chat(self.model, messages)
thought = response.get("content", "")
tool_calls = response.get("tool_calls", [])
# 3. Execute tool calls if any
if tool_calls:
results = []
for tc in tool_calls:
tool_name = tc["function"]["name"]
args = json.loads(tc["function"]["arguments"])
try:
tool = self.registry.get(tool_name)
result = tool.fn(**args)
status = "success"
except Exception as e:
result = f"Error: {e}"
status = "error"
results.append({"tool": tool_name, "result": result, "status": status})
# Store tool interaction as memory
self.memory.add(Memory(
id=self._mkid(f"{tool_name}-{args}"),
type="tool_result",
content=str(result),
context=f"Called {tool_name}({args})",
timestamp=datetime.now().isoformat(),
))
# Feed results back to LLM
for r in results:
messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"content": r["result"],
})
# Get final response after tool execution
response = self.llm.chat(self.model, messages)
thought = response.get("content", "")
return {
"type": "action",
"step": self.step_count,
"thought": thought,
"actions": results,
"output": thought,
}
# No tool calls — agent has produced a final answer
if self._has_progress(messages):
return {
"type": "success",
"step": self.step_count,
"thought": thought,
"output": thought,
}
return {
"type": "blocked",
"step": self.step_count,
"thought": thought,
"reason": "No progress detected and no tool calls made.",
}
def _has_progress(self, messages: list) -> bool:
"""Heuristic: check if latest message is meaningfully different."""
if len(messages) < 2:
return False
last = messages[-1].get("content", "")
if len(last) < 20:
return False
return True
def _mkid(self, content: str) -> str:
return hashlib.sha256(content.encode()).hexdigest()[:12]
def get_memory_stats(self) -> dict:
return self.memory.stats()
Wiring It Together
Here's how you'd run the full system end-to-end:
# main.py
import json
from tools import ToolRegistry, registry
from memory import VectorMemoryStore
from orchestrator import AgentOrchestrator
# Minimal mock LLM client — swap with your actual provider
class MockLLMClient:
"""Replace this with OpenAI, Anthropic, or any chat-compatible client."""
def __init__(self):
self.call_count = 0
def chat(self, model: str, messages: list) -> dict:
"""A deterministic mock that simulates agent reasoning."""
self.call_count += 1
last_msg = messages[-1]["content"] if messages else ""
# Simulate multi-step tool use for demonstration
if "research" in last_msg.lower() or self.call_count <= 2:
return {
"content": "I need to search the web first, then analyze the results.",
"tool_calls": [
{
"id": f"call_{self.call_count}",
"type": "function",
"function": {
"name": "search_web",
"arguments": json.dumps({"query": last_msg}),
},
}
],
}
if "search_web" in last_msg or "result" in last_msg.lower():
return {
"content": "Based on my research, here is a comprehensive answer to the original question.",
"tool_calls": [],
}
return {
"content": "I cannot complete this task without additional information or tools.",
"tool_calls": [],
}
def main():
llm = MockLLMClient()
memory = VectorMemoryStore()
orchestrator = AgentOrchestrator(
llm_client=llm,
model="mock",
registry=registry,
memory=memory,
)
goal = "Research the best practices for building reliable AI agents in 2025"
print(f"🤖 Agent started. Goal: {goal}")
print("=" * 60)
trace = orchestrator.run(goal)
print(f"\n✅ Completed in {trace['steps'][-1]['step']} steps")
print(f"\nFinal output:")
print(trace["final_output"])
print(f"\n🧠 Memory stats: {json.dumps(orchestrator.get_memory_stats(), indent=2)}")
print("\n--- Execution Trace ---")
for step in trace["steps"]:
print(f"\n[Step {step['step']}] {step['type'].upper()}")
print(f" Thought: {step['thought'][:100]}...")
if "actions" in step:
for action in step["actions"]:
print(f" → {action['tool']}: {action['result'][:80]}...")
if __name__ == "__main__":
main()
Beyond the Basics: Production Considerations
Building a working prototype is one thing. Shipping it is another. Here's what separates lab demos from production agents:
1. Hierarchical Memory
Flat vector recall works for small systems. Production agents need a hierarchy:
- Semantic memory: General knowledge and patterns, persisted indefinitely
- Episodic memory: Task-specific experiences, indexed by session
- Procedural memory: Learned tool-composition strategies that improve over time
The key insight is that these layers have different TTLs and update frequencies. Semantic memories rarely change. Episodic memories decay. Procedural memories are reinforced through success and penalized through failure.
2. Reflection as a First-Class Operation
The most powerful agents don't just act—they think about their thinking. After each step, a reflection module evaluates:
- Did the action move us closer to the goal?
- Were the tool choices appropriate?
- Is there a better strategy we haven't tried?
This transforms the agent from a reactive executor into an adaptive reasoner. You can implement this as a separate LLM call with a dedicated reflection prompt, or embed it in the main loop with constrained output formats.
3. Memory-Guided Planning
Instead of planning from scratch every turn, the agent should consult its memory for analogous past situations. This is analogous to how humans solve new problems—they don't derive solutions from first principles; they adapt approaches that worked before.
def recall_past_similar(self, current_goal: str) -> list[str]:
"""Find past goals that are semantically similar and return their strategies."""
memories = self.memory.recall(current_goal, k=5)
strategies = []
for m in memories:
if m.type == "decision" and m.importance > 1.0:
strategies.append(f"Previously: {m.content}")
return strategies
4. Context Budgeting
Every token in your prompt costs money and adds latency. A disciplined agent manages its context window like a scarce resource:
- Compress old memories into summaries rather than keeping raw text
- Drop irrelevant memories before each step
- Maintain a sliding context window (last N steps) alongside long-term memory
Conclusion: The Architecture That Changes Everything
The single most impactful architectural decision you can make for an AI agent is where memory lives.
When memory lives in the prompt, you get fragile, expensive, context-limited agents that forget everything between turns. When memory lives externally—in structured, searchable, semantically-aware stores—you get agents that accumulate experience, avoid repeating mistakes, and compound their capabilities over time.
The self-driving architecture I've outlined here isn't a single library or framework. It's a pattern:
- Explicit memory that survives across sessions
- Declarative tool registration that decouples capabilities from reasoning
- Autonomous orchestration that closes the perception-action loop
- Reflection and learning that turns experience into improvement
This pattern works with any LLM, any tool set, and any deployment target. It works today. You don't need a new framework to adopt it—you just need to stop treating memory as an afterthought and start treating it as the foundation.
The agents that succeed won't be the ones with the biggest context windows. They'll be the ones that remember.
Build accordingly.
