Why Your AI Agent Keeps Crashing: Memory, Context, and State Management
You deploy your AI agent, it works beautifully for the first few requests, and then—silence. A 500 error. A timeout. A garbled response that makes no sense. If this sounds familiar, you're not alone. The root cause almost always traces back to three interconnected systems: memory, context, and state management. This tutorial will walk you through each one, show you why they fail, and give you production-ready patterns to keep your agent running reliably.
1. Understanding the Three Pillars
Before diving into code, let's define precisely what each term means in the context of AI agents—because conflating them is the first source of bugs.
1.1 Memory: The Agent's Long-Term Knowledge
Memory is anything the agent retains across sessions or conversations. It can be:
- Episodic memory — records of past interactions ("User asked about pricing last Tuesday")
- Semantic memory — facts and knowledge ("The company's refund policy is 30 days")
- Procedural memory — learned workflows ("When a user says 'cancel', first confirm, then route to retention")
Memory is persistent. It survives restarts. The crash happens when memory retrieval is too slow, too large, or returns irrelevant data that pollutes the next step.
1.2 Context: The Agent's Working Window
Context is the information the model can see right now—the concatenation of system prompts, conversation history, retrieved documents, tool outputs, and reasoning traces all stuffed into a single prompt. Every LLM has a context window limit (e.g., 128K tokens for GPT-4 Turbo, 200K for Claude 3.5 Sonnet). When you exceed it, the model either truncates silently, refuses with an error, or produces degraded output.
1.3 State: The Agent's Current Position in a Workflow
State answers: "Where am I right now in this multi-step process?" It tracks which tools have been called, what decisions have been made, what's pending, and whether the agent is waiting for user input. Unlike context, state is often structured (a JSON object, a finite state machine) and doesn't need to be fully loaded into the prompt.
Here's the critical insight: memory overload crashes context, context corruption breaks state, and lost state forces the agent to repeat itself endlessly until it exhausts the context window. They form a vicious cycle.
2. Why Your Agent Crashes: The Failure Patterns
2.1 Context Window Overflow
The most common crash. You naively append every conversation turn, every retrieved document, and every tool result to the prompt. Around message 20, the agent starts producing truncated JSON, missing function call arguments, or simply times out.
# THE NAIVE APPROACH THAT WILL CRASH
prompt = system_prompt
for turn in conversation_history:
prompt += f"User: {turn.user}\nAssistant: {turn.assistant}\n"
prompt += f"User: {current_message}\n"
# By message 30 with retrieved docs, you're at 150K tokens on a 128K model
# The API returns: "context_length_exceeded" or silently truncates early messages
response = llm.complete(prompt) # BOOM2.2 Memory Retrieval Poisoning
You store every interaction in a vector database and retrieve "relevant" chunks before each call. But similarity search isn't perfect. You pull in 15 "relevant" memories that are actually noise. The model now has to reason through irrelevant data, gets confused, and produces a wrong tool call that puts the agent into an unreachable state.
# POISONED RETRIEVAL
memories = vector_store.similarity_search(user_query, k=20)
memory_text = "\n".join([m.content for m in memories])
# Now your prompt is 60% irrelevant memories, 30% conversation, 10% instructions
# The model loses track of what it's supposed to do
prompt = f"{system_prompt}\n\nRelevant Memories:\n{memory_text}\n\nConversation:\n{history}\n\nUser: {user_query}"
# Agent responds with a hallucinated tool name that doesn't exist2.3 State Drift and Orphaned Actions
Your agent is in the middle of a booking flow: it has asked for dates, the user provided them, and now it needs to ask for passenger count. But the conversation history got truncated due to context overflow, so the agent "forgot" it already asked for dates and asks again. The user gets frustrated, the conversation grows longer, the context fills faster, and the agent crashes harder. This is the death spiral.
# STATE DRIFT WITHOUT PERSISTENCE
# Turn 1
agent: "What dates would you like to book?"
# Turn 2
user: "March 10-15"
# Turn 3 - but context was truncated, losing Turn 1
agent: "Sure! First, what dates would you like to book?" # ASKS AGAIN
# User gets angry, types a long complaint, context overflows, agent crashes3. How to Fix It: Practical Implementations
3.1 Context Management: Sliding Windows with Summarization
Stop appending everything. Use a sliding window combined with progressive summarization. Keep only the last N turns in raw form, and compress older turns into a growing summary.
class ContextManager:
def __init__(self, max_recent_turns=10, summary_trigger=15):
self.recent_turns = [] # Last N turns in raw form
self.summary = "" # Compressed history of everything older
self.max_recent_turns = max_recent_turns
self.summary_trigger = summary_trigger
def add_turn(self, user_msg: str, assistant_msg: str):
self.recent_turns.append({
"user": user_msg,
"assistant": assistant_msg
})
# When recent turns exceed trigger, compress oldest into summary
if len(self.recent_turns) > self.summary_trigger:
# Take turns that will be evicted
to_summarize = self.recent_turns[:-self.max_recent_turns]
new_summary = self._summarize_turns(to_summarize)
# Merge with existing summary
if self.summary:
self.summary = self._merge_summaries(self.summary, new_summary)
else:
self.summary = new_summary
# Keep only the recent window
self.recent_turns = self.recent_turns[-self.max_recent_turns:]
def _summarize_turns(self, turns: list) -> str:
prompt = f"""Summarize this conversation segment into a compact paragraph.
Preserve: decisions made, facts shared, actions taken, pending items.
Discard: pleasantries, filler, repeated information.
Segment:
{self._format_turns(turns)}
"""
return llm.complete(prompt, max_tokens=300)
def _merge_summaries(self, old: str, new: str) -> str:
prompt = f"""Merge these two conversation summaries into one cohesive summary.
Eliminate duplicates. Preserve chronological order of key events.
Existing Summary:
{old}
New Segment Summary:
{new}
"""
return llm.complete(prompt, max_tokens=400)
def build_context(self, system_prompt: str, retrieved_knowledge: str = "") -> str:
parts = [system_prompt]
if self.summary:
parts.append(f"## Conversation Summary (Earlier)\n{self.summary}")
if retrieved_knowledge:
parts.append(f"## Relevant Knowledge\n{retrieved_knowledge}")
if self.recent_turns:
parts.append("## Recent Conversation\n" + self._format_turns(self.recent_turns))
return "\n\n".join(parts)
def _format_turns(self, turns: list) -> str:
return "\n".join([f"User: {t['user']}\nAssistant: {t['assistant']}" for t in turns])
# Usage
ctx = ContextManager(max_recent_turns=10, summary_trigger=15)
ctx.add_turn("I need to book a flight", "Sure! Where are you flying to?")
ctx.add_turn("New York to London", "Got it. What dates?")
# ... many turns later, context stays compact
prompt = ctx.build_context(system_prompt, retrieved_knowledge="")
# prompt is consistently ~8-12K tokens regardless of conversation length3.2 Memory: Tiered Retrieval with Relevance Filtering
Don't dump raw vector search results into context. Implement a tiered memory system with strict filtering. Use semantic search to find candidates, then use an LLM-based filter to decide what actually matters for the current task.
class TieredMemory:
def __init__(self, vector_store, relational_store):
self.vector_store = vector_store # For semantic search
self.relational_store = relational_store # For exact lookups (user_id, dates)
def retrieve(self, query: str, user_id: str, max_tokens_budget: int = 2000) -> str:
# Tier 1: Exact relational lookups (fast, deterministic)
user_profile = self.relational_store.get(f"user:{user_id}:profile")
recent_actions = self.relational_store.get_range(
f"user:{user_id}:actions",
last_n=5
)
# Tier 2: Semantic search candidates
candidates = self.vector_store.similarity_search(
query,
filter={"user_id": user_id},
k=20
)
# Tier 3: Relevance filter using a cheap model
relevant = []
for candidate in candidates:
# Quick classifier: is this memory relevant to the query?
relevance_score = self._check_relevance(query, candidate.content)
if relevance_score > 0.6:
relevant.append(candidate)
# Tier 4: Rank and truncate to token budget
relevant.sort(key=lambda x: x.relevance_score, reverse=True)
result_parts = []
token_count = 0
if user_profile:
result_parts.append(f"[User Profile] {user_profile}")
token_count += self._estimate_tokens(user_profile)
for mem in relevant:
mem_tokens = self._estimate_tokens(mem.content)
if token_count + mem_tokens > max_tokens_budget:
break
result_parts.append(f"[Memory] {mem.content}")
token_count += mem_tokens
return "\n".join(result_parts)
def _check_relevance(self, query: str, memory: str) -> float:
prompt = f"""Rate how relevant this memory is to the query on a scale of 0-1.
Query: {query}
Memory: {memory[:500]}
Respond with only a number like 0.85"""
try:
return float(llm.complete(prompt, max_tokens=5))
except:
return 0.0
def _estimate_tokens(self, text: str) -> int:
# Rough estimate: 1 token ≈ 4 characters for English
return len(text) // 4
# Usage
memory = TieredMemory(vector_store, redis_client)
relevant_context = memory.retrieve(
query="What's the refund policy for my last order?",
user_id="user_123",
max_tokens_budget=2000
)
# Only truly relevant memories make it into the prompt3.3 State Management: Finite State Machines with Checkpointing
State must be explicit, persisted, and recoverable. Model the agent's workflow as a finite state machine. At every transition, checkpoint the state to a durable store. If the agent crashes mid-flow, it resumes exactly where it left off—not from the beginning.
from enum import Enum
from typing import Optional, Dict, Any
import json
import redis
class BookingState(str, Enum):
IDLE = "idle"
GATHERING_ORIGIN = "gathering_origin"
GATHERING_DESTINATION = "gathering_destination"
GATHERING_DATES = "gathering_dates"
GATHERING_PASSENGERS = "gathering_passengers"
CONFIRMING = "confirming"
BOOKING = "booking"
COMPLETED = "completed"
FAILED = "failed"
class StateMachine:
def __init__(self, session_id: str, persistence: redis.Redis):
self.session_id = session_id
self.persistence = persistence
self.state = self._load_state()
def _load_state(self) -> Dict[str, Any]:
raw = self.persistence.get(f"state:{self.session_id}")
if raw:
return json.loads(raw)
return {
"current": BookingState.IDLE,
"collected_data": {},
"retry_count": 0,
"last_transition": None,
"checkpoint_timestamp": None
}
def _save_state(self):
self.state["checkpoint_timestamp"] = time.time()
self.persistence.set(
f"state:{self.session_id}",
json.dumps(self.state),
ex=3600 # 1 hour TTL
)
def transition(self, to: BookingState, data_update: Optional[Dict] = None):
"""Atomically transition state and checkpoint."""
previous = self.state["current"]
if not self._is_valid_transition(previous, to):
raise ValueError(f"Invalid transition: {previous} -> {to}")
self.state["current"] = to
self.state["last_transition"] = f"{previous.value} -> {to.value}"
self.state["retry_count"] = 0 # Reset on successful transition
if data_update:
self.state["collected_data"].update(data_update)
self._save_state()
def _is_valid_transition(self, from_state: BookingState, to: BookingState) -> bool:
transitions = {
BookingState.IDLE: [BookingState.GATHERING_ORIGIN, BookingState.GATHERING_DESTINATION],
BookingState.GATHERING_ORIGIN: [BookingState.GATHERING_DESTINATION],
BookingState.GATHERING_DESTINATION: [BookingState.GATHERING_DATES],
BookingState.GATHERING_DATES: [BookingState.GATHERING_PASSENGERS],
BookingState.GATHERING_PASSENGERS: [BookingState.CONFIRMING],
BookingState.CONFIRMING: [BookingState.BOOKING, BookingState.GATHERING_DATES], # Can go back
BookingState.BOOKING: [BookingState.COMPLETED, BookingState.FAILED],
BookingState.FAILED: [BookingState.IDLE, BookingState.CONFIRMING],
BookingState.COMPLETED: [BookingState.IDLE]
}
return to in transitions.get(from_state, [])
def increment_retry(self):
self.state["retry_count"] += 1
self._save_state()
if self.state["retry_count"] > 3:
self.state["current"] = BookingState.FAILED
self._save_state()
return False # Signal that we've exceeded retry limit
return True
def get_resume_prompt(self) -> str:
"""Generate a prompt that tells the agent exactly where it left off."""
current = self.state["current"]
collected = self.state["collected_data"]
resume_instructions = {
BookingState.GATHERING_ORIGIN: "You are collecting the departure city. Ask the user for their origin.",
BookingState.GATHERING_DESTINATION: f"You have origin: {collected.get('origin')}. Now ask for the destination.",
BookingState.GATHERING_DATES: f"You have origin: {collected.get('origin')}, destination: {collected.get('destination')}. Ask for travel dates.",
BookingState.GATHERING_PASSENGERS: f"You have the route and dates. Now ask for the number of passengers.",
BookingState.CONFIRMING: f"Summarize the booking details and ask for confirmation: {json.dumps(collected)}",
BookingState.FAILED: "Apologize and offer to start over or resume from where the error occurred."
}
return resume_instructions.get(current, "Determine the next appropriate step.")
# Usage with the agent loop
async def agent_loop(session_id: str, user_input: str):
redis_client = redis.Redis()
sm = StateMachine(session_id, redis_client)
try:
# Build context INCLUDING state resume instructions
resume_prompt = sm.get_resume_prompt()
context = ctx_manager.build_context(
system_prompt=f"You are a flight booking agent.\nCurrent Step: {resume_prompt}",
retrieved_knowledge=memory.retrieve(user_input, session_id)
)
response = await llm.complete_async(context + f"\n\nUser: {user_input}")
# Parse response for state transitions
parsed = parse_agent_response(response)
if parsed.intent == "provide_info":
if "origin" in parsed.data:
sm.transition(BookingState.GATHERING_DESTINATION, parsed.data)
elif "destination" in parsed.data:
sm.transition(BookingState.GATHERING_DATES, parsed.data)
# ... etc
return response
except Exception as e:
can_retry = sm.increment_retry()
if can_retry:
# Retry with the same state
return await agent_loop(session_id, user_input)
else:
return "I'm having trouble. Let's start over. What can I help you with?"4. Putting It All Together: The Resilient Agent Architecture
Here's a complete agent loop that integrates all three systems. This is the pattern that prevents crashes in production.
import asyncio
import time
import logging
from typing import Optional
logger = logging.getLogger(__name__)
class ResilientAgent:
def __init__(
self,
context_manager: ContextManager,
memory: TieredMemory,
state_machine: StateMachine,
llm_client,
max_context_tokens: int = 100_000,
safety_margin: int = 10_000
):
self.ctx = context_manager
self.memory = memory
self.state = state_machine
self.llm = llm_client
self.max_context_tokens = max_context_tokens
self.safety_margin = safety_margin
async def handle_message(self, user_id: str, session_id: str, message: str) -> str:
"""Process a user message with full resilience patterns."""
# 1. Load state checkpoint
resume_instructions = self.state.get_resume_prompt()
collected_data = self.state.state.get("collected_data", {})
# 2. Retrieve memories within strict budget
memory_budget = 2000 # tokens for memories
relevant_memories = self.memory.retrieve(
query=message,
user_id=user_id,
max_tokens_budget=memory_budget
)
# 3. Build context with explicit token counting
system_prompt = self._build_system_prompt(resume_instructions, collected_data)
base_prompt = self.ctx.build_context(system_prompt, relevant_memories)
current_prompt = f"{base_prompt}\n\nUser: {message}"
# 4. Pre-flight token check
estimated_tokens = self._estimate_tokens(current_prompt)
available = self.max_context_tokens - self.safety_margin
if estimated_tokens > available:
logger.warning(f"Token overflow risk: {estimated_tokens} > {available}")
# Aggressive context trimming
self.ctx.max_recent_turns = max(3, self.ctx.max_recent_turns - 3)
current_prompt = self._rebuild_with_trimmed_context(
system_prompt, relevant_memories, message
)
# 5. Execute with timeout and retry
try:
response = await asyncio.wait_for(
self.llm.complete_async(current_prompt),
timeout=30.0
)
# 6. Validate response structure
if not self._validate_response(response):
raise ValueError("Invalid response structure")
# 7. Update state based on response
self._update_state_from_response(response)
# 8. Store this turn in memory
self._store_turn(user_id, session_id, message, response)
# 9. Update context manager
self.ctx.add_turn(message, response)
return response
except asyncio.TimeoutError:
logger.error("LLM timeout")
return await self._handle_error(user_id, session_id, "timeout")
except Exception as e:
logger.error(f"Agent error: {e}")
return await self._handle_error(user_id, session_id, str(e))
def _build_system_prompt(self, resume_instructions: str, collected_data: dict) -> str:
return f"""You are a helpful AI assistant.
## Current Task
{resume_instructions}
## Data Collected So Far
{json.dumps(collected_data, indent=2) if collected_data else "None yet"}
## Important Rules
- Stay focused on the current task step
- Do not repeat questions already answered in Collected Data
- If you encounter an error, clearly state what went wrong
- Never invent information not provided by the user"""
def _estimate_tokens(self, text: str) -> int:
# Conservative estimate: 1 token per 3.5 characters for English
return len(text) // 3
def _validate_response(self, response: str) -> bool:
# Check for truncation markers
truncation_indicators = [
response.endswith("..."),
len(response) < 10,
response.count("{") != response.count("}"), # Unbalanced JSON
]
return not any(truncation_indicators)
def _update_state_from_response(self, response: str):
"""Parse response and update state machine."""
# Extract structured data from response using regex or JSON parsing
import re
# Look for JSON blocks
json_match = re.search(r'\{[\s\S]*\}', response)
if json_match:
try:
data = json.loads(json_match.group(0))
if "action" in data:
# Map actions to state transitions
action_to_state = {
"collect_origin": BookingState.GATHERING_ORIGIN,
"collect_destination": BookingState.GATHERING_DESTINATION,
"collect_dates": BookingState.GATHERING_DATES,
"collect_passengers": BookingState.GATHERING_PASSENGERS,
"confirm": BookingState.CONFIRMING,
"book": BookingState.BOOKING,
}
new_state = action_to_state.get(data["action"])
if new_state:
self.state.transition(new_state, data.get("collected", {}))
except json.JSONDecodeError:
pass
def _store_turn(self, user_id: str, session_id: str, user_msg: str, assistant_msg: str):
"""Persist this interaction to both vector and relational memory."""
# Store full turn in relational store
self.memory.relational_store.push(
f"user:{user_id}:actions",
json.dumps({"user": user_msg, "assistant": assistant_msg, "ts": time.time()})
)
# Store embedding for semantic retrieval
embedding = embed(user_msg)
self.memory.vector_store.add(
text=f"User: {user_msg}\nAssistant: {assistant_msg}",
embedding=embedding,
metadata={"user_id": user_id, "session_id": session_id}
)
def _rebuild_with_trimmed_context(self, system_prompt: str, memories: str, message: str) -> str:
"""Emergency context rebuild when near token limit."""
# Keep only 3 most recent turns
self.ctx.recent_turns = self.ctx.recent_turns[-3:]
base = self.ctx.build_context(system_prompt, memories)
return f"{base}\n\nUser: {message}"
async def _handle_error(self, user_id: str, session_id: str, error_type: str) -> str:
"""Graceful error recovery."""
can_retry = self.state.increment_retry()
if can_retry:
# Wait briefly, then allow retry
await asyncio.sleep(1)
return "I encountered a hiccup. Could you repeat your last request?"
else:
# Reset state machine
self.state.state["current"] = BookingState.IDLE
self.state.state["retry_count"] = 0
self.state._save_state()
return "I apologize—I'm having persistent issues. Let's start fresh. How can I help you?"5. Best Practices and Design Principles
5.1 Token Budgeting Is Non-Negotiable
Never guess. Always estimate tokens before calling the LLM. Allocate explicit budgets for each component:
- System prompt: fixed allocation (e.g., 2,000 tokens)
- Retrieved memories: capped allocation (e.g., 2,000 tokens)
- Conversation summary: dynamic, but monitored (e.g., up to 3,000 tokens)
- Recent conversation turns: whatever remains within the safety margin
- Safety margin: always reserve 10-15% of the context window for the model's response and unexpected tool outputs
5.2 Checkpoint Before Every LLM Call
State must be persisted before you invoke the model, not after. If the call times out or crashes, you can resume from the exact pre-call state. This is the single most important reliability pattern.
5.3 Use Structured Outputs for State Transitions
Don't rely on natural language parsing to determine state changes. Force the model to emit structured data (JSON, function calls) that explicitly signals state transitions. This eliminates ambiguity.
# Good: Structured output forces explicit state signaling
system_prompt = """...
When you have collected a piece of information, output:
{
"action": "collect_origin",
"collected": {"origin": "New York"},
"next_question": "Where would you like to fly to?"
}"""5.4 Implement Graceful Degradation
When context is nearly full, don't crash—degrade. Reduce the conversation window to the last 3 turns. Drop lower-priority memories. Simplify the system prompt. The agent may be less context-aware, but it won't crash.
5.5 Test with Token Counting Enabled
Most developers test with short conversations and never hit the limit. Use a token counter in your test harness and simulate 50+ turn conversations. Verify that your context manager keeps the prompt size stable.
# Token-aware test harness
def test_long_conversation():
agent = ResilientAgent(...)
for i in range(100):
prompt = agent._build_full_prompt(f"Test message {i}")
token_count = count_tokens(prompt)
assert token_count < agent.max_context_tokens - agent.safety_margin, \
f"Token overflow at turn {i}: {token_count} tokens"
print("Passed: Context remains stable over 100 turns")5.6 Separate Memory Storage from Memory Retrieval
Storage should be write-only and fire-and-forget (async). Retrieval should be read-optimized with strict filtering. Never let retrieval block the main agent loop.
5.7 Monitor the Death Spiral Indicators
Set up alerts for: retry count spikes, average tokens per call creeping up, state machine reset frequency, and time-to-first-token increasing. These are early warnings that your context management is failing.
Conclusion
AI agents crash not because LLMs are unreliable, but because developers treat memory, context, and state as an afterthought. The fix is architectural: implement progressive summarization to keep context bounded, tiered memory retrieval with relevance filtering to prevent noise pollution, and persistent finite state machines with checkpointing so the agent never loses its place. When these three systems work in concert—with explicit token budgets, structured state transitions, and graceful degradation—your agent will handle conversations of any length without crashing. The patterns in this tutorial are battle-tested. Start with the ResilientAgent class above, adapt it to your framework, and you'll eliminate the most common production failures overnight.