Memory is the cornerstone of intelligent agents. Without it, every interaction\nstarts from zero. This skill covers the architecture of agent memory: short-term\n(context window), long-term (vector stores), and the cognitive architectures\nthat organize them.
\nKey insight: Memory isn't just storage - it's retrieval. A million stored facts\nmean nothing if you can't find the right one. Chunking, embedding, and retrieval\nstrategies determine whether your agent remembers or forgets.
\nThe field is fragmented with inconsistent terminology. We use the CoALA cognitive\narchitecture framework: semantic memory (facts), episodic memory (experiences),\nand procedural memory (how-to knowledge).
\nChoosing the right memory type for different information
\nWhen to use: Designing agent memory system
\n\"\"\"\nThree memory types for different purposes:
\nSemantic Memory: Facts and knowledge
\nEpisodic Memory: Experiences and events
\nProcedural Memory: How to do things
\n\"\"\"\nfrom langmem import MemoryStore\nfrom langgraph.graph import StateGraph
\nmemory = MemoryStore(\nconnection_string=os.environ[\"POSTGRES_URL\"]\n)
\nawait memory.semantic.upsert(\nnamespace=\"user_profile\",\nkey=user_id,\ncontent={\n\"name\": \"Alice\",\n\"preferences\": [\"dark mode\", \"concise responses\"],\n\"expertise_level\": \"developer\",\n}\n)
\nawait memory.episodic.add(\nnamespace=\"conversations\",\ncontent={\n\"timestamp\": datetime.now(),\n\"summary\": \"Helped debug authentication issue\",\n\"outcome\": \"resolved\",\n\"key_insights\": [\"Token expiry was root cause\"],\n},\nmetadata={\"user_id\": user_id, \"topic\": \"debugging\"}\n)
\nawait memory.procedural.add(\nnamespace=\"skills\",\ncontent={\n\"task_type\": \"debug_auth\",\n\"steps\": [\"Check token expiry\", \"Verify refresh flow\"],\n\"example_interaction\": few_shot_example,\n}\n)\n\"\"\"
\n\"\"\"\nasync def prepare_context(user_id, query):\n# Get user profile (semantic)\nprofile = await memory.semantic.get(\nnamespace=\"user_profile\",\nkey=user_id\n)
\n# Find relevant past experiences (episodic)\nsimilar_experiences = await memory.episodic.search(\n namespace=\"conversations\",\n query=query,\n filter={\"user_id\": user_id},\n limit=3\n)\n\n# Find relevant skills (procedural)\nrelevant_skills = await memory.procedural.search(\n namespace=\"skills\",\n query=query,\n limit=2\n)\n\nreturn {\n \"profile\": profile,\n \"past_experiences\": similar_experiences,\n \"relevant_skills\": relevant_skills,\n}\n\"\"\"
\nChoosing the right vector database for your use case
\nWhen to use: Setting up persistent memory storage
\n\"\"\"\nDecision matrix:
\n| | Pinecone | Qdrant | Weaviate | ChromaDB | pgvector |\n|------------|----------|--------|----------|----------|----------|\n| Scale | Billions | 100M+ | 100M+ | 1M | 1M |\n| Managed | Yes | Both | Both | Self | Self |\n| Filtering | Basic | Best | Good | Basic | SQL |\n| Hybrid | No | Yes | Best | No | Yes |\n| Cost | High | Medium | Medium | Free | Free |\n| Latency | 5ms | 7ms | 10ms | 20ms | 15ms |\n\"\"\"
\n\"\"\"\nfrom pinecone import Pinecone
\npc = Pinecone(api_key=os.environ[\"PINECONE_API_KEY\"])\nindex = pc.Index(\"agent-memory\")
\nindex.upsert(\nvectors=[\n{\n\"id\": f\"memory-{uuid4()}\",\n\"values\": embedding,\n\"metadata\": {\n\"user_id\": user_id,\n\"timestamp\": datetime.now().isoformat(),\n\"type\": \"episodic\",\n\"content\": memory_text,\n}\n}\n],\nnamespace=namespace\n)
\nresults = index.query(\nvector=query_embedding,\nfilter={\"user_id\": user_id, \"type\": \"episodic\"},\ntop_k=5,\ninclude_metadata=True\n)\n\"\"\"
\n\"\"\"\nfrom qdrant_client import QdrantClient\nfrom qdrant_client.models import PointStruct, Filter, FieldCondition
\nclient = QdrantClient(url=\"http://localhost:6333\")
\nresults = client.search(\ncollection_name=\"agent_memory\",\nquery_vector=query_embedding,\nquery_filter=Filter(\nmust=[\nFieldCondition(key=\"user_id\", match={\"value\": user_id}),\nFieldCondition(key=\"type\", match={\"value\": \"semantic\"}),\n],\nshould=[\nFieldCondition(key=\"topic\", match={\"any\": [\"auth\", \"security\"]}),\n]\n),\nlimit=5\n)\n\"\"\"
\n\"\"\"\nimport chromadb
\nclient = chromadb.PersistentClient(path=\"./memory_db\")\ncollection = client.get_or_create_collection(\"agent_memory\")
\ncollection.add(\nids=[str(uuid4())],\nembeddings=[embedding],\ndocuments=[memory_text],\nmetadatas=[{\"user_id\": user_id, \"type\": \"episodic\"}]\n)
\nresults = collection.query(\nquery_embeddings=[query_embedding],\nn_results=5,\nwhere={\"user_id\": user_id}\n)\n\"\"\"
\nBreaking documents into retrievable chunks
\nWhen to use: Processing documents for memory storage
\n\"\"\"\nThe chunking dilemma:
\nOptimal chunk size depends on:
\nGeneral guidance: 256-512 tokens for most use cases\n\"\"\"
\n\"\"\"\nfrom langchain.text_splitter import RecursiveCharacterTextSplitter
\nsplitter = RecursiveCharacterTextSplitter(\nchunk_size=500, # Characters\nchunk_overlap=50, # Overlap prevents cutting sentences\nseparators=[\"\\n\\n\", \"\\n\", \". \", \" \", \"\"] # Priority order\n)
\nchunks = splitter.split_text(document)\n\"\"\"
\n\"\"\"\nfrom langchain_experimental.text_splitter import SemanticChunker\nfrom langchain_openai import OpenAIEmbeddings
\nsplitter = SemanticChunker(\nembeddings=OpenAIEmbeddings(),\nbreakpoint_threshold_type=\"percentile\",\nbreakpoint_threshold_amount=95\n)
\nchunks = splitter.split_text(document)\n\"\"\"
\n\"\"\"\nfrom langchain.text_splitter import MarkdownHeaderTextSplitter
\nsplitter = MarkdownHeaderTextSplitter(\nheaders_to_split_on=[\n(\"#\", \"Header 1\"),\n(\"##\", \"Header 2\"),\n(\"###\", \"Header 3\"),\n]\n)
\nchunks = splitter.split_text(markdown_doc)
\n\"\"\"
\n\"\"\"
\ndef add_context_to_chunk(chunk, document_summary):\ncontext_prompt = f'''\nDocument summary:
\nThe following is a chunk from this document:\n{chunk}\n'''\nreturn context_prompt\nfor chunk in chunks:\ncontextualized = add_context_to_chunk(chunk, summary)\nembedding = embed(contextualized)\nstore(chunk, embedding) # Store original, embed contextualized\n\"\"\"
\n\"\"\"\nfrom langchain.text_splitter import Language, RecursiveCharacterTextSplitter
\npython_splitter = RecursiveCharacterTextSplitter.from_language(\nlanguage=Language.PYTHON,\nchunk_size=1000,\nchunk_overlap=200\n)
\nchunks = python_splitter.split_text(python_code)\n\"\"\"
\nProcessing memories asynchronously for better quality
\nWhen to use: You want higher recall without slowing interactions
\n\"\"\"\nReal-time memory extraction slows conversations and adds\ncomplexity to agent tool calls. Background processing after\nconversations yields higher quality memories.
\nPattern: Subconscious memory formation\n\"\"\"
\n\"\"\"\nfrom langgraph.graph import StateGraph\nfrom langgraph.checkpoint.postgres import PostgresSaver
\nasync def background_memory_processor(thread_id: str):\n# Run after conversation ends or goes idle\nconversation = await load_conversation(thread_id)
\n# Extract insights without time pressure\ninsights = await llm.invoke('''\n Analyze this conversation and extract:\n 1. Key facts learned about the user\n 2. User preferences revealed\n 3. Tasks completed or pending\n 4. Patterns in user behavior\n\n Be thorough - this runs in background.\n\n Conversation:\n {conversation}\n''')\n\n# Store to long-term memory\nfor insight in insights:\n await memory.semantic.upsert(\n namespace=\"user_insights\",\n key=generate_key(insight),\n content=insight,\n metadata={\"source_thread\": thread_id}\n )\n@on_conversation_idle(timeout_minutes=5)\nasync def process_conversation(thread_id):\nawait background_memory_processor(thread_id)\n\"\"\"
\n\"\"\"
\nasync def consolidate_memories(user_id: str):\n# Get all memories for user\nmemories = await memory.semantic.list(\nnamespace=\"user_insights\",\nfilter={\"user_id\": user_id}\n)
\n# Find similar memories (potential duplicates)\nclusters = cluster_by_similarity(memories, threshold=0.9)\n\n# Merge similar memories\nfor cluster in clusters:\n if len(cluster) > 1:\n merged = await llm.invoke(f'''\n Consolidate these related memories into one:\n {cluster}\n\n Preserve all important information.\n ''')\n await memory.semantic.upsert(\n namespace=\"user_insights\",\n key=generate_key(merged),\n content=merged\n )\n # Delete originals\n for old in cluster:\n await memory.semantic.delete(old.id)\n\"\"\"
\nForgetting old, irrelevant memories
\nWhen to use: Memory grows large, retrieval slows down
\n\"\"\"\nNot all memories should live forever:
\nImplement intelligent decay based on:
\n\"\"\"\nfrom datetime import datetime, timedelta
\nasync def decay_old_memories(namespace: str, max_age_days: int):\ncutoff = datetime.now() - timedelta(days=max_age_days)
\nold_memories = await memory.episodic.list(\n namespace=namespace,\n filter={\"last_accessed\": {\"$lt\": cutoff.isoformat()}}\n)\n\nfor mem in old_memories:\n # Soft delete (mark as archived)\n await memory.episodic.update(\n id=mem.id,\n metadata={\"archived\": True, \"archived_at\": datetime.now()}\n )\n\"\"\"
\n\"\"\"\ndef calculate_memory_utility(memory):\n'''\nComposite utility score inspired by cognitive science:\n- Recency: When was it last accessed?\n- Frequency: How often is it accessed?\n- Importance: How critical is this information?\n'''\nnow = datetime.now()
\n# Recency score (exponential decay with 72h half-life)\nhours_since_access = (now - memory.last_accessed).total_seconds() / 3600\nrecency_score = 0.5 ** (hours_since_access / 72)\n\n# Frequency score\nfrequency_score = min(memory.access_count / 10, 1.0)\n\n# Importance (from metadata or heuristic)\nimportance = memory.metadata.get(\"importance\", 0.5)\n\n# Weighted combination\nutility = (\n 0.4 * recency_score +\n 0.3 * frequency_score +\n 0.3 * importance\n)\n\nreturn utility\nasync def prune_low_utility_memories(threshold=0.2):\nall_memories = await memory.list_all()\nfor mem in all_memories:\nif calculate_memory_utility(mem) < threshold:\nawait memory.archive(mem.id)\n\"\"\"
\nSeverity: CRITICAL
\nSituation: Processing documents for vector storage
\nSymptoms:\nRetrieval finds chunks but they don't make sense alone. Agent\nanswers miss the big picture. \"The function returns X\" retrieved\nwithout knowing which function. References to \"this\" without\nknowing what \"this\" refers to.
\nWhy this breaks:\nWhen we chunk for AI processing, we're breaking connections,\nreducing a holistic narrative to isolated fragments that often\nmiss the big picture. A chunk about \"the configuration\" without\ncontext about what system is being configured is nearly useless.
\nRecommended fix:
\ndef contextualize_chunk(chunk, document):\nsummary = summarize(document)
\n# LLM generates context for chunk\ncontext = llm.invoke(f'''\n Document summary: {summary}\n\n Generate a brief context statement for this chunk\n that would help someone understand what it refers to:\n\n {chunk}\n''')\n\nreturn f\"{context}\\n\\n{chunk}\"\nfor chunk in chunks:\ncontextualized = contextualize_chunk(chunk, full_doc)\nembedding = embed(contextualized)\n# Store original chunk, embed contextualized\nstore(original=chunk, embedding=embedding)
\nchunks_small = split(doc, size=256)\nchunks_medium = split(doc, size=512)\nchunks_large = split(doc, size=1024)
\nSeverity: HIGH
\nSituation: Configuring chunking for memory storage
\nSymptoms:\nHigh-quality documents produce low-quality retrievals. Simple\nquestions miss relevant information. Complex questions get\nfragments instead of complete answers.
\nWhy this breaks:\nOptimal chunk size depends on query patterns:
\nThe sweet spot varies by document type and embedding model.\nDefault 1000 characters works for nothing specific.
\nRecommended fix:
\nfrom sklearn.metrics import recall_score
\ndef evaluate_chunk_size(documents, test_queries, chunk_size):\nchunks = split_documents(documents, size=chunk_size)\nindex = build_index(chunks)
\ncorrect_retrievals = 0\nfor query, expected_chunk in test_queries:\n results = index.search(query, k=5)\n if expected_chunk in results:\n correct_retrievals += 1\n\nreturn correct_retrievals / len(test_queries)\nfor size in [256, 512, 768, 1024]:\nrecall = evaluate_chunk_size(docs, test_queries, size)\nprint(f\"Size : Recall@5 = {recall:.2%}\")
\nCHUNK_SIZES = {\n\"documentation\": 512, # Complete concepts\n\"code\": 1000, # Function-level\n\"conversation\": 256, # Turn-level\n\"articles\": 768, # Paragraph-level\n}
\nsplitter = RecursiveCharacterTextSplitter(\nchunk_size=512,\nchunk_overlap=50, # 10% overlap\n)
\nSeverity: HIGH
\nSituation: Querying memory for context
\nSymptoms:\nAgent retrieves memories that seem related but aren't useful.\n\"Tell me about the user's preferences\" returns conversation\nabout preferences in general, not this user's. High similarity\nscores for wrong content.
\nWhy this breaks:\nSemantic similarity isn't the same as relevance. \"The user\nlikes Python\" and \"Python is a programming language\" are\nsemantically similar but very different types of information.\nWithout metadata filtering, retrieval is just word matching.
\nRecommended fix:
\nresults = index.query(\nvector=query_embedding,\ntop_k=5\n)
\nresults = index.query(\nvector=query_embedding,\nfilter={\n\"user_id\": current_user.id,\n\"type\": \"preference\",\n\"created_after\": cutoff_date,\n},\ntop_k=5\n)
\nfrom qdrant_client import QdrantClient
\nclient = QdrantClient(...)
\nresults = client.search(\ncollection_name=\"memories\",\nquery_vector=semantic_embedding,\nquery_text=query, # Also keyword match\nfusion={\"method\": \"rrf\"}, # Reciprocal Rank Fusion\n)
\nfrom sentence_transformers import CrossEncoder
\nreranker = CrossEncoder(\"cross-encoder/ms-marco-MiniLM-L-6-v2\")
\ncandidates = index.query(query_embedding, top_k=20)
\npairs = [(query, c.text) for c in candidates]\nscores = reranker.predict(pairs)\nreranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
\nSeverity: HIGH
\nSituation: User preferences or facts change over time
\nSymptoms:\nAgent uses outdated preferences. \"User prefers dark mode\" from\n6 months ago overrides recent \"switch to light mode\" request.\nAgent confidently uses stale data.
\nWhy this breaks:\nVector stores don't have temporal awareness by default. A memory\nfrom a year ago has the same retrieval weight as one from today.\nRecent information should generally override old information\nfor preferences and mutable facts.
\nRecommended fix:
\nfrom datetime import datetime, timedelta
\ndef time_decay_score(memory, half_life_days=30):\nage = (datetime.now() - memory.created_at).days\ndecay = 0.5 ** (age / half_life_days)\nreturn decay
\ndef retrieve_with_recency(query, user_id):\n# Get candidates\ncandidates = index.query(\nvector=embed(query),\nfilter={\"user_id\": user_id},\ntop_k=20\n)
\n# Apply time decay\nfor candidate in candidates:\n time_score = time_decay_score(candidate)\n candidate.final_score = candidate.similarity * 0.7 + time_score * 0.3\n\n# Re-sort by final score\nreturn sorted(candidates, key=lambda x: x.final_score, reverse=True)[:5]\nasync def update_preference(user_id, category, value):\n# Delete old preference\nawait memory.delete(\nfilter={\"user_id\": user_id, \"type\": \"preference\", \"category\": category}\n)
\n# Store new preference\nawait memory.upsert(\n id=f\"pref-{user_id}-{category}\",\n content={\"category\": category, \"value\": value},\n metadata={\"updated_at\": datetime.now()}\n)\nawait memory.upsert(\nid=f\"fact--v\",\ncontent=new_fact,\nmetadata={\n\"version\": version,\n\"supersedes\": previous_id,\n\"valid_from\": datetime.now()\n}\n)
\nSeverity: MEDIUM
\nSituation: User has changed preferences or provided conflicting info
\nSymptoms:\nAgent retrieves \"user prefers dark mode\" and \"user prefers light\nmode\" in same context. Gives inconsistent answers. Seems confused\nor forgetful to user.
\nWhy this breaks:\nWithout conflict resolution, both old and new information coexist.\nSemantic search might return both because they're both about the\nsame topic (preferences). Agent has no way to know which is current.
\nRecommended fix:
\nasync def store_with_conflict_check(memory, user_id):\n# Find potentially conflicting memories\nsimilar = await index.query(\nvector=embed(memory.content),\nfilter={\"user_id\": user_id, \"type\": memory.type},\nthreshold=0.9, # Very similar\ntop_k=5\n)
\nfor existing in similar:\n if is_contradictory(memory.content, existing.content):\n # Ask for resolution\n resolution = await resolve_conflict(memory, existing)\n if resolution == \"replace\":\n await index.delete(existing.id)\n elif resolution == \"version\":\n await mark_superseded(existing.id, memory.id)\n\nawait index.upsert(memory)\ndef is_contradictory(new_content, old_content):\n# Use LLM to detect contradiction\nresult = llm.invoke(f'''\nDo these two statements contradict each other?
\n Statement 1: {old_content}\n Statement 2: {new_content}\n\n Respond with just YES or NO.\n''')\nreturn result.strip().upper() == \"YES\"\nasync def consolidate_memories(user_id):\nall_memories = await index.list(filter={\"user_id\": user_id})\nclusters = cluster_by_topic(all_memories)
\nfor cluster in clusters:\n if has_conflicts(cluster):\n resolved = await llm.invoke(f'''\n These memories may conflict. Create one consolidated\n memory that represents the current truth:\n {cluster}\n ''')\n await replace_cluster(cluster, resolved)\nSeverity: MEDIUM
\nSituation: Retrieving too many memories at once
\nSymptoms:\nToken limit errors. Agent truncates important information.\nSystem prompt gets cut off. Retrieved memories compete with\nuser query for space.
\nWhy this breaks:\nRetrieval typically returns top-k results. If k is too high or\nchunks are too large, retrieved context overwhelms the window.\nCritical information (system prompt, recent messages) gets pushed\nout.
\nRecommended fix:
\nTOKEN_BUDGET = {\n\"system_prompt\": 500,\n\"user_profile\": 200,\n\"recent_messages\": 2000,\n\"retrieved_memories\": 1000,\n\"current_query\": 500,\n\"buffer\": 300, # Safety margin\n}
\ndef budget_aware_retrieval(query, context_limit=4000):\nremaining = context_limit - TOKEN_BUDGET[\"system_prompt\"] - TOKEN_BUDGET[\"buffer\"]
\n# Prioritize recent messages\nrecent = get_recent_messages(limit=TOKEN_BUDGET[\"recent_messages\"])\nremaining -= count_tokens(recent)\n\n# Then user profile\nprofile = get_user_profile(limit=TOKEN_BUDGET[\"user_profile\"])\nremaining -= count_tokens(profile)\n\n# Finally retrieved memories with remaining budget\nmemories = retrieve_memories(query, max_tokens=remaining)\n\nreturn build_context(profile, recent, memories)\ndef retrieve_with_budget(query, max_tokens=1000):\navg_chunk_tokens = 150 # From your data\nmax_k = max_tokens // avg_chunk_tokens
\nresults = index.query(query, top_k=max_k)\n\n# Trim if still over budget\ntotal_tokens = 0\nfiltered = []\nfor result in results:\n tokens = count_tokens(result.text)\n if total_tokens + tokens <= max_tokens:\n filtered.append(result)\n total_tokens += tokens\n else:\n break\n\nreturn filtered\nSeverity: MEDIUM
\nSituation: Upgrading embedding model or mixing providers
\nSymptoms:\nRetrieval quality suddenly drops. Relevant documents not found.\nRandom results returned. Works for new documents, fails for old.
\nWhy this breaks:\nEmbedding models produce different vector spaces. A query embedded\nwith text-embedding-3 won't match documents embedded with text-ada-002.\nMixing models creates garbage similarity scores.
\nRecommended fix:
\nawait index.upsert(\nid=doc_id,\nvector=embedding,\nmetadata={\n\"embedding_model\": \"text-embedding-3-small\",\n\"embedding_version\": \"2024-01\",\n\"content\": content\n}\n)
\nresults = index.query(\nvector=query_embedding,\nfilter={\"embedding_model\": current_model},\ntop_k=10\n)
\nasync def migrate_embeddings(old_model, new_model):\n# Get all documents with old model\nold_docs = await index.list(filter={\"embedding_model\": old_model})
\nfor doc in old_docs:\n # Re-embed with new model\n new_embedding = await embed(doc.content, model=new_model)\n\n # Update in place\n await index.update(\n id=doc.id,\n vector=new_embedding,\n metadata={\"embedding_model\": new_model}\n )\nSeverity: ERROR
\nIn-memory stores lose data on restart
\nMessage: In-memory store detected. Use persistent storage (Postgres, Qdrant, Pinecone) for production.
\nSeverity: WARNING
\nVectors should have metadata for filtering
\nMessage: Vector upsert without metadata. Add user_id, type, timestamp for proper filtering.
\nSeverity: ERROR
\nQueries should filter by user to prevent data leakage
\nMessage: Vector query without user filtering. Always filter by user_id to prevent data leakage.
\nSeverity: INFO
\nChunk size should be tested and justified
\nMessage: Hardcoded chunk size. Test different sizes for your content type and measure retrieval accuracy.
\nSeverity: WARNING
\nChunk overlap prevents boundary issues
\nMessage: Text splitting without overlap. Add chunk_overlap (10-20%) to prevent boundary issues.
\nSeverity: WARNING
\nPure semantic search often returns irrelevant results
\nMessage: Pure semantic search. Add metadata filters (user, type, time) for better relevance.
\nSeverity: WARNING
\nUnbounded retrieval can overflow context
\nMessage: Retrieval without limit. Set top_k to prevent context overflow.
\nSeverity: WARNING
\nTrack embedding model to handle migrations
\nMessage: Store embedding model version in metadata to handle model migrations.
\nSeverity: ERROR
\nDocuments and queries must use same embedding model
\nMessage: Ensure same embedding model for indexing and querying.
\nWorks well with: autonomous-agents, multi-agent-orchestration, llm-architect, agent-tool-builder