Introduction to Agent Memory Architectures
Memory is what separates a truly intelligent agent from a stateless chatbot. When you build agents with the OpenAI Agents SDK, giving them the ability to remember past interactions, retain context across sessions, and recall relevant facts at the right moment transforms them from single-turn responders into persistent, capable assistants. In this guide, we'll explore the full landscape of agent memory architectures and how to implement them using the OpenAI Agents SDK.
What Is Agent Memory?
Agent memory refers to the mechanisms an agent uses to store, organize, and retrieve information across interactions. Unlike a simple conversation buffer that holds the last few messages, a proper memory architecture lets an agent recall facts from days ago, summarize long-running projects, and selectively pull in only what's relevant to the current task.
There are three primary categories of memory that most production agents need:
- Short-term memory — the working context for the current task or conversation, typically managed through the conversation history.
- Long-term memory — persistent storage of facts, preferences, and past interactions that survive across sessions.
- Entity memory — structured knowledge about specific people, documents, projects, or objects the agent interacts with repeatedly.
Why Memory Architecture Matters
Without a deliberate memory architecture, agents hit several painful limitations. Context windows are finite, so long conversations eventually drop important details. Each new session starts from scratch, forcing users to re-explain their situation. And agents can't learn from past mistakes or build on previous work. A well-designed memory system addresses all of these problems.
Here are the key reasons memory architecture deserves careful attention:
- Context window management — Even with large context windows, stuffing every message into the prompt becomes expensive and degrades performance. Memory lets you load only what matters.
- Personalization — Remembering user preferences, past decisions, and working style makes agents feel tailored rather than generic.
- Continuity — Long-running projects benefit from agents that can pick up where they left off without a full re-briefing.
- Cost efficiency — Retrieving a few relevant memory snippets is far cheaper than reprocessing entire conversation logs.
- Reliability — Structured memory reduces hallucinations because the agent can ground its responses in stored facts rather than guessing.
Understanding the OpenAI Agents SDK
The OpenAI Agents SDK is a framework for building agentic applications with tools, handoffs, guardrails, and structured outputs. While the SDK doesn't ship with a built-in persistent memory module out of the box, it provides the primitives — agents, tools, context objects, and lifecycle hooks — that make it straightforward to implement any memory architecture you need.
The key SDK concepts relevant to memory are:
- RunContext — A typed context object passed through every agent run, ideal for holding session-scoped memory.
- Tools — Function tools that an agent can call to store or retrieve memories.
- Session — An abstraction for conversation persistence that can be extended for custom backends.
- Lifecycle hooks — Callbacks that fire during agent execution, useful for auto-saving memories after each turn.
Setting Up Your Environment
Before diving into code, install the SDK and set up your environment. The examples below use Python, which is the primary language supported by the OpenAI Agents SDK.
pip install openai-agents python-dotenv
Create a .env file with your API key:
OPENAI_API_KEY=sk-your-key-here
Now let's set up the basic imports and configuration we'll use throughout this guide:
import os
import json
from datetime import datetime
from typing import Any, Optional
from dataclasses import dataclass, field
from dotenv import load_dotenv
from agents import Agent, Runner, function_tool, RunContextWrapper
load_dotenv()
Short-Term Memory: Managing Conversation Context
Short-term memory is the simplest form of agent memory and is largely handled by the SDK's session mechanism. The conversation history itself serves as the working memory for the current interaction. Let's look at how to manage this effectively.
Using the RunContext for Session State
The RunContextWrapper is the cleanest way to hold session-scoped data. You define a dataclass that represents your session state, and the SDK passes it to every tool and hook during the run.
@dataclass
class SessionContext:
user_id: str
conversation_summary: str = ""
key_facts: list[str] = field(default_factory=list)
turn_count: int = 0
@function_tool
def record_key_fact(ctx: RunContextWrapper[SessionContext], fact: str) -> str:
"""Record an important fact from the conversation."""
ctx.context.key_facts.append(fact)
return f"Recorded fact: {fact}"
@function_tool
def get_session_summary(ctx: RunContextWrapper[SessionContext]) -> str:
"""Retrieve the current session summary."""
return ctx.context.conversation_summary or "No summary available yet."
agent = Agent[SessionContext](
name="Assistant",
instructions="""You are a helpful assistant.
When the user shares important personal information, use the record_key_fact tool.
When you need to recall what has been discussed, use get_session_summary.""",
tools=[record_key_fact, get_session_summary],
)
Running the Agent with Context
async def run_session():
ctx = SessionContext(user_id="user_123")
result = await Runner.run(
agent,
"Hi, I'm Sarah and I work as a data scientist at a healthcare startup.",
context=ctx,
)
print(result.final_output)
print(f"Key facts recorded: {ctx.key_facts}")
# Second turn in the same session
result2 = await Runner.run(
agent,
"What do you know about me so far?",
context=ctx,
)
print(result2.final_output)
import asyncio
asyncio.run(run_session())
Notice how the SessionContext persists across multiple runs within the same session. The agent can record facts in one turn and retrieve them in the next, all without any external storage.
Long-Term Memory: Persistent Storage Across Sessions
Short-term memory disappears when the session ends. For agents that need to remember users across days, weeks, or months, you need a persistent storage layer. Let's build a memory store that saves and retrieves memories using a combination of a storage backend and semantic search.
Designing the Memory Store
A good long-term memory store needs four capabilities: writing memories, reading memories, searching memories by relevance, and managing memory lifecycle (updating, forgetting). Here's a complete implementation using a simple JSON file backend with OpenAI embeddings for semantic search.
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
class MemoryStore:
def __init__(self, storage_path: str = "memories.json"):
self.storage_path = storage_path
self.memories: list[dict] = self._load()
def _load(self) -> list[dict]:
if os.path.exists(self.storage_path):
with open(self.storage_path, "r") as f:
return json.load(f)
return []
def _save(self):
with open(self.storage_path, "w") as f:
json.dump(self.memories, f, indent=2)
async def _get_embedding(self, text: str) -> list[float]:
response = await client.embeddings.create(
model="text-embedding-3-small",
input=text,
)
return response.data[0].embedding
async def add_memory(self, user_id: str, content: str, metadata: dict = None):
embedding = await self._get_embedding(content)
memory = {
"id": f"mem_{len(self.memories)}",
"user_id": user_id,
"content": content,
"embedding": embedding,
"metadata": metadata or {},
"timestamp": datetime.now().isoformat(),
}
self.memories.append(memory)
self._save()
return memory["id"]
async def search(self, user_id: str, query: str, limit: int = 5) -> list[dict]:
query_embedding = await self._get_embedding(query)
user_memories = [m for m in self.memories if m["user_id"] == user_id]
scored = []
for mem in user_memories:
score = self._cosine_similarity(query_embedding, mem["embedding"])
scored.append((score, mem))
scored.sort(key=lambda x: x[0], reverse=True)
return [m for _, m in scored[:limit]]
@staticmethod
def _cosine_similarity(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
def delete_memory(self, memory_id: str):
self.memories = [m for m in self.memories if m["id"] != memory_id]
self._save()
# Global store instance
memory_store = MemoryStore()
Wiring the Memory Store to Agent Tools
Now we create tools that let the agent interact with the memory store. The agent decides when to save something worth remembering and when to search for relevant past context.
@dataclass
class AgentContext:
user_id: str
memory_store: MemoryStore
@function_tool
async def save_memory(ctx: RunContextWrapper[AgentContext], content: str) -> str:
"""Save an important piece of information to long-term memory.
Use this when the user shares preferences, facts, or decisions worth remembering."""
mem_id = await ctx.context.memory_store.add_memory(
user_id=ctx.context.user_id,
content=content,
metadata={"source": "conversation"},
)
return f"Saved to memory (id: {mem_id}): {content}"
@function_tool
async def recall_memory(ctx: RunContextWrapper[AgentContext], query: str) -> str:
"""Search long-term memory for information relevant to the query."""
results = await ctx.context.memory_store.search(
user_id=ctx.context.user_id,
query=query,
limit=5,
)
if not results:
return "No relevant memories found."
formatted = "\n".join(
f"- [{m['timestamp']}] {m['content']}" for m in results
)
return f"Found {len(results)} relevant memories:\n{formatted}"
memory_agent = Agent[AgentContext](
name="MemoryAgent",
instructions="""You are a personal assistant with long-term memory.
Guidelines:
- When the user shares personal information, preferences, or important decisions,
use save_memory to store it.
- When the user asks a question that might relate to past conversations,
use recall_memory to find relevant context before responding.
- Be proactive about remembering things that will help future conversations.
- Reference past memories naturally when they're relevant.""",
tools=[save_memory, recall_memory],
)
Running a Multi-Session Conversation
async def session_one():
ctx = AgentContext(user_id="user_123", memory_store=memory_store)
result = await Runner.run(
memory_agent,
"I prefer Python over JavaScript, and I'm currently learning Rust.",
context=ctx,
)
print("Session 1:", result.final_output)
async def session_two():
# New session, same user — memory persists
ctx = AgentContext(user_id="user_123", memory_store=memory_store)
result = await Runner.run(
memory_agent,
"What programming languages am I interested in?",
context=ctx,
)
print("Session 2:", result.final_output)
async def main():
await session_one()
await session_two()
asyncio.run(main())
In session two, the agent has no conversation history from session one, but it can search the memory store and recall that the user prefers Python and is learning Rust. This is the foundation of cross-session continuity.
Entity Memory: Structured Knowledge About Specific Things
While long-term memory stores free-form text, entity memory keeps structured information about specific people, projects, documents, or objects. This is especially useful for agents that work in domains where relationships between entities matter — for example, a coding agent that tracks repositories, branches, and pull requests.
Implementing an Entity Memory Layer
@dataclass
class Entity:
name: str
entity_type: str # "person", "project", "document", etc.
attributes: dict[str, Any] = field(default_factory=dict)
related_entities: list[str] = field(default_factory=list)
class EntityStore:
def __init__(self):
self.entities: dict[str, Entity] = {}
def upsert(self, entity: Entity):
self.entities[entity.name] = entity
def get(self, name: str) -> Optional[Entity]:
return self.entities.get(name)
def search_by_type(self, entity_type: str) -> list[Entity]:
return [e for e in self.entities.values() if e.entity_type == entity_type]
def get_related(self, name: str) -> list[Entity]:
entity = self.get(name)
if not entity:
return []
return [self.get(r) for r in entity.related_entities if self.get(r)]
entity_store = EntityStore()
Agent Tools for Entity Management
@dataclass
class EntityContext:
user_id: str
memory_store: MemoryStore
entity_store: EntityStore
@function_tool
def create_or_update_entity(
ctx: RunContextWrapper[EntityContext],
name: str,
entity_type: str,
attributes: str, # JSON string
) -> str:
"""Create or update a structured entity record.
Pass attributes as a JSON string like '{"role": "manager", "team": "platform"}'."""
try:
attrs = json.loads(attributes)
except json.JSONDecodeError:
return "Error: attributes must be valid JSON."
entity = Entity(name=name, entity_type=entity_type, attributes=attrs)
ctx.context.entity_store.upsert(entity)
return f"Entity '{name}' ({entity_type}) saved with attributes: {attrs}"
@function_tool
def lookup_entity(
ctx: RunContextWrapper[EntityContext],
name: str,
) -> str:
"""Look up a specific entity by name and return its details."""
entity = ctx.context.entity_store.get(name)
if not entity:
return f"No entity found with name '{name}'."
related = ctx.context.entity_store.get_related(name)
related_names = [r.name for r in related]
return json.dumps({
"name": entity.name,
"type": entity.entity_type,
"attributes": entity.attributes,
"related": related_names,
}, indent=2)
entity_agent = Agent[EntityContext](
name="EntityAgent",
instructions="""You are an assistant that tracks structured information about
people, projects, and documents. When the user mentions a person or project,
use lookup_entity to see if you already know about it. When you learn new
structured details, use create_or_update_entity to store them.""",
tools=[create_or_update_entity, lookup_entity, save_memory, recall_memory],
)
Memory Summarization and Compaction
As conversations grow long, you need a strategy to prevent the context window from overflowing. Summarization and compaction are the standard techniques. Instead of keeping every message, you periodically summarize older messages into a compact representation and keep only recent messages in full.
Implementing a Summarization Pipeline
summarizer_agent = Agent(
name="Summarizer",
instructions="""You are a summarization agent. Given a conversation history,
produce a concise summary that preserves:
- Key decisions made
- Important facts shared by the user
- Open questions or unresolved topics
- Action items
Keep the summary under 300 words.""",
model="gpt-4o-mini",
)
async def compact_conversation(
messages: list[dict],
keep_recent: int = 6,
) -> tuple[str, list[dict]]:
"""Summarize older messages and keep only recent ones in full."""
if len(messages) <= keep_recent:
return "", messages
older_messages = messages[:-keep_recent]
recent_messages = messages[-keep_recent:]
# Format older messages for the summarizer
history_text = "\n".join(
f"{m['role']}: {m['content']}" for m in older_messages
)
result = await Runner.run(
summarizer_agent,
f"Summarize this conversation:\n\n{history_text}",
)
return result.final_output, recent_messages
Integrating Compaction into a Conversation Loop
async def conversation_loop(agent, user_id: str, memory_store: MemoryStore):
messages = []
summary = ""
print("Conversation started. Type 'quit' to exit.\n")
while True:
user_input = input("You: ")
if user_input.lower() == "quit":
break
messages.append({"role": "user", "content": user_input})
# Build the prompt with summary + recent messages
context_prompt = ""
if summary:
context_prompt = f"[Previous conversation summary]\n{summary}\n\n"
full_prompt = context_prompt + user_input
ctx = AgentContext(user_id=user_id, memory_store=memory_store)
result = await Runner.run(agent, full_prompt, context=ctx)
messages.append({"role": "assistant", "content": result.final_output})
print(f"Assistant: {result.final_output}\n")
# Compact if conversation is getting long
if len(messages) > 12:
summary, messages = await compact_conversation(messages, keep_recent=6)
print(f"[Context compacted. Summary length: {len(summary)} chars]\n")
# Save final summary to long-term memory
if summary:
await memory_store.add_memory(
user_id=user_id,
content=f"Conversation summary: {summary}",
metadata={"type": "session_summary"},
)
Auto-Saving Memories with Lifecycle Hooks
Relying on the agent to decide when to save memories can be unreliable. The SDK's lifecycle hooks let you automatically extract and store important information after each agent run, without requiring the agent to explicitly call a save tool.
from agents import AgentHooks
extraction_agent = Agent(
name="MemoryExtractor",
instructions="""You are a memory extraction agent. Given a conversation turn
(user message and assistant response), extract any facts worth remembering
long-term. Return a JSON array of strings, each being a self-contained memory.
If nothing is worth remembering, return an empty array: []""",
model="gpt-4o-mini",
)
class AutoMemoryHooks(AgentHooks[AgentContext]):
def __init__(self, memory_store: MemoryStore):
self.memory_store = memory_store
async def on_agent_end(
self,
context: RunContextWrapper[AgentContext],
output: Any,
) -> None:
"""Extract and save memories after each agent run."""
# In a real implementation, you'd have access to the input and output
# Here we use the final output as a simplified example
result = await Runner.run(
extraction_agent,
f"Extract memories from this assistant response: {output}",
)
try:
memories = json.loads(result.final_output)
for mem in memories:
await self.memory_store.add_memory(
user_id=context.context.user_id,
content=mem,
metadata={"source": "auto_extraction"},
)
print(f"[Auto-saved memory]: {mem}")
except json.JSONDecodeError:
pass # Extraction didn't produce valid JSON, skip
# Attach hooks to your agent
auto_memory_agent = Agent[AgentContext](
name="SmartAssistant",
instructions="You are a helpful personal assistant.",
tools=[recall_memory],
hooks=AutoMemoryHooks(memory_store),
)
Memory Retrieval Strategies
How you retrieve memories is just as important as how you store them. Naive semantic search often returns memories that are topically similar but not actually useful for the current task. Here are several retrieval strategies to consider.
Recency-Weighted Retrieval
Combine semantic similarity with a recency boost so that recent memories are slightly favored, reflecting the intuition that recent context is often more relevant.
async def search_with_recency(
store: MemoryStore,
user_id: str,
query: str,
limit: int = 5,
recency_weight: float = 0.15,
) -> list[dict]:
results = await store.search(user_id, query, limit=limit * 2)
now = datetime.now()
scored = []
for mem in results:
mem_time = datetime.fromisoformat(mem["timestamp"])
days_old = (now - mem_time).days
recency_score = max(0, 1 - days_old / 30) # Decay over 30 days
# Re-rank: combine semantic similarity with recency
# (In practice, you'd have the similarity score from the search)
combined = recency_score * recency_weight
scored.append((combined, mem))
scored.sort(key=lambda x: x[0], reverse=True)
return [m for _, m in scored[:limit]]
Multi-Query Retrieval
Generate multiple search queries from the user's input to cast a wider net and retrieve more diverse relevant memories.
query_expansion_agent = Agent(
name="QueryExpander",
instructions="""Given a user message, generate 3 different search queries
that could find relevant past memories. Return them as a JSON array of strings.""",
model="gpt-4o-mini",
)
async def multi_query_search(
store: MemoryStore,
user_id: str,
user_message: str,
limit: int = 5,
) -> list[dict]:
result = await Runner.run(
query_expansion_agent,
f"User message: {user_message}",
)
try:
queries = json.loads(result.final_output)
except json.JSONDecodeError:
queries = [user_message]
all_results = []
seen_ids = set()
for q in queries:
results = await store.search(user_id, q, limit=limit)
for mem in results:
if mem["id"] not in seen_ids:
all_results.append(mem)
seen_ids.add(mem["id"])
return all_results[:limit]
Best Practices for Agent Memory
Building a memory system that works reliably in production requires attention to several design principles. Here are the most important best practices to follow.
- Separate storage from retrieval. Keep your memory store and your retrieval logic decoupled so you can swap out backends (JSON files, PostgreSQL, a vector database) without changing your agent code.
- Let the agent decide when to remember, but provide guardrails. Give the agent tools to save memories, but also use lifecycle hooks for automatic extraction as a safety net.
- Scope memories by user. Always tag memories with a user identifier. Cross-contaminating memories between users is a serious bug that's easy to introduce and hard to detect.
- Implement memory expiration. Not all memories should live forever. Add TTLs or periodic cleanup for transient information like session-specific debugging context.
- Deduplicate aggressively. Before saving a new memory, check if a similar one already exists. Duplicate memories pollute search results and waste context window space.
- Test memory behavior explicitly. Write integration tests that run multi-session scenarios and verify that the agent recalls the right information at the right time.
- Log memory operations. Track what gets saved, what gets retrieved, and what the agent does with it. This is invaluable for debugging unexpected agent behavior.
- Respect privacy and give users control. Provide tools for users to view, edit, and delete their memories. This is both a best practice and increasingly a legal requirement.
Implementing Deduplication
async def add_memory_dedup(
store: MemoryStore,
user_id: str,
content: str,
similarity_threshold: float = 0.85,
) -> str:
"""Add a memory only if no similar memory already exists."""
existing = await store.search(user_id, content, limit=1)
if existing:
# Re-check similarity (store.search already ranks by similarity)
# In a real implementation, you'd compare the actual similarity scores
# For this example, we check if the top result is very similar
existing_content = existing[0]["content"]
# Simple heuristic: if contents share many words, consider them duplicates
words_new = set(content.lower().split())
words_old = set(existing_content.lower().split())
if words_new and words_old:
overlap = len(words_new & words_old) / len(words_new | words_old)
if overlap > similarity_threshold:
return f"Similar memory already exists: {existing_content}"
return await store.add_memory(user_id, content)
Putting It All Together: A Complete Memory-Aware Agent
Let's combine everything into a single, cohesive example that uses short-term context, long-term memory, entity tracking, and automatic summarization.
@dataclass
class CompleteContext:
user_id: str
memory_store: MemoryStore
entity_store: EntityStore
session_summary: str = ""
@function_tool
async def save_memory_tool(ctx: RunContextWrapper[CompleteContext], content: str) -> str:
"""Save something important to long-term memory."""
mem_id = await add_memory_dedup(
ctx.context.memory_store, ctx.context.user_id, content
)
return f"Saved: {content}" if "Saved" in str(mem_id) or len(mem_id) > 20 else mem_id
@function_tool
async def recall_memory_tool(ctx: RunContextWrapper[CompleteContext], query: str) -> str:
"""Search long-term memory for relevant past information."""
results = await multi_query_search(
ctx.context.memory_store, ctx.context.user_id, query, limit=5
)
if not results:
return "No relevant memories found."
return "\n".join(f"- {m['content']}" for m in results)
@function_tool
def save_entity_tool(
ctx: RunContextWrapper[CompleteContext],
name: str,
entity_type: str,
attributes: str,
) -> str:
"""Store structured information about a person, project, or document."""
try:
attrs = json.loads(attributes)
except json.JSONDecodeError:
return "Error: attributes must be valid JSON."
entity = Entity(name=name, entity_type=entity_type, attributes=attrs)
ctx.context.entity_store.upsert(entity)
return f"Entity '{name}' saved."
@function_tool
def lookup_entity_tool(ctx: RunContextWrapper[CompleteContext], name: str) -> str:
"""Look up a stored entity by name."""
entity = ctx.context.entity_store.get(name)
if not entity:
return f"No entity named '{name}' found."
return json.dumps({
"name": entity.name,
"type": entity.entity_type,
"attributes": entity.attributes,
}, indent=2)
complete_agent = Agent[CompleteContext](
name="CompleteAssistant",
instructions="""You are an intelligent assistant with multiple memory systems.
- Use recall_memory_tool to find relevant past conversations when the user
references something that might have been discussed before.
- Use save_memory_tool when the user shares preferences, decisions, or facts
worth remembering for future sessions.
- Use save_entity_tool and lookup_entity_tool to track structured information
about people, projects, and documents the user works with.
- Be natural and proactive. Don't mention your memory systems unless asked.""",
tools=[save_memory_tool, recall_memory_tool, save_entity_tool, lookup_entity_tool],
)
async def demo():
store = MemoryStore("user_memories.json")
entities = EntityStore()
# Session 1
ctx1 = CompleteContext(user_id="user_42", memory_store=store, entity_store=entities)
r1 = await Runner.run(
complete_agent,
"I'm starting a new project called 'Atlas' with my teammate Jordan. "
"Jordan is our backend lead and prefers Go for services.",
context=ctx1,
)
print("Session 1 response:", r1.final_output)
# Session 2 — new session, same user
ctx2 = CompleteContext(user_id="user_42", memory_store=store, entity_store=entities)
r2 = await Runner.run(
complete_agent,
"Who's working on the Atlas project and what's their tech preference?",
context=ctx2,
)
print("Session 2 response:", r2.final_output)
asyncio.run(demo())
Conclusion
Memory is the backbone of any agent that needs to be useful beyond a single conversation. By layering short-term context management, long-term semantic memory, structured entity storage, and automatic summarization, you can build agents that feel genuinely intelligent and continuously helpful. The OpenAI Agents SDK provides all the primitives you need — context objects, function tools, and lifecycle hooks — and the patterns in this guide give you a production-ready foundation to build on. Start simple with a basic memory store, add complexity as your use case demands, and always test your memory behavior with realistic multi-session scenarios. The effort you invest in memory architecture will pay off in every interaction your users have with your agent.