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.
To understand the fix, we must first diagnose the disease. An AI agent typically operates in a loop:
search_web
, execute_code
).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.
LLMs are probabilistic engines. As context grows, the probability mass spreads thinner across irrelevant tokens. This leads to two common failure modes:
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.
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:
How do we build this? We need three technical components.
Instead of stuffing the entire conversation history into the context window, we extract critical state into a semantic memory store.
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.
This is the "self-driving" brain. It sits between the LLM and the tools. Its responsibilities include:
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.
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):
relevant_context = self.memory.query(user_query, top_k=3)
prompt = self._construct_prompt(user_query, relevant_context)
decision = self.llm.generate(prompt)
if decision.action == "call_tool":
tool_result = self.tools.execute(decision.tool_name, decision.params)
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}")
)
return self._handle_result(decision, tool_result)
return decision.final_answer
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
def retrieve_memory(state: AgentState) -> AgentState:
query = state['messages'][-1].content
relevant_docs = vector_store.similarity_search(query, k=2)
state['memory'] = relevant_docs
return state
def call_tool(state: AgentState) -> AgentState:
state['tool_results'][tool_name] = result
return state
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()
memory
key in the state is populated from an external source, not just accumulated history.tool_results
dict acts as a short-term cache, preventing the LLM from needing to remember raw outputs from previous turns.The most robust self-driving agents incorporate self-reflection. After a tool execution, the agent should evaluate:
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.
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.
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 off 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.
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.
Tools are the agent's hands. Every capability must be declaratively registered so the orchestrator can reason about them.
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())
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]:
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,
))
This is where we solve the memory bottleneck. Every observation, tool result, and decision becomes a structured memory with semantic embedding.
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,
}
This is the core—where autonomous decision-making happens. The orchestrator runs a loop: observe → plan → act → reflect → store.
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 = []
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."""
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}"
messages.append({
"role": "system",
"content": f"Recalled context:{memory_context}",
})
response = self.llm.chat(self.model, messages)
thought = response.get("content", "")
tool_calls = response.get("tool_calls", [])
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})
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(),
))
for r in results:
messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"content": r["result"],
})
response = self.llm.chat(self.model, messages)
thought = response.get("content", "")
return {
"type": "action",
"step": self.step_count,
"thought": thought,
"actions": results,
"output": thought,
}
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()
Here's how you'd run the full system end-to-end:
import json
from tools import ToolRegistry, registry
from memory import VectorMemoryStore
from orchestrator import AgentOrchestrator
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 ""
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()
Building a working prototype is one thing. Shipping it is another. Here's what separates lab demos from production agents:
Flat vector recall works for small systems. Production agents need a hierarchy:
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.
The most powerful agents don't just act—they think about their thinking. After each step, a reflection module evaluates:
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.
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
Every token in your prompt costs money and adds latency. A disciplined agent manages its context window like a scarce resource:
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:
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.