skills/agent-memory-systems/SKILL.md
Memory is the cornerstone of intelligent agents. Without it, every interaction starts from zero. This skill covers the architecture of agent memory: short-term (context window), long-term (vector stores), and the cognitive architectures that organize them.
npx skillsauth add Regtransfers/agency-agents-mcp agent-memory-systemsInstall this skill globally with one command. Works with Claude Code, Cursor, and Windsurf.
3 of 9 scanners reported clean
Some scanners were skipped, did not run, or reported a non-clean status. Review each row below.
@ Agent Memory Systems
Memory is the cornerstone of intelligent agents. Without it, every interaction starts from zero. This skill covers the architecture of agent memory: short-term (context window), long-term (vector stores), and the cognitive architectures that organize them.
Key insight: Memory isn't just storage - it's retrieval. A million stored facts mean nothing if can't find the right one. Chunking, embedding, and retrieval strategies determine whether your agent remembers or forgets.
The field is fragmented with inconsistent terminology. We use the CoALA cognitive architecture framework: semantic memory (facts), episodic memory (experiences), and procedural memory (how-to knowledge).
@ Principles
@ Capabilities
@ Scope
@ Tooling
@ Memory_frameworks
@ Vector_stores
@ Embedding_models
@ Patterns
@ Memory Type Architecture
Choosing the right memory type for different information
When to use: Designing agent memory system
@ MEMORY TYPE ARCHITECTURE (CoALA Framework):
""" Three memory types for different purposes:
@ LangMem Implementation """ from langmem import MemoryStore from langgraph.graph import StateGraph
@ Initialize memory store memory = MemoryStore( connectionstring=os.environ["POSTGRESURL"] )
@ Semantic memory: user profile await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["dark mode", "concise responses"], "expertise_level": "developer", } )
@ Episodic memory: past interaction await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "Helped debug authentication issue", "outcome": "resolved", "key_insights": ["Token expiry was root cause"], }, metadata={"userid": userid, "topic": "debugging"} )
@ Procedural memory: learned pattern await memory.procedural.add( namespace="skills", content={ "tasktype": "debugauth", "steps": ["Check token expiry", "Verify refresh flow"], "exampleinteraction": fewshot_example, } ) """
@ Memory Retrieval at Runtime """ async def preparecontext(userid, query): @ Get user profile (semantic) profile = await memory.semantic.get( namespace="user_profile", key=user_id )
@ Find relevant past experiences (episodic) similar_experiences = await memory.episodic.search( namespace="conversations", query=query, filter={"userid": userid}, limit=3 )
@ Find relevant skills (procedural) relevant_skills = await memory.procedural.search( namespace="skills", query=query, limit=2 )
return { "profile": profile, "pastexperiences": similarexperiences, "relevantskills": relevantskills, } """
@ Vector Store Selection Pattern
Choosing the right vector database for your use case
When to use: Setting up persistent memory storage
@ VECTOR STORE SELECTION:
""" Decision matrix:
Pinecone; Qdrant; Weaviate; ChromaDB; pgvector
Scale; Billions; 100M+; 100M+; 1M; 1M Managed; Yes; Both; Both; Self; Self Filtering; Basic; Best; Good; Basic; SQL Hybrid; No; Yes; Best; No; Yes Cost; High; Medium; Medium; Free; Free Latency; 5ms; 7ms; 10ms; 20ms; 15ms """
@ Pinecone (Enterprise Scale) """ from pinecone import Pinecone
pc = Pinecone(apikey=os.environ["PINECONEAPI_KEY"]) index = pc.Index("agent-memory")
@ Upsert with metadata index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "userid": userid, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=namespace )
@ Query with filter results = index.query( vector=query_embedding, filter={"userid": userid, "type": "episodic"}, top_k=5, include_metadata=True ) """
@ Qdrant (Complex Filtering) """ from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition
client = QdrantClient(url="http://localhost:6333")
@ Complex filtering with Qdrant results = client.search( collectionname="agentmemory", queryvector=queryembedding, query_filter=Filter( must=[ FieldCondition(key="userid", match={"value": userid}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """
@ ChromaDB (Prototyping) """ import chromadb
client = chromadb.PersistentClient(path="./memory_db") collection = client.getorcreatecollection("agentmemory")
@ Simple and fast for prototypes collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"userid": userid, "type": "episodic"}] )
results = collection.query( queryembeddings=[queryembedding], n_results=5, where={"userid": userid} ) """
@ Chunking Strategy Pattern
Breaking documents into retrievable chunks
When to use: Processing documents for memory storage
@ CHUNKING STRATEGIES:
""" The chunking dilemma:
Optimal chunk size depends on:
General guidance: 256-512 tokens for most use cases """
@ Fixed-Size Chunking (Baseline) """ from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter( chunk_size=500, # Characters chunk_overlap=50, # Overlap prevents cutting sentences separators=["\n\n", "\n", ". ", " ", ""] # Priority order )
chunks = splitter.split_text(document) """
@ Semantic Chunking (Better Quality) """ from langchainexperimental.textsplitter import SemanticChunker from langchain_openai import OpenAIEmbeddings
@ Splits based on semantic similarity splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpointthresholdtype="percentile", breakpointthresholdamount=95 )
chunks = splitter.split_text(document) """
@ Structure-Aware Chunking (Documents with Hierarchy) """ from langchain.text_splitter import MarkdownHeaderTextSplitter
@ Respect document structure splitter = MarkdownHeaderTextSplitter( headerstosplit_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] )
chunks = splitter.splittext(markdowndoc) @ Each chunk has header metadata for context """
@ Contextual Chunking (Anthropic's Approach) """ @ Add context to each chunk before embedding @ Reduces retrieval failures by 35%
def addcontexttochunk(chunk, documentsummary): context_prompt = f''' Document summary: {document_summary}
The following is a chunk from this document: {chunk} ''' return context_prompt
@ Embed the contextualized chunk, not raw chunk for chunk in chunks: contextualized = addcontextto_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # Store original, embed contextualized """
@ Code-Specific Chunking """ from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
@ Language-aware splitting pythonsplitter = RecursiveCharacterTextSplitter.fromlanguage( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )
@ Respects function/class boundaries chunks = pythonsplitter.splittext(python_code) """
@ Background Memory Formation
Processing memories asynchronously for better quality
When to use: You want higher recall without slowing interactions
@ BACKGROUND MEMORY FORMATION:
""" Real-time memory extraction slows conversations and adds complexity to agent tool calls. Background processing after conversations yields higher quality memories.
Pattern: Subconscious memory formation """
@ LangGraph Background Processing """ from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver
async def backgroundmemoryprocessor(thread_id: str): @ Run after conversation ends or goes idle conversation = await loadconversation(threadid)
@ Extract insights without time pressure insights = await llm.invoke(''' Analyze this conversation and extract:
Be thorough - this runs in background.
Conversation: {conversation} ''')
@ Store to long-term memory for insight in insights: await memory.semantic.upsert( namespace="user_insights", key=generate_key(insight), content=insight, metadata={"sourcethread": threadid} )
@ Trigger on conversation end or idle timeout @onconversationidle(timeout_minutes=5) async def processconversation(threadid): await backgroundmemoryprocessor(thread_id) """
@ Memory Consolidation (Like Sleep) """ @ Periodically consolidate and deduplicate memories
async def consolidatememories(userid: str): @ Get all memories for user memories = await memory.semantic.list( namespace="user_insights", filter={"userid": userid} )
@ Find similar memories (potential duplicates) clusters = clusterbysimilarity(memories, threshold=0.9)
@ Merge similar memories for cluster in clusters: if len(cluster) > 1: merged = await llm.invoke(f''' Consolidate these related memories into one: {cluster}
Preserve all important information. ''') await memory.semantic.upsert( namespace="user_insights", key=generate_key(merged), content=merged ) @ Delete originals for old in cluster: await memory.semantic.delete(old.id) """
@ Memory Decay Pattern
Forgetting old, irrelevant memories
When to use: Memory grows large, retrieval slows down
@ MEMORY DECAY:
""" Not all memories should live forever:
Implement intelligent decay based on:
@ Time-Based Decay """ from datetime import datetime, timedelta
async def decayoldmemories(namespace: str, maxagedays: int): cutoff = datetime.now() - timedelta(days=maxagedays)
old_memories = await memory.episodic.list( namespace=namespace, filter={"last_accessed": {"$lt": cutoff.isoformat()}} )
for mem in old_memories: @ Soft delete (mark as archived) await memory.episodic.update( id=mem.id, metadata={"archived": True, "archived_at": datetime.now()} ) """
@ Utility-Based Decay (MIRIX Approach) """ def calculatememoryutility(memory): ''' Composite utility score inspired by cognitive science:
@ Recency score (exponential decay with 72h half-life) hourssinceaccess = (now - memory.lastaccessed).totalseconds() / 3600 recencyscore = 0.5 ** (hourssince_access / 72)
@ Frequency score frequencyscore = min(memory.accesscount / 10, 1.0)
@ Importance (from metadata or heuristic) importance = memory.metadata.get("importance", 0.5)
@ Weighted combination utility = ( 0.4 * recency_score + 0.3 * frequency_score + 0.3 * importance )
return utility
async def prunelowutility_memories(threshold=0.2): allmemories = await memory.listall() for mem in all_memories: if calculatememoryutility(mem) < threshold: await memory.archive(mem.id) """
@ Sharp Edges
@ Chunking Isolates Information From Its Context
Severity: CRITICAL
Situation: Processing documents for vector storage
Symptoms: Retrieval finds chunks but they don't make sense alone. Agent answers miss the big picture. "The function returns X" retrieved without knowing which function. References to "this" without knowing what "this" refers to.
Why this breaks: When we chunk for AI processing, we're breaking connections, reducing a holistic narrative to isolated fragments that often miss the big picture. A chunk about "the configuration" without context about what system is being configured is nearly useless.
Recommended fix:
@ Contextual Chunking (Anthropic's approach) @ Add document context to each chunk before embedding @ Reduces retrieval failures by 35%
def contextualize_chunk(chunk, document): summary = summarize(document)
@ LLM generates context for chunk context = llm.invoke(f''' Document summary: {summary}
Generate a brief context statement for this chunk that would help someone understand what it refers to:
{chunk} ''')
return f"{context}\n\n{chunk}"
@ Embed the contextualized version for chunk in chunks: contextualized = contextualizechunk(chunk, fulldoc) embedding = embed(contextualized) @ Store original chunk, embed contextualized store(original=chunk, embedding=embedding)
@ Hierarchical Chunking @ Store at multiple granularities chunks_small = split(doc, size=256) chunks_medium = split(doc, size=512) chunks_large = split(doc, size=1024)
@ Retrieve at appropriate level based on query
@ Chunk Size Mismatched to Query Patterns
Severity: HIGH
Situation: Configuring chunking for memory storage
Symptoms: High-quality documents produce low-quality retrievals. Simple questions miss relevant information. Complex questions get fragments instead of complete answers.
Why this breaks: Optimal chunk size depends on query patterns:
The sweet spot varies by document type and embedding model. Default 1000 characters works for nothing specific.
Recommended fix:
@ Test different sizes from sklearn.metrics import recall_score
def evaluatechunksize(documents, testqueries, chunksize): chunks = splitdocuments(documents, size=chunksize) index = build_index(chunks)
correct_retrievals = 0 for query, expectedchunk in testqueries: results = index.search(query, k=5) if expected_chunk in results: correct_retrievals += 1
return correctretrievals / len(testqueries)
@ Test multiple sizes for size in [256, 512, 768, 1024]: recall = evaluatechunksize(docs, test_queries, size) print(f"Size {size}: Recall@5 = {recall:.2%}")
@ Size recommendations by content type CHUNK_SIZES = { "documentation": 512, # Complete concepts "code": 1000, # Function-level "conversation": 256, # Turn-level "articles": 768, # Paragraph-level }
@ Use overlap to prevent boundary issues splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=50, # 10% overlap )
@ Semantic Search Returns Irrelevant Results
Severity: HIGH
Situation: Querying memory for context
Symptoms: Agent retrieves memories that seem related but aren't useful. "Tell me about the user's preferences" returns conversation about preferences in general, not this user's. High similarity scores for wrong content.
Why this breaks: Semantic similarity isn't the same as relevance. "The user likes Python" and "Python is a programming language" are semantically similar but very different types of information. Without metadata filtering, retrieval is just word matching.
Recommended fix:
@ Always filter by metadata first @ Don't rely on semantic similarity alone
@ Bad: Only semantic search results = index.query( vector=query_embedding, top_k=5 )
@ Good: Filter then search results = index.query( vector=query_embedding, filter={ "userid": currentuser.id, "type": "preference", "createdafter": cutoffdate, }, top_k=5 )
@ Use hybrid search (semantic + keyword) from qdrant_client import QdrantClient
client = QdrantClient(...)
@ Hybrid search with fusion results = client.search( collection_name="memories", queryvector=semanticembedding, query_text=query, # Also keyword match fusion={"method": "rrf"}, # Reciprocal Rank Fusion )
@ Rerank results with cross-encoder from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
@ Initial retrieval (recall-oriented) candidates = index.query(queryembedding, topk=20)
@ Rerank (precision-oriented) pairs = [(query, c.text) for c in candidates] scores = reranker.predict(pairs) reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
@ Old Memories Override Current Information
Severity: HIGH
Situation: User preferences or facts change over time
Symptoms: Agent uses outdated preferences. "User prefers dark mode" from 6 months ago overrides recent "switch to light mode" request. Agent confidently uses stale data.
Why this breaks: Vector stores don't have temporal awareness by default. A memory from a year ago has the same retrieval weight as one from today. Recent information should generally override old information for preferences and mutable facts.
Recommended fix:
@ Add temporal scoring from datetime import datetime, timedelta
def timedecayscore(memory, halflifedays=30): age = (datetime.now() - memory.created_at).days decay = 0.5 ** (age / halflifedays) return decay
def retrievewithrecency(query, user_id): @ Get candidates candidates = index.query( vector=embed(query), filter={"userid": userid}, top_k=20 )
@ Apply time decay for candidate in candidates: timescore = timedecay_score(candidate) candidate.finalscore = candidate.similarity 0.7 + timescore 0.3
@ Re-sort by final score return sorted(candidates, key=lambda x: x.final_score, reverse=True)[:5]
@ Update instead of append for preferences async def updatepreference(userid, category, value): @ Delete old preference await memory.delete( filter={"userid": userid, "type": "preference", "category": category} )
@ Store new preference await memory.upsert( id=f"pref-{user_id}-{category}", content={"category": category, "value": value}, metadata={"updated_at": datetime.now()} )
@ Explicit versioning for facts await memory.upsert( id=f"fact-{fact_id}-v{version}", content=new_fact, metadata={ "version": version, "supersedes": previous_id, "valid_from": datetime.now() } )
@ Contradictory Memories Retrieved Together
Severity: MEDIUM
Situation: User has changed preferences or provided conflicting info
Symptoms: Agent retrieves "user prefers dark mode" and "user prefers light mode" in same context. Gives inconsistent answers. Seems confused or forgetful to user.
Why this breaks: Without conflict resolution, both old and new information coexist. Semantic search might return both because they're both about the same topic (preferences). Agent has no way to know which is current.
Recommended fix:
@ Detect conflicts on storage async def storewithconflictcheck(memory, userid): @ Find potentially conflicting memories similar = await index.query( vector=embed(memory.content), filter={"userid": userid, "type": memory.type}, threshold=0.9, # Very similar top_k=5 )
for existing in similar: if is_contradictory(memory.content, existing.content): @ Ask for resolution resolution = await resolve_conflict(memory, existing) if resolution == "replace": await index.delete(existing.id) elif resolution == "version": await mark_superseded(existing.id, memory.id)
await index.upsert(memory)
@ Conflict detection heuristic def iscontradictory(newcontent, old_content): @ Use LLM to detect contradiction result = llm.invoke(f''' Do these two statements contradict each other?
Statement 1: {old_content} Statement 2: {new_content}
Respond with just YES or NO. ''') return result.strip().upper() == "YES"
@ Periodic consolidation async def consolidatememories(userid): allmemories = await index.list(filter={"userid": user_id}) clusters = clusterbytopic(all_memories)
for cluster in clusters: if has_conflicts(cluster): resolved = await llm.invoke(f''' These memories may conflict. Create one consolidated memory that represents the current truth: {cluster} ''') await replace_cluster(cluster, resolved)
@ Retrieved Memories Exceed Context Window
Severity: MEDIUM
Situation: Retrieving too many memories at once
Symptoms: Token limit errors. Agent truncates important information. System prompt gets cut off. Retrieved memories compete with user query for space.
Why this breaks: Retrieval typically returns top-k results. If k is too high or chunks are too large, retrieved context overwhelms the window. Critical information (system prompt, recent messages) gets pushed out.
Recommended fix:
@ Budget tokens for different memory types TOKEN_BUDGET = { "system_prompt": 500, "user_profile": 200, "recent_messages": 2000, "retrieved_memories": 1000, "current_query": 500, "buffer": 300, # Safety margin }
def budgetawareretrieval(query, context_limit=4000): remaining = contextlimit - TOKENBUDGET["systemprompt"] - TOKENBUDGET["buffer"]
@ Prioritize recent messages recent = getrecentmessages(limit=TOKENBUDGET["recentmessages"]) remaining -= count_tokens(recent)
@ Then user profile profile = getuserprofile(limit=TOKENBUDGET["userprofile"]) remaining -= count_tokens(profile)
@ Finally retrieved memories with remaining budget memories = retrievememories(query, maxtokens=remaining)
return build_context(profile, recent, memories)
@ Dynamic k based on chunk size def retrievewithbudget(query, max_tokens=1000): avgchunktokens = 150 # From your data maxk = maxtokens // avgchunktokens
results = index.query(query, topk=maxk)
@ Trim if still over budget total_tokens = 0 filtered = [] for result in results: tokens = count_tokens(result.text) if totaltokens + tokens <= maxtokens: filtered.append(result) total_tokens += tokens else: break
return filtered
@ Query and Document Embeddings From Different Models
Severity: MEDIUM
Situation: Upgrading embedding model or mixing providers
Symptoms: Retrieval quality suddenly drops. Relevant documents not found. Random results returned. Works for new documents, fails for old.
Why this breaks: Embedding models produce different vector spaces. A query embedded with text-embedding-3 won't match documents embedded with text-ada-002. Mixing models creates garbage similarity scores.
Recommended fix:
@ Track embedding model in metadata await index.upsert( id=doc_id, vector=embedding, metadata={ "embedding_model": "text-embedding-3-small", "embedding_version": "2024-01", "content": content } )
@ Filter by model version on retrieval results = index.query( vector=query_embedding, filter={"embeddingmodel": currentmodel}, top_k=10 )
@ Migration strategy for model upgrades async def migrateembeddings(oldmodel, new_model): @ Get all documents with old model olddocs = await index.list(filter={"embeddingmodel": old_model})
for doc in old_docs: @ Re-embed with new model newembedding = await embed(doc.content, model=newmodel)
@ Update in place await index.update( id=doc.id, vector=new_embedding, metadata={"embeddingmodel": newmodel} )
@ Use separate collections during migration @ Old collection: production queries @ New collection: re-embedding in progress @ Switch over when complete
@ Validation Checks
@ In-Memory Store in Production Code
Severity: ERROR
In-memory stores lose data on restart
Message: In-memory store detected. Use persistent storage (Postgres, Qdrant, Pinecone) for production.
@ Vector Upsert Without Metadata
Severity: WARNING
Vectors should have metadata for filtering
Message: Vector upsert without metadata. Add user_id, type, timestamp for proper filtering.
@ Query Without User Filtering
Severity: ERROR
Queries should filter by user to prevent data leakage
Message: Vector query without user filtering. Always filter by user_id to prevent data leakage.
@ Hardcoded Chunk Size Without Justification
Severity: INFO
Chunk size should be tested and justified
Message: Hardcoded chunk size. Test different sizes for your content type and measure retrieval accuracy.
@ Chunking Without Overlap
Severity: WARNING
Chunk overlap prevents boundary issues
Message: Text splitting without overlap. Add chunk_overlap (10-20%) to prevent boundary issues.
@ Semantic Search Without Filters
Severity: WARNING
Pure semantic search often returns irrelevant results
Message: Pure semantic search. Add metadata filters (user, type, time) for better relevance.
@ Retrieval Without Result Limit
Severity: WARNING
Unbounded retrieval can overflow context
Message: Retrieval without limit. Set top_k to prevent context overflow.
@ Embeddings Without Model Version Tracking
Severity: WARNING
Track embedding model to handle migrations
Message: Store embedding model version in metadata to handle model migrations.
@ Different Models for Document and Query Embedding
Severity: ERROR
Documents and queries must use same embedding model
Message: Ensure same embedding model for indexing and querying.
@ Collaboration
@ Delegation Triggers
@ Related Skills
Works well with: autonomous-agents, multi-agent-orchestration, llm-architect, agent-tool-builder
@ When to Use
@ Limitations
tools
Build AI agents that interact with computers like humans do - viewing screens, moving cursors, clicking buttons, and typing text. Covers Anthropic's Computer Use, OpenAI's Operator/CUA, and open-source alternatives.
testing
Generate structured PR descriptions from diffs, add review checklists, risk assessments, and test coverage summaries. Use when the user says "write a PR description", "improve this PR", "summarize my changes", "PR review", "pull request", or asks to document a diff for reviewers.
tools
Use when working with comprehensive review full review
development
You are an expert in creating competitor comparison and alternative pages. Your goal is to build pages that rank for competitive search terms, provide genuine value to evaluators, and position your product effectively.