7. Conversation Management and Memory Patterns

Overview

In the previous capsules you built every piece of the memory system separately: short-term memory for the active conversation (02), checkpointing to persist state (03-04), long-term memory for data that crosses sessions (05), and time-travel debugging to navigate the history (06). Each piece solves a specific problem. But in a real agent, problems don't come one at a time. A user 40 turns into researching RAG needs simultaneously: a summary of what's been discussed (summary memory), the fact that they work at Google and prefer academic papers (entity memory), the memory that they researched embeddings last week and already have that context (episodic memory), and all of this without blowing up the token budget or the cost per request.

This capsule synthesizes everything. You're not introducing new tools — you already learned everything you need in capsules 02-06. What's new is the criteria for combining them: when to use each memory pattern, how to combine them without redundancy, and how to manage memory as a limited resource with garbage collection strategies.

Connection to the module: This is the last technical capsule before the project. In capsule 08 you'll implement persistent memory in the Research Agent, and you'll need to decide which combination of memory patterns to use. This capsule gives you the decision framework for that.

Summary Memory

The problem it solves

A 50-turn conversation can have 30,000+ tokens of history. Sending all of it to the LLM on every request is expensive, slow, and counterproductive (the model loses attention in long contexts — "Lost in the Middle"). Summary memory compresses the old conversation into a ~200-token paragraph that captures the essentials: what the user asked for, what was found, what preferences they expressed, what's still pending.

You already implemented summarization in capsule 02. Here the focus is when to trigger it and how to integrate it with the other patterns.

When to trigger summary memory

Don't summarize from the first message. Summarization has a cost (an extra LLM call) and it's only justified when the history grows enough:

SUMMARY_THRESHOLDS = {
    "chatbot_casual": 20,     # short turns, 50-100 tokens each
    "research_agent": 12,     # medium turns with tool responses
    "customer_support": 15,   # medium turns
    "code_assistant": 10,     # long turns, 500+ token snippets
}

Implementation as a LangGraph node

from langchain_core.messages import HumanMessage, RemoveMessage
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

SUMMARY_PROMPT = """Summarize this conversation, preserving:
1. The user's main task
2. Preferences expressed (format, sources, constraints)
3. Key findings so far
4. Pending tasks

200 words maximum. Don't include code — context only.

Conversation:
{conversation}"""


def summary_memory_node(state: dict) -> dict:
    messages = state["messages"]
    if len(messages) <= 12:
        return {}

    old_messages = messages[:-6]
    conversation = "\n".join(
        f"{m.__class__.__name__}: {m.content[:300]}"
        for m in old_messages
        if hasattr(m, "content") and m.content
    )

    existing_summary = state.get("summary", "")
    if existing_summary:
        prompt = (
            f"Update this summary with the new information:\n\n"
            f"Previous summary:\n{existing_summary}\n\n"
            f"New conversation:\n{conversation}"
        )
    else:
        prompt = SUMMARY_PROMPT.format(conversation=conversation)

    new_summary = model.invoke([HumanMessage(content=prompt)]).content
    return {
        "summary": new_summary,
        "messages": [RemoveMessage(id=m.id) for m in old_messages],
    }

Trade-offs

Rule of thumb: If your average conversation is under 15 turns, summary memory isn't justified. If you regularly hit 30+, it's essential.

Entity Memory

The problem it solves

Summary memory compresses the whole conversation. But there's specific data you need with exact precision: the user's name, their company, the technologies they mentioned. These are entities — structured data the agent must remember faithfully, not summarized.

Picture it: turn 3 says "I work at Google, on the ML Infrastructure team." Turn 12 says "I'm migrating from TensorFlow to JAX." At turn 25, when you recommend deployment tools, the summary might say "works at a big company, migrating frameworks." With entity memory, you have: {company: "Google", team: "ML Infrastructure", migration: "TensorFlow → JAX"}. The recommendation is specific to Google's scale and the JAX ecosystem.

Implementation with structured extraction

from pydantic import BaseModel, Field
from langgraph.store.memory import InMemoryStore


class ExtractedEntities(BaseModel):
    people: list[str] = Field(default_factory=list)
    companies: list[str] = Field(default_factory=list)
    technologies: list[str] = Field(default_factory=list)
    preferences: list[str] = Field(default_factory=list)


entity_extractor = model.with_structured_output(ExtractedEntities)
store = InMemoryStore()


def extract_and_store_entities(messages: list, user_id: str) -> ExtractedEntities:
    recent = messages[-4:]
    conversation = "\n".join(
        f"{m.__class__.__name__}: {m.content[:500]}"
        for m in recent if hasattr(m, "content") and m.content
    )

    entities = entity_extractor.invoke(
        f"Extract entities from this conversation:\n\n{conversation}"
    )

    existing = store.get(("entities", user_id), "profile")
    if existing and existing.value:
        old = existing.value
        merged = {
            k: list(set(old.get(k, []) + getattr(entities, k)))
            for k in ["people", "companies", "technologies", "preferences"]
        }
    else:
        merged = entities.model_dump()

    store.put(("entities", user_id), "profile", merged)
    return entities

Injecting entities into the prompt

def build_context_from_entities(user_id: str) -> str:
    item = store.get(("entities", user_id), "profile")
    if not item or not item.value:
        return ""

    e = item.value
    parts = []
    if e.get("companies"):
        parts.append(f"Company: {', '.join(e['companies'])}")
    if e.get("technologies"):
        parts.append(f"Technologies: {', '.join(e['technologies'])}")
    if e.get("preferences"):
        parts.append(f"Preferences: {', '.join(e['preferences'])}")
    return "User context:\n" + "\n".join(parts) if parts else ""

When to extract entities

Don't extract on every turn — it's an expensive LLM call. Extract every 5 turns (the periodic strategy). It costs ~$0.001 per extraction and captures most of the relevant entities. The alternative: event-driven (only when the model detects new information), more precise but it requires a classifier.

Episodic Memory

The problem it solves

Summary memory compresses the current conversation. Entity memory extracts data about the user. Neither answers: "What happened in the previous session?" Episodic memory stores episodes — summaries of complete sessions: what was researched, what was found, and what the outcome was.

The key case: a user comes back three days later and says "Remember what we researched about RAG? I want to go deeper on embeddings." Without episodic memory, the agent has no idea. With episodic memory, it has a record: {session: "session-42", topic: "RAG", findings: [...], outcome: "Recommendation: Cohere + Pinecone"}.

Implementation: saving episodes at session close

from pydantic import BaseModel, Field
from datetime import datetime


class Episode(BaseModel):
    session_id: str
    date: str
    topic: str = Field(description="The session's main topic")
    findings: list[str] = Field(description="Key findings")
    outcome: str = Field(description="The result or conclusion")
    pending: list[str] = Field(default_factory=list)


episode_creator = model.with_structured_output(Episode)


def create_episode(session_id: str, messages: list, summary: str = "") -> Episode:
    context = summary if summary else "\n".join(
        f"{m.__class__.__name__}: {m.content[:200]}"
        for m in messages if hasattr(m, "content") and m.content
    )
    episode = episode_creator.invoke(
        f"Create a structured episode:\n\nSession: {session_id}\n"
        f"Date: {datetime.now().strftime('%Y-%m-%d')}\n\n{context}"
    )
    store.put(("episodes", session_id.split("-")[0]), session_id, episode.model_dump())
    return episode


def format_episodes_for_prompt(user_id: str, limit: int = 5) -> str:
    items = store.search(("episodes", user_id))
    episodes = sorted(
        [item.value for item in items],
        key=lambda e: e.get("date", ""), reverse=True,
    )[:limit]
    if not episodes:
        return ""
    lines = ["Previous sessions:"]
    for ep in episodes:
        lines.append(f"- [{ep.get('date')}] {ep.get('topic')}: {ep.get('outcome', '')[:80]}")
        if ep.get("pending"):
            lines.append(f"  Pending: {', '.join(ep['pending'])}")
    return "\n".join(lines)

The best moment to create an episode is at session close. In REST APIs where there's no explicit close, use an inactivity trigger (no messages for N minutes → create the episode).

Combining Memory Patterns

Why one pattern isn't enough

The combined architecture

Each memory type feeds a different part of the prompt:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import SystemMessage, trim_messages


class FullMemoryState(TypedDict):
    messages: Annotated[list, add_messages]
    summary: str
    user_id: str


SYSTEM_TEMPLATE = """You are an expert research agent.

{entity_context}

{episode_context}

{summary_context}"""


def agent_with_full_memory(state: FullMemoryState) -> dict:
    user_id = state.get("user_id", "anonymous")
    system_content = SYSTEM_TEMPLATE.format(
        entity_context=build_context_from_entities(user_id),
        episode_context=format_episodes_for_prompt(user_id),
        summary_context=f"Summary: {state['summary']}" if state.get("summary") else "",
    )
    trimmed = trim_messages(
        [SystemMessage(content=system_content)] + state["messages"],
        max_tokens=6000, strategy="last",
        token_counter=len, include_system=True, allow_partial=False,
    )
    return {"messages": [model.invoke(trimmed)]}

A token budget per memory type

Without garbage collection, entities pile up, episodes grow, and the context inflates silently.

Memory as a Limited Resource

The parallel with garbage collection

In programming, RAM is finite — without freeing objects, you have a memory leak. An agent's memory has the same problem. Every entity, every episode, every summary burns tokens when it's injected into the prompt. A user with 100 sessions would have 100 episodes in their context — thousands of irrelevant tokens on every request.

Strategy 1: TTL (Time-to-Live)

Data that isn't accessed in N days gets deleted:

from datetime import datetime, timedelta


def cleanup_by_ttl(user_id: str, ttl_days: int = 30) -> int:
    items = store.search(("episodes", user_id))
    cutoff = (datetime.now() - timedelta(days=ttl_days)).strftime("%Y-%m-%d")
    removed = 0
    for item in items:
        if item.value.get("date", "9999-99-99") < cutoff:
            store.delete(("episodes", user_id), item.key)
            removed += 1
    return removed

Strategy 2: Relevance scoring

Not every episode is worth the same. Assign a score and prioritize:

def score_episode(episode: dict) -> float:
    score = 0.0
    score += min(len(episode.get("findings", [])) * 0.15, 0.3)
    score += min(len(episode.get("pending", [])) * 0.2, 0.3)
    if len(episode.get("outcome", "")) > 50:
        score += 0.2
    try:
        age = (datetime.now() - datetime.strptime(episode["date"], "%Y-%m-%d")).days
        score += max(0, 0.2 * (1 - age / 30))
    except (ValueError, KeyError):
        pass
    return min(score, 1.0)

Strategy 3: Size limits

Define a maximum budget per type and trim when it's exceeded:

MEMORY_LIMITS = {
    "entities_max_per_category": 20,
    "episodes_max": 10,
    "summary_max_tokens": 300,
}


def enforce_limits(user_id: str):
    item = store.get(("entities", user_id), "profile")
    if item and item.value:
        profile = item.value
        max_per = MEMORY_LIMITS["entities_max_per_category"]
        for key in profile:
            if isinstance(profile[key], list) and len(profile[key]) > max_per:
                profile[key] = profile[key][-max_per:]
        store.put(("entities", user_id), "profile", profile)

The real cost of memory

Every context token gets sent on every request:

500 tokens of well-managed memory > 5,000 tokens of uncurated memory.

Decision Framework

Which memory pattern for which use case

Quick decision criteria

1. Is the average conversation > 15 turns? → Summary memory
2. Do you need user data between sessions? → Entity memory
3. Does the user come back and expect continuity? → Episodic memory + Checkpointing
4. Is the token budget limited? → Garbage collection is mandatory
5. Should the agent "learn" from past interactions? → Entity + Episodic + relevance scoring

Anti-patterns

Connection to the Project

In capsule 08 you'll implement persistent memory in the Research Agent:

• Summary memory: Research runs of 15-30 turns. The summary triggers past 12 messages, compressing the old conversation to control the token budget.
• Entity memory: Extracts the user's preferences (preferred sources, output format) and injects them into the system prompt to personalize the research.
• Checkpointing with PostgresSaver: The research persists at every step. If the session gets interrupted, the user picks up where they left off (capsules 03-04).
• Garbage collection: Size limits for entities (max 20 per category) and a 30-day TTL for episodes.

What you won't implement (but now you know when you'd need it): episodic memory. The Research Agent is single-session. You add episodic memory when you have recurring users with multiple sessions.

Troubleshooting

Problem 1: The summary loses the user's critical instructions

Symptom: The user said "academic sources only" at turn 3. After summarization, the agent uses blogs.

Solution: Don't rely on the summary for preferences. Extract them with entity memory and store them in a separate field that never gets summarized:

def agent_node(state: dict) -> dict:
    preferences = get_user_preferences(state["user_id"])
    system = f"PREFERENCES (always respect): {preferences}\n\n{state.get('summary', '')}"

Problem 2: Entity extraction is inconsistent

Symptom: "Google" gets extracted sometimes and not others. Duplicate entities with variations ("Python 3.11" and "Python").

Solution: Normalize before saving and deduplicate:

def normalize_entity(entity: str) -> str:
    return entity.strip().lower()

def deduplicate(entities: list[str]) -> list[str]:
    seen = {}
    for e in entities:
        key = normalize_entity(e)
        if key not in seen:
            seen[key] = e
    return list(seen.values())

Problem 3: There's no explicit "session close" to create episodes

Symptom: In a REST API, the user simply stops sending messages.

Solution: An inactivity trigger — if there's no activity on a thread_id for 5 minutes, create the episode as a background job.

Problem 4: The memory context exceeds the token budget

Symptom: Entities + episodes + summary add up to 3,000+ tokens in the system prompt.

Solution: A budget per component, trimming when exceeded:

def build_context_within_budget(user_id: str, summary: str, budget: int = 800) -> str:
    entity_ctx = build_context_from_entities(user_id)[:budget // 3]
    episode_ctx = format_episodes_for_prompt(user_id)[:budget // 3]
    summary_ctx = summary[:budget // 3] if summary else ""
    return f"{entity_ctx}\n\n{episode_ctx}\n\n{summary_ctx}".strip()

Exercises

Exercise 1: Summary memory with an adaptive threshold

Build an AdaptiveSummarizer that adjusts the threshold based on the average message size. Long messages (>200 average tokens) → summarize sooner. Short messages (<50 tokens) → allow more messages.

from langchain_core.messages import HumanMessage


class AdaptiveSummarizer:
    def __init__(self, base_threshold: int = 12, min_t: int = 6, max_t: int = 25):
        self.base = base_threshold
        self.min_t = min_t
        self.max_t = max_t

    def get_threshold(self, messages: list) -> int:
        if not messages:
            return self.base
        avg_chars = sum(
            len(m.content) for m in messages if hasattr(m, "content") and m.content
        ) / len(messages)
        avg_tokens = avg_chars / 4

        if avg_tokens > 200:
            t = self.base - 4
        elif avg_tokens < 50:
            t = self.base + 6
        else:
            t = self.base
        return max(self.min_t, min(self.max_t, t))

    def should_summarize(self, messages: list) -> bool:
        return len(messages) > self.get_threshold(messages)


s = AdaptiveSummarizer()
short = [HumanMessage(content="Yes")] * 20
long = [HumanMessage(content="x" * 1000)] * 8
print(f"Short: threshold={s.get_threshold(short)}, summarize={s.should_summarize(short)}")
print(f"Long: threshold={s.get_threshold(long)}, summarize={s.should_summarize(long)}")

Exercise 2: Entity memory with deduplication

Implement an EntityStore that normalizes entities (lowercase, trim), deduplicates, and respects size limits per category.

from dataclasses import dataclass, field


@dataclass
class EntityStore:
    profiles: dict[str, dict[str, list[str]]] = field(default_factory=dict)
    max_per_category: int = 20

    def merge(self, user_id: str, new_entities: dict[str, list[str]]):
        existing = self.profiles.get(user_id, {})
        for category, values in new_entities.items():
            combined = existing.get(category, []) + values
            seen = {}
            for v in combined:
                key = v.strip().lower()
                if key not in seen:
                    seen[key] = v
            deduped = list(seen.values())
            existing[category] = deduped[-self.max_per_category:]
        self.profiles[user_id] = existing

    def format_for_prompt(self, user_id: str) -> str:
        profile = self.profiles.get(user_id, {})
        return "\n".join(f"{k}: {', '.join(v)}" for k, v in profile.items() if v)


es = EntityStore(max_per_category=5)
es.merge("u1", {"tech": ["Python", "FastAPI", "python", "LangChain"]})
es.merge("u1", {"tech": ["LangGraph", "Python 3.11", "FAISS", "Pinecone", "ChromaDB"]})
print(es.format_for_prompt("u1"))
print(f"Count: {len(es.profiles['u1']['tech'])}")  # max 5

Exercise 3: Episodic memory with relevance scoring

Implement episodic memory that assigns a score based on findings, pending tasks, and recency. Return only the top-N most relevant episodes.

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class EpisodicMemory:
    episodes: dict[str, list[dict]] = field(default_factory=dict)

    def add(self, user_id: str, topic: str, findings: list[str],
            outcome: str, pending: list[str] | None = None):
        ep = {"date": datetime.now().strftime("%Y-%m-%d"), "topic": topic,
              "findings": findings, "outcome": outcome, "pending": pending or []}
        self.episodes.setdefault(user_id, []).append(ep)

    def score(self, ep: dict) -> float:
        s = min(len(ep.get("findings", [])) * 0.15, 0.3)
        s += min(len(ep.get("pending", [])) * 0.2, 0.3)
        if len(ep.get("outcome", "")) > 50:
            s += 0.2
        try:
            age = (datetime.now() - datetime.strptime(ep["date"], "%Y-%m-%d")).days
            s += max(0, 0.2 * (1 - age / 30))
        except (ValueError, KeyError):
            pass
        return round(min(s, 1.0), 3)

    def get_relevant(self, user_id: str, limit: int = 3) -> list[dict]:
        eps = self.episodes.get(user_id, [])
        return sorted(eps, key=self.score, reverse=True)[:limit]


mem = EpisodicMemory()
mem.add("alice", "Basic RAG", ["FAISS < 100K docs"], "Chose FAISS")
mem.add("alice", "Advanced RAG", ["Pinecone scales", "Re-ranking +15%"],
        "Migrate to Pinecone", ["Benchmark FAISS vs Pinecone"])
mem.add("alice", "Casual chat", [], "Greetings")

for ep in mem.get_relevant("alice"):
    print(f"  [{mem.score(ep):.3f}] {ep['topic']}: {ep['outcome']}")

Exercise 4: A garbage collector with TTL + relevance + size limits

Implement a MemoryGC that applies all three cleanup strategies over episodes and entities. Generate a report of what was removed.

from datetime import datetime, timedelta


class MemoryGC:
    def __init__(self, ttl_days=30, max_episodes=10, max_entities_per_cat=15):
        self.ttl_days = ttl_days
        self.max_episodes = max_episodes
        self.max_entities = max_entities_per_cat

    def _score(self, ep):
        s = len(ep.get("findings", [])) * 0.15 + len(ep.get("pending", [])) * 0.2
        try:
            age = (datetime.now() - datetime.strptime(ep["date"], "%Y-%m-%d")).days
            s += max(0, 0.2 * (1 - age / 30))
        except (ValueError, KeyError):
            pass
        return s

    def run(self, episodes, entities):
        report = {"ttl": 0, "relevance": 0, "entities_pruned": {}}

        cutoff = (datetime.now() - timedelta(days=self.ttl_days)).strftime("%Y-%m-%d")
        before = len(episodes)
        episodes = [e for e in episodes if e.get("date", "9999") >= cutoff]
        report["ttl"] = before - len(episodes)

        if len(episodes) > self.max_episodes:
            episodes = sorted(episodes, key=self._score, reverse=True)
            report["relevance"] = len(episodes) - self.max_episodes
            episodes = episodes[:self.max_episodes]

        for cat, vals in entities.items():
            if isinstance(vals, list) and len(vals) > self.max_entities:
                report["entities_pruned"][cat] = len(vals) - self.max_entities
                entities[cat] = vals[-self.max_entities:]

        return episodes, entities, report


gc = MemoryGC(ttl_days=30, max_episodes=3, max_entities_per_cat=5)
eps = [
    {"date": "2026-01-01", "topic": "Old", "findings": [], "pending": [], "outcome": ""},
    {"date": "2026-03-05", "topic": "RAG", "findings": ["f1", "f2"], "pending": ["p1"], "outcome": "OK"},
    {"date": "2026-03-07", "topic": "MCP", "findings": ["f1", "f2", "f3"], "pending": ["p1"], "outcome": "Good"},
    {"date": "2026-03-08", "topic": "Testing", "findings": [], "pending": ["p1"], "outcome": "Pending"},
]
ents = {"tech": ["Python", "FastAPI", "LangChain", "LangGraph", "React", "Docker", "K8s"]}
eps, ents, report = gc.run(eps, ents)
print(f"Report: {report}")
print(f"Episodes: {[e['topic'] for e in eps]}")
print(f"Tech: {ents['tech']}")

Exercise 5: A complete system with three memory patterns

Build a MemoryManager that combines summary, entity, and episodic memory. Methods: (1) process a new turn, (2) build the complete context with a per-component budget, (3) run garbage collection. Simulate 15 turns and show how it evolves.

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class MemoryManager:
    summary: str = ""
    entities: dict[str, list[str]] = field(default_factory=dict)
    episodes: list[dict] = field(default_factory=list)
    turn_count: int = 0
    summary_threshold: int = 10
    extract_interval: int = 5
    max_entities: int = 10

    def process_turn(self, user_msg: str, messages: list):
        self.turn_count += 1
        if self.turn_count % self.extract_interval == 0:
            for word in user_msg.split():
                if word[0:1].isupper() and len(word) > 2 and word.isalpha():
                    self.entities.setdefault("mentions", [])
                    if word not in self.entities["mentions"]:
                        self.entities["mentions"].append(word)
                        if len(self.entities["mentions"]) > self.max_entities:
                            self.entities["mentions"] = self.entities["mentions"][-self.max_entities:]
        if len(messages) > self.summary_threshold:
            texts = [str(m)[:80] for m in messages[-5:]]
            self.summary = f"A {len(messages)}-turn conversation. Recent: {'; '.join(texts)}"

    def build_context(self, budget: int = 600) -> str:
        parts = []
        third = budget // 3
        if self.entities:
            e_text = "; ".join(f"{k}: {', '.join(v)}" for k, v in self.entities.items())
            parts.append(f"User: {e_text[:third]}")
        if self.episodes:
            ep_text = " | ".join(f"[{e['date']}] {e['topic']}" for e in self.episodes[-3:])
            parts.append(f"Sessions: {ep_text[:third]}")
        if self.summary:
            parts.append(f"Context: {self.summary[:third]}")
        return "\n".join(parts)

    def close_session(self, topic: str, findings: list[str]):
        self.episodes.append({
            "date": datetime.now().strftime("%Y-%m-%d"),
            "topic": topic, "findings": findings,
            "outcome": f"{self.turn_count} turns on {topic}", "pending": [],
        })
        if len(self.episodes) > 5:
            self.episodes = self.episodes[-5:]

    def stats(self) -> str:
        n_ent = sum(len(v) for v in self.entities.values())
        return (f"Turn {self.turn_count} | Summary: {len(self.summary)}ch | "
                f"Entities: {n_ent} | Episodes: {len(self.episodes)}")


mm = MemoryManager(summary_threshold=8, extract_interval=3)
conversations = [
    "I want to research RAG with Python", "I prefer academic sources",
    "I work at Google on Infrastructure", "Which vector database do you recommend?",
    "FAISS for a prototype?", "I need to support 1M documents",
    "Give me an example with LangChain", "Compare it with fine-tuning",
    "How much does each approach cost?", "I prefer the cheaper option",
    "How do I integrate it with FastAPI?", "Show me the deployment",
    "Redis or PostgreSQL?", "Use Redis", "Summarize everything",
]
msgs = []
for i, msg in enumerate(conversations):
    msgs.append({"role": "user", "content": msg})
    mm.process_turn(msg, msgs)
    if (i + 1) % 5 == 0:
        print(mm.stats())

mm.close_session("RAG Investigation", ["FAISS proto", "Pinecone prod"])
print(f"\n{mm.stats()}")
print(f"\nContext:\n{mm.build_context()}")

Summary

• Summary memory compresses the old conversation into ~200 tokens. Trigger it when the history passes a threshold (12-20 messages). It preserves context that window trimming would lose, in exchange for an extra LLM call. Don't use it in short conversations.
• Entity memory extracts structured data about the user (name, company, technologies, preferences) and persists it in the Store. It enables precise personalization that a generic summary can't achieve. Extract periodically (every 5 turns), not on every turn.
• Episodic memory stores summaries of complete sessions to give continuity between conversations. The user feels the agent "knows them." Create episodes at session close or on inactivity.
• Combining patterns is the key: summary for compression, entities for personalization, episodic for continuity. Each type feeds a part of the system prompt with an allocated token budget.
• Memory is a limited resource. Without garbage collection, the context grows out of control. Three strategies: TTL (delete old data), relevance scoring (delete the least useful), and size limits (trim per category).
• The decision framework depends on the use case: a casual chatbot needs no memory patterns; a research agent needs summary + entities + checkpointing; a personal assistant needs everything including episodic.

Additional Resources

1. LangGraph Memory Concepts — Official documentation on the memory types in LangGraph
2. LangGraph Store API — The Store reference for long-term memory and entity storage
3. How to add summary of conversation history — Summarization tutorial in LangGraph
4. Lost in the Middle (Liu et al., 2023) — The paper on how LLMs lose attention in long contexts
5. MemGPT: Towards LLMs as Operating Systems — A paper on operating-system-inspired memory management for LLMs