Why this matters

Memory is what transforms a stateless LLM call into a coherent AI application. Without memory, every conversation starts from scratch. The challenge is that LLMs have no built-in persistent state — memory is entirely an application-layer concern.

Core concepts

Three-tier memory architecture

Working memory — the current context window. Capacity: ~8-128K tokens. Duration: one API call.

Short-term memory — the current session. Stored in Redis or DynamoDB with TTL. Duration: hours to days.

Long-term memory — persistent facts about the user or domain. Stored in a database or vector store. Duration: indefinite.

Conversation summarization

When session history exceeds 60-70% of the history budget, summarize the oldest 50% of turns.

async def maybe_summarize(
    history: list[dict],
    history_budget_tokens: int,
    summarize_fn,
    count_fn=estimate_tokens,
) -> list[dict]:
    total = sum(count_fn(m["content"]) for m in history)
    if total < history_budget_tokens * 0.7:
        return history

    midpoint = len(history) // 2
    summary_text = await summarize_fn(history[:midpoint])
    return [
        {"role": "user", "content": f"[Summary of earlier conversation]
{summary_text}"},
        {"role": "assistant", "content": "Understood. I have the context from our earlier discussion."},
    ] + history[midpoint:]

User preference storage

from dataclasses import dataclass


@dataclass
class UserMemory:
    user_id: str
    preferences: dict
    facts: list[str]
    past_project_summaries: list[str]

    def to_context_string(self) -> str:
        lines = ["## User context"]
        if self.preferences:
            prefs = ", ".join(f"{k}: {v}" for k, v in self.preferences.items())
            lines.append(f"Preferences: {prefs}")
        if self.facts:
            lines.append("Known facts:")
            lines.extend(f"- {fact}" for fact in self.facts[:10])
        return "
".join(lines)

Code: MemoryManager

import json
from dataclasses import dataclass, field
from typing import Optional, Protocol


class StorageBackend(Protocol):
    async def get(self, key: str) -> Optional[str]: ...
    async def set(self, key: str, value: str) -> None: ...


@dataclass
class MemoryManager:
    user_id: str
    storage: StorageBackend
    summarize_fn: object
    history_budget_tokens: int = 4000
    _working_history: list[dict] = field(default_factory=list)

    async def load_session(self, session_id: str) -> None:
        raw = await self.storage.get(f"session:{session_id}:history")
        if raw:
            self._working_history = json.loads(raw)

    async def save_session(self, session_id: str) -> None:
        await self.storage.set(
            f"session:{session_id}:history", json.dumps(self._working_history)
        )

    async def get_user_memory(self) -> Optional[UserMemory]:
        raw = await self.storage.get(f"user:{self.user_id}:memory")
        if not raw:
            return None
        return UserMemory(**json.loads(raw))

    def add_turn(self, role: str, content: str) -> None:
        self._working_history.append({"role": role, "content": content})

    async def assemble_context(
        self,
        system_prompt: str,
        user_input: str,
        retrieved_docs: list[str] | None = None,
    ) -> list[dict]:
        messages = [{"role": "system", "content": system_prompt}]

        user_memory = await self.get_user_memory()
        if user_memory:
            messages.append({"role": "user", "content": user_memory.to_context_string()})
            messages.append({"role": "assistant", "content": "Got it."})

        if retrieved_docs:
            doc_block = "

---

".join(retrieved_docs[:5])
            messages.append({"role": "user", "content": f"Relevant docs:
{doc_block}"})
            messages.append({"role": "assistant", "content": "I have reviewed these documents."})

        history = await maybe_summarize(
            self._working_history, self.history_budget_tokens, self.summarize_fn
        )
        messages.extend(history)
        messages.append({"role": "user", "content": user_input})
        return messages

Common mistakes

Storing everything in the session cache only. Preferences learned in one session are lost when it expires.

Summarizing too aggressively. Summarize at 60-70% capacity, not 20%.

No retrieval from long-term memory. Use semantic search to retrieve only what is relevant to the current request.

Not persisting summaries. Session-end summaries should become long-term memory entries.

Try it yourself

1. Implement StorageBackend with an in-memory dict. Verify save_session and load_session round-trip correctly.
2. Extend UserMemory with an update_from_turn method that detects preference statements.
3. Design a schema for storing past project summaries for effective retrieval.