April 23, 2026 · 11 min read
Stop letting context windows bankrupt your AI budget. Learn sliding window, summarization, and vectorized memory strategies with real cost savings from production deployments.
I got the invoice at the end of March: $2,847 for LLM API calls. For a side project. The culprit? I was appending every conversation turn to the context window. After 50 multi-turn sessions, I was sending 50,000 tokens per request.
You cannot just append everything to the prompt forever. Context windows have limits (128k for GPT-4) and costs scale linearly with token count. In this post, I'll show you three memory management strategies that cut my costs by 70% while improving response quality.
The math:
The constraint:
In the DeepAgent project, I implemented multi-turn conversation support with LangGraph. Without memory management, sessions with 20+ turns would cost $0.50+ per message.
Keep only the last N messages. Simple, predictable, effective for short sessions.
Implementation:
from typing import List, Dict
from dataclasses import dataclass
@dataclass
class Message:
role: str # "user" or "assistant"
content: str
timestamp: float
class SlidingWindowMemory:
def __init__(self, max_messages: int = 10):
self.max_messages = max_messages
self.messages: List[Message] = []
def add_message(self, role: str, content: str):
"""Add message and maintain window size."""
import time
message = Message(role=role, content=content, timestamp=time.time())
self.messages.append(message)
# Remove oldest if over limit
if len(self.messages) > self.max_messages:
self.messages = self.messages[-self.max_messages:]
def get_context(self) -> List[Dict[str, str]]:
"""Return messages formatted for LLM API."""
return [
{"role": msg.role, "content": msg.content}
for msg in self.messages
]
def estimate_tokens(self) -> int:
"""Rough token estimation."""
total_chars = sum(len(msg.content) for msg in self.messages)
return total_chars // 4 # ~4 chars per token
# Usage with LangGraph
from langgraph.graph import StateGraph
class AgentState:
def __init__(self):
self.memory = SlidingWindowMemory(max_messages=10)
self.session_id: str = ""
def chat_node(state: AgentState, user_input: str):
# Add user message
state.memory.add_message("user", user_input)
# Get sliding window context
messages = state.memory.get_context()
# Call LLM with limited context
response = llm_client.chat.completions.create(
model="gpt-4",
messages=messages
)
# Store assistant response
state.memory.add_message("assistant", response.choices[0].message.content)
return response
Trade-offs:
When to use:
Once context reaches a threshold, summarize older messages and replace them with the summary.
Implementation:
class SummarizationMemory:
def __init__(
self,
max_messages: int = 10,
token_threshold: int = 3000,
summary_model: str = "gpt-3.5-turbo" # Cheaper for summaries
):
self.max_messages = max_messages
self.token_threshold = token_threshold
self.summary_model = summary_model
self.messages: List[Message] = []
self.summary: str = ""
def add_message(self, role: str, content: str):
import time
message = Message(role=role, content=content, timestamp=time.time())
self.messages.append(message)
# Check if we need to summarize
if self.estimate_tokens() > self.token_threshold:
self._summarize_older_messages()
def _summarize_older_messages(self):
"""Summarize oldest half of messages."""
if len(self.messages) < 4:
return
# Take oldest 50% for summarization
to_summarize = self.messages[:len(self.messages)//2]
to_keep = self.messages[len(self.messages)//2:]
# Create summary prompt
conversation_text = "\n".join([
f"{msg.role}: {msg.content}"
for msg in to_summarize
])
summary_prompt = f"""Summarize this conversation concisely, preserving key facts and context:
{conversation_text}
Summary:"""
# Generate summary with cheaper model
summary_response = llm_client.chat.completions.create(
model=self.summary_model,
messages=[{"role": "user", "content": summary_prompt}],
max_tokens=200
)
new_summary = summary_response.choices[0].message.content
# Combine with existing summary
if self.summary:
self.summary = f"Previous context: {self.summary}\nRecent: {new_summary}"
else:
self.summary = new_summary
# Replace summarized messages
self.messages = to_keep
def get_context(self) -> List[Dict[str, str]]:
"""Return summary + recent messages."""
context = []
if self.summary:
context.append({
"role": "system",
"content": f"Previous conversation summary: {self.summary}"
})
context.extend([
{"role": msg.role, "content": msg.content}
for msg in self.messages
])
return context
def estimate_tokens(self) -> int:
summary_tokens = len(self.summary) // 4 if self.summary else 0
message_tokens = sum(len(msg.content) // 4 for msg in self.messages)
return summary_tokens + message_tokens
Cost analysis:
Trade-offs:
When to use:
Store conversation embeddings in a vector database. Retrieve only semantically relevant past messages for each turn.
Architecture:
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict, Tuple
class VectorizedMemory:
def __init__(
self,
embedding_model: str = "all-MiniLM-L6-v2",
vector_db_client=None, # Pinecone, Milvus, or Chroma
top_k: int = 5
):
self.embedding_model = SentenceTransformer(embedding_model)
self.vector_db = vector_db_client
self.top_k = top_k
self.recent_messages: List[Message] = [] # Keep last 3 for recency
self.max_recent = 3
def add_message(self, role: str, content: str, session_id: str):
"""Store message in vector DB with embedding."""
import time
message = Message(role=role, content=content, timestamp=time.time())
# Generate embedding
embedding = self.embedding_model.encode(content)
# Store in vector DB
message_id = f"{session_id}_{int(time.time() * 1000)}"
self.vector_db.upsert(
vectors=[{
"id": message_id,
"values": embedding.tolist(),
"metadata": {
"role": role,
"content": content,
"timestamp": message.timestamp,
"session_id": session_id
}
}]
)
# Maintain recent messages
self.recent_messages.append(message)
if len(self.recent_messages) > self.max_recent:
self.recent_messages = self.recent_messages[-self.max_recent:]
def get_relevant_context(
self,
current_query: str,
session_id: str
) -> List[Dict[str, str]]:
"""Retrieve semantically relevant past messages."""
# Get query embedding
query_embedding = self.embedding_model.encode(current_query)
# Search vector DB
results = self.vector_db.query(
vector=query_embedding.tolist(),
top_k=self.top_k,
filter={"session_id": session_id} # Only this session
)
# Format results
relevant_messages = []
for match in results.matches:
relevant_messages.append({
"role": match.metadata["role"],
"content": match.metadata["content"]
})
# Combine with recent messages (recency bias)
recent_context = [
{"role": msg.role, "content": msg.content}
for msg in self.recent_messages
]
# Deduplicate while preserving order
seen = set()
combined = []
for msg in recent_context + relevant_messages:
key = (msg["role"], msg["content"])
if key not in seen:
seen.add(key)
combined.append(msg)
return combined
def estimate_tokens(self, context: List[Dict[str, str]]) -> int:
"""Estimate tokens for context."""
total_chars = sum(len(msg["content"]) for msg in context)
return total_chars // 4
# Usage with LangGraph
class LangGraphVectorMemory:
"""Vectorized memory for LangGraph state management."""
def __init__(self, session_id: str):
self.session_id = session_id
self.memory = VectorizedMemory()
self.short_term = [] # Last 2 turns for immediate context
def add_turn(self, user_input: str, assistant_response: str):
"""Add conversation turn to memory."""
self.memory.add_message("user", user_input, self.session_id)
self.memory.add_message("assistant", assistant_response, self.session_id)
# Update short-term memory
self.short_term.append({"role": "user", "content": user_input})
self.short_term.append({"role": "assistant", "content": assistant_response})
if len(self.short_term) > 4: # Keep last 2 exchanges
self.short_term = self.short_term[-4:]
def get_context_for_query(self, query: str) -> List[Dict[str, str]]:
"""Get context combining short-term + relevant historical."""
# Get semantically relevant historical messages
relevant = self.memory.get_relevant_context(query, self.session_id)
# Combine with short-term (recency is always included)
context = self.short_term + relevant
return context
Cost analysis:
Trade-offs:
When to use:
After testing all three in production, here's the winning combination:
class HybridMemoryManager:
"""
Production memory strategy:
- Always keep last 3 turns (immediate context)
- Summarize turns 4-10 (medium-term context)
- Vector retrieval for turns 11+ (long-term context)
"""
def __init__(self, session_id: str):
self.session_id = session_id
# Short-term: Sliding window (always included)
self.short_term = SlidingWindowMemory(max_messages=6) # 3 turns
# Medium-term: Summarization
self.summary = ""
self.messages_since_summary = 0
# Long-term: Vectorized storage
self.vector_memory = VectorizedMemory()
def add_message(self, role: str, content: str):
import time
# Always add to short-term
self.short_term.add_message(role, content)
# Add to vector DB for long-term retrieval
self.vector_memory.add_message(role, content, self.session_id)
self.messages_since_summary += 1
# Trigger summarization every 4 messages (2 turns)
if self.messages_since_summary >= 4 and len(self.short_term.messages) >= 6:
self._update_summary()
def _update_summary(self):
"""Summarize older messages."""
# Get messages to summarize (oldest 2 turns)
to_summarize = self.short_term.messages[:4]
conversation = "\n".join([
f"{msg.role}: {msg.content}" for msg in to_summarize
])
summary_prompt = f"Summarize key facts from this conversation:\n\n{conversation}"
# Use cheaper model for summary
response = llm_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": summary_prompt}],
max_tokens=150
)
new_summary = response.choices[0].message.content
if self.summary:
self.summary = f"{self.summary} | {new_summary}"
else:
self.summary = new_summary
self.messages_since_summary = 0
def get_context(self, current_query: str = None) -> List[Dict[str, str]]:
"""Build context from all three memory layers."""
context = []
# Layer 1: Summary (if exists)
if self.summary:
context.append({
"role": "system",
"content": f"Previous conversation context: {self.summary}"
})
# Layer 2: Short-term (recent 3 turns)
context.extend(self.short_term.get_context())
# Layer 3: Vector retrieval (if query provided)
if current_query:
relevant = self.vector_memory.get_relevant_context(
current_query, self.session_id
)
# Filter out messages already in short-term
short_term_contents = {msg.content for msg in self.short_term.messages}
for msg in relevant:
if msg["content"] not in short_term_contents:
context.append(msg)
return context
def estimate_cost_savings(self) -> Dict[str, float]:
"""Calculate cost vs naive approach."""
# Assume 50-turn conversation
naive_tokens = 50 * 300 # 300 tokens per message avg
# Hybrid approach
summary_tokens = len(self.summary) // 4 if self.summary else 0
short_term_tokens = self.short_term.estimate_tokens()
vector_retrieval_tokens = 5 * 200 # Top 5 × 200 tokens
total_tokens = summary_tokens + short_term_tokens + vector_retrieval_tokens
savings = naive_tokens - total_tokens
cost_savings = (savings / 1000) * 0.03 # $0.03 per 1K tokens
return {
"naive_tokens": naive_tokens,
"hybrid_tokens": total_tokens,
"savings_tokens": savings,
"cost_savings_per_request": cost_savings,
"percent_reduction": (savings / naive_tokens) * 100
}
Before (no memory management):
After (hybrid approach):
Latency impact:
| Criteria | Sliding Window | Summarization | Vectorized | Hybrid |
|---|---|---|---|---|
| Session length | < 10 turns | 10-30 turns | 20+ turns | Any length |
| Latency requirement | < 200ms | < 500ms | < 300ms | < 400ms |
| Cost sensitivity | High | Medium | High | High |
| Infrastructure | Minimal | Minimal | Vector DB | Vector DB |
| Context quality | Low | Medium | High | Highest |
My recommendation:
LangGraph's built-in memory support makes this easier:
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage
# Configure memory
memory = MemorySaver()
# Build graph with checkpointing
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
# Add memory checkpointing
app = workflow.compile(checkpointer=memory)
# Each thread gets isolated memory
thread_id = "user_123_session_456"
config = {"configurable": {"thread_id": thread_id}}
# Run with automatic memory management
result = app.invoke(
{"messages": [HumanMessage(content="Hello")]},
config=config
)
# Memory persists across invocations
result2 = app.invoke(
{"messages": [HumanMessage(content="What did I ask earlier?")]},
config=config # Same thread_id = same memory
)
Learn from my mistakes:
Before shipping multi-turn agents:
Memory management keeps costs down. But how do you know if your agent is actually good? In the next post, I'll cover evaluation frameworks—LLM-as-judge, deterministic checks, and A/B testing for generative AI.
Code examples: DeepAgent GitHub
Related:
Written by
Vishvdeep Dashadiya
Lead AI Engineer. Agentic AI, real-time ML systems, and cloud-native infrastructure.
Keep reading
September 20, 2026 · 6 min read
Jev is a model that never generates text — it answers typed questions with calibrated probabilities. Where that fits in a LangGraph agent, with code, and where I would not use it.
September 5, 2026 · 8 min read
Every run of the pipeline becomes a nested Langfuse trace, and a judge model scores it with DeepEval. Neither adds a millisecond to the response, because scoring starts after the client already has the answer. The wiring: a per-request callback, a fire-and-forget task that cannot fail the request, a judge decoupled from the graph LLM, and an offline golden set for the metrics live traffic cannot compute.
September 4, 2026 · 7 min read
A skill is a folder with two files. SKILL.md becomes the system prompt, skill.yaml declares which sources the run may read, and the shared LangGraph pipeline runs unchanged on an isolated thread. Slash-invocable, like a command. Here is how a domain persona parametrizes four points of one graph without forking it, and the allowlist bug a code review found along the way.
Tell me about it. I reply within one working day with a first take and no sales pitch.
