paulpas AI Agent Memory Systems
Implements conversation memory patterns that keep AI agents coherent across extended interactions. Provides bounded conversation buffers with turn limits and token budgets, auto-summarization of stale turns to preserve conversational continuity, and vector-backed long-term storage for semantic recall of facts discovered during agent sessions. A well-architected memory system is the backbone of any production-grade agent — without it, every prompt starts from zero and every session forgets what happened five minutes ago.
TL;DR Checklist
- Set bounded conversation buffer with explicit turn limits and token budget tracking before each model call
- Implement sliding window truncation that preserves system prompt and recent turns while discarding oldest messages first
- Detect memory pressure using token ratio thresholds (warn at 70%, trigger summary at 85%)
- Configure vector store backend (FAISS, Chroma, or Qdrant) with appropriate embeddings model for the target domain
- Run similarity search with configurable top-k and minimum score before injecting memories into context
- Reference
code-philosophy(5 Laws of Elegant Defense) — parse inputs at boundaries, fail fast on invalid states, never mutate shared state
When to Use
Use this skill when:
- Building an AI agent that must maintain coherent conversation across many turns (10+)
- Designing memory architecture for a chat application where context window limits are a constraint
- Implementing long-term factual recall so the agent remembers information from earlier in a session
- Needing to compress old conversation turns into summaries before hitting context token limits
- Building a RAG-enhanced agent that combines short-term buffer with vector-backed semantic retrieval
- Integrating memory management into an existing agent framework (LangChain, LlamaIndex, custom)
When NOT to Use
Avoid this skill for:
- Single-turn interactions where no conversation history needs preservation — the overhead outweighs any benefit
- Ultra-low-latency inference contexts (<50ms response budget) — vector similarity search and summarization add measurable latency
- Simple echo or passthrough chatbots that do not need memory of prior turns — just append to a raw list
- Sessions where the full conversation fits comfortably within the model's context window with significant headroom (<40% utilization)
Orchestration Flow
User Input ──► Token Budget Check ─────────────────────────────────┐
↓ │
┌───────────▼───────────┐ │
│ Memory Pressure? │ │
│ │ │
│ < 70%: Full pass │ │
│ 70-85%: Log warning │ │
│ > 85%: Summarize │ │
└───────────┬───────────┘ │
↓ │
┌───────────▼───────────┐ │
│ Short-Term Buffer │◄──────────────┐ │
│ (bounded window) │ │ │
└───────────┬───────────┘ │ │
↓ │ │
┌───────────▼───────────┐ │ │
│ Long-Term Memory │ │ │
│ (vector store query) │ │ │
└───────────┬───────────┘ │ │
↓ │ │
┌───────────▼───────────┐ │ │
│ Assemble Prompt: │ │ │
│ System + Memories + │ │ │
│ Recent History │ │ │
└───────────┬───────────┘ │ │
↓ │ │
┌───────────▼───────────┐ │ │
│ Model Call │ │ │
└───────────┬───────────┘ │ │
↓ │ │
┌───────────▼───────────┐ │ │
│ Store Turn in Buffer │───────────────┘ │
│ Extract facts → Store │ │
└───────────────────────────────────────────────────┘
Core Workflow
Step 1: Configure Short-Term Memory
Set up a bounded conversation buffer that enforces turn limits and tracks token consumption. The buffer must never grow unbounded — every model call starts from a known, bounded state. Use both a maximum message count (N recent turns) and a maximum token budget (tokens for dynamic content only).
"""
Bounded conversation memory with token-aware window management.
Enforces turn limits via configurable max_messages and tracks cumulative
token usage through an optional tokenizer callback. The buffer supports
append-only operations with automatic eviction of oldest messages when
budget constraints are exceeded.
Designed for production agents that must maintain coherent context
across extended conversations without exceeding model context windows.
"""
from __future__ import annotations
import logging
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable, Optional
logger = logging.getLogger(__name__)
@dataclass
class ConversationMessage:
"""A single message in the conversation buffer.
Attributes:
role: The sender role ('system', 'user', 'assistant', or 'tool').
content: The message text content.
tokens: Cached token count, computed lazily via tokenizer.
timestamp: When this message was added to the buffer.
metadata: Optional structured data (e.g., tool call results, citations).
"""
role: str
content: str
tokens: int = 0
timestamp: datetime = field(
default_factory=lambda: datetime.now(timezone.utc)
)
metadata: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
"""Serialize this message for API transmission."""
return {
"role": self.role,
"content": self.content,
**self.metadata if self.metadata else {},
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ConversationMessage:
"""Deserialize a message from API format."""
return cls(
role=data["role"],
content=data["content"],
metadata={k: v for k, v in data.items() if k not in ("role", "content")},
)
class BoundedConversationBuffer:
"""Bounded conversation buffer with token-aware window management.
Maintains a deque of ConversationMessage instances, enforcing both
a maximum message count and a maximum token budget for dynamic
(non-system) content. System messages are always preserved regardless
of budget — only user/assistant/tool messages are evicted.
Follows Law 3 (Atomic Predictability): all mutations return explicit
confirmation and never mutate the caller's input data.
"""
def __init__(
self,
max_messages: int = 50,
max_tokens: int = 128_000,
system_message_max_tokens: Optional[int] = None,
tokenizer: Optional[Callable[[str], int]] = None,
) -> None:
"""Initialize the bounded conversation buffer.
Args:
max_messages: Maximum number of dynamic messages to retain (excl. system).
Evicts oldest first when exceeded. Set 0 for unlimited count.
max_tokens: Token budget for dynamic messages (excl. system prompt).
When exceeded, oldest messages are evicted until under budget.
Set 0 for unlimited tokens.
system_message_max_tokens: Optional hard limit for total system message tokens.
tokenizer: Callable that returns token count for a string.
If None, uses a rough character-based estimate (~4 chars/token).
"""
self._max_messages = max_messages
self._max_tokens = max_tokens
self._system_message_max_tokens = system_message_max_tokens or 0
self._tokenizer = tokenizer or (lambda text: len(text) // 4)
self._messages: deque[ConversationMessage] = deque()
self._dynamic_token_count: int = 0
self._static_token_count: int = 0
@property
def message_count(self) -> int:
"""Total number of messages in the buffer."""
return len(self._messages)
@property
def dynamic_token_count(self) -> int:
"""Token count for non-system (dynamic) messages only."""
return self._dynamic_token_count
@property
def static_token_count(self) -> int:
"""Token count for system/developer messages."""
return self._static_token_count
@property
def total_token_count(self) -> int:
"""Total token count across all messages."""
return self._dynamic_token_count + self._static_token_count
@property
def utilization_ratio(self) -> float:
"""Ratio of current dynamic tokens to the budget (0.0 - 1.0+)."""
if self._max_tokens <= 0:
return 0.0
return self._dynamic_token_count / self._max_tokens
@property
def is_under_budget(self) -> bool:
"""Whether the buffer is within both message count and token budgets."""
if self._max_messages > 0 and len(self._messages) >= self._max_messages:
return False
if self._max_tokens > 0 and self._dynamic_token_count > self._max_tokens:
return False
return True
def add_system_message(self, content: str, metadata: Optional[dict[str, Any]] = None) -> None:
"""Add or replace the system message. Always preserved regardless of budget.
Args:
content: The system prompt text.
metadata: Optional metadata attached to this message.
"""
# Remove existing system messages if present
self._messages = deque(
msg for msg in self._messages if msg.role not in ("system", "developer")
)
self._static_token_count -= self.total_system_tokens
token_count = self._tokenizer(content)
if self._system_message_max_tokens > 0 and token_count > self._system_message_max_tokens:
logger.warning(
"System message (%d tokens) exceeds max budget (%d). Truncating.",
token_count,
self._system_message_max_tokens,
)
msg = ConversationMessage(
role="system",
content=content,
tokens=token_count,
metadata=metadata or {},
)
self._messages.appendleft(msg)
self._static_token_count += token_count
def append(self, role: str, content: str, metadata: Optional[dict[str, Any]] = None) -> bool:
"""Append a message to the buffer, evicting old messages if needed.
Args:
role: The message role ('user', 'assistant', or 'tool').
content: The message content text.
metadata: Optional structured data for this message.
Returns:
True if the message was added successfully, False if budget exceeded
and all dynamic messages were already evicted.
"""
if role in ("system", "developer"):
raise ValueError(f"Use add_system_message() for system roles, got '{role}'")
token_count = self._tokenizer(content)
# Evict oldest messages until budget is satisfied
while not self.is_under_budget and self._messages:
oldest = self._messages[0]
if oldest.role in ("system", "developer"):
break # Never evict system messages
self._dynamic_token_count -= oldest.tokens
self._messages.popleft()
new_msg = ConversationMessage(
role=role,
content=content,
tokens=token_count,
metadata=metadata or {},
)
self._messages.append(new_msg)
self._dynamic_token_count += token_count
return True
def get_history(self) -> list[dict[str, Any]]:
"""Return the full conversation history as a list of API-compatible dicts.
Preserves order: system message first, then chronological messages.
"""
return [msg.to_dict() for msg in self._messages]
def get_recent_messages(self, n: int = 10) -> list[dict[str, Any]]:
"""Return the N most recent messages (excluding system)."""
dynamic = [m for m in reversed(self._messages) if m.role not in ("system", "developer")]
return [m.to_dict() for m in dynamic[:n]]
@property
def total_system_tokens(self) -> int:
"""Sum of tokens used by all system/developer messages."""
return sum(m.tokens for m in self._messages if m.role in ("system", "developer"))
def reset(self) -> None:
"""Clear all dynamic messages, preserving only the system message.
Useful when starting a new sub-task within an agent workflow.
"""
self._messages = deque(
msg for msg in self._messages if msg.role in ("system", "developer")
)
self._dynamic_token_count = 0
def __repr__(self) -> str:
return (
f"BoundedConversationBuffer(messages={len(self._messages)}, "
f"tokens={self.total_token_count}/{self._max_tokens}, "
f"utilization={self.utilization_ratio:.1%})"
)
Checkpoint: After configuring the buffer, verify that is_under_budget returns True and that system messages are preserved after any eviction cycle. The token budget should account for 60-70% of the model's maximum context window, leaving headroom for retrieval-augmented memories and tool output.
Step 2: Implement Context Window Bounds Management
When memory pressure is detected (utilization ratio crosses thresholds), apply graduated responses: log warnings at 70%, begin truncating oldest messages at 85%, and trigger auto-summarization above 95%. Never silently drop messages — always record what was compressed.
"""
Context window bounds management with graduated memory pressure response.
Tracks token usage against model context limits and triggers appropriate
compression actions when thresholds are crossed. Ensures system instructions,
recent conversation, and retrieved memories all fit within budget.
"""
class ContextWindowManager:
"""Manages the agent's active context window with graduated pressure response.
Works alongside BoundedConversationBuffer to enforce hard context limits.
When the buffer approaches capacity, this manager selects the appropriate
compression strategy based on current utilization and available strategies.
"""
# Pressure thresholds as fractions of max_tokens budget
WARN_THRESHOLD: float = 0.70
TRUNCATE_THRESHOLD: float = 0.85
SUMMARIZE_THRESHOLD: float = 0.95
def __init__(
self,
buffer: BoundedConversationBuffer,
summary_handler: Optional[Any] = None,
) -> None:
"""Initialize the context window manager.
Args:
buffer: The bounded conversation buffer to monitor.
summary_handler: Callable that accepts a list of messages and returns
a compressed summary string. Required when SUMMARIZE_THRESHOLD
is reachable. Signature: (messages: list[dict]) -> str.
"""
self._buffer = buffer
self._summary_handler = summary_handler
self._pressure_log: list[dict[str, Any]] = []
def check_and_compress(self) -> dict[str, Any]:
"""Evaluate memory pressure and apply appropriate compression.
Returns a dict with the action taken and its impact on token count.
Possible actions: 'none', 'log_warning', 'truncate_oldest', 'summarize'.
Returns:
Action result describing what compression was applied (if any).
"""
utilization = self._buffer.utilization_ratio
if utilization < self.WARN_THRESHOLD:
return {"action": "none", "utilization": round(utilization, 3)}
if utilization < self.TRUNCATE_THRESHOLD:
action_result = self._log_warning_and_truncate()
return {**action_result, "action": "log_warning"}
if utilization < self.SUMMARIZE_THRESHOLD:
action_result = self._truncate_oldest_messages()
return {**action_result, "action": "truncate_oldest"}
# Above summarize threshold — requires summary handler
if self._summary_handler is None:
logger.error(
"Memory pressure at %.1f%% but no summary handler configured.",
utilization * 100,
)
return {"action": "error", "reason": "no_summary_handler"}
action_result = self._trigger_summarization()
return {**action_result, "action": "summarize"}
def _log_warning_and_truncate(self) -> dict[str, Any]:
"""Log a warning and evict the single oldest dynamic message."""
old_count = self._buffer.dynamic_token_count
if self._buffer.is_under_budget:
return {"tokens_freed": 0}
# Remove oldest non-system message
for msg in list(self._buffer._messages):
if msg.role not in ("system", "developer"):
self._buffer._dynamic_token_count -= msg.tokens
self._buffer._messages.remove(msg)
break
new_count = self._buffer.dynamic_token_count
freed = old_count - new_count
logger.warning(
"Memory pressure at %.1f%% — evicted oldest message, freed %d tokens",
self._buffer.utilization_ratio * 100,
freed,
)
return {"tokens_freed": freed}
def _truncate_oldest_messages(self) -> dict[str, Any]:
"""Evict oldest messages until utilization drops below truncate threshold."""
target_utilization = self.TRUNCATE_THRESHOLD - 0.05 # Small safety margin
target_tokens = int(self._buffer._max_tokens * target_utilization)
old_count = self._buffer.dynamic_token_count
while (
self._buffer.dynamic_token_count > target_tokens
and self._buffer._messages
):
for msg in list(self._buffer._messages):
if msg.role not in ("system", "developer"):
self._buffer._dynamic_token_count -= msg.tokens
self._buffer._messages.remove(msg)
break
else:
break # No more dynamic messages to remove
new_count = self._buffer.dynamic_token_count
freed = old_count - new_count
logger.info(
"Truncated %d messages, freed %d tokens (utilization now %.1f%%)",
max(1, len([m for m in self._buffer._messages if m.role not in ("system", "developer")])),
freed,
(self._buffer.dynamic_token_count / max(self._buffer._max_tokens, 1)) * 100,
)
return {"tokens_freed": freed}
def _trigger_summarization(self) -> dict[str, Any]:
"""Compress old messages via the summary handler.
Keeps recent messages raw and replaces older turns with a structured summary.
The summary preserves: goals discovered, facts learned, pending actions, constraints.
"""
# Separate system messages from dynamic history
static_msgs = [m for m in self._buffer._messages if m.role in ("system", "developer")]
dynamic_msgs = [m.to_dict() for m in self._buffer._messages if m.role not in ("system", "developer")]
if len(dynamic_msgs) < 6:
# Not enough history to meaningfully summarize — just truncate
return self._truncate_oldest_messages()
# Keep the last 4 turns raw; summarize the rest
recent_turns = dynamic_msgs[-4:]
older_turns = dynamic_msgs[:-4]
if not older_turns:
return {"tokens_freed": 0}
try:
summary_text = self._summary_handler(older_turns)
except Exception as exc:
logger.error("Summarization failed: %s — falling back to truncation", exc)
return self._truncate_oldest_messages()
# Replace old messages with a single summary message
old_dynamic_count = len(dynamic_msgs)
summary_msg = ConversationMessage(
role="assistant",
content=summary_text,
tokens=self._buffer._tokenizer(summary_text),
metadata={"source": "auto_summary", "turns_compressed": old_dynamic_count - 4},
)
self._buffer._messages.clear()
for msg in static_msgs:
self._buffer._messages.append(msg)
self._buffer._messages.append(summary_msg)
for msg_dict in recent_turns:
self._buffer._messages.append(ConversationMessage.from_dict(msg_dict))
freed = old_dynamic_count * (self._buffer.dynamic_token_count / max(old_dynamic_count, 1))
return {
"tokens_freed": int(freed),
"turns_compressed": len(older_turns),
"summary_tokens": summary_msg.tokens,
}
def get_pressure_status(self) -> dict[str, Any]:
"""Return current pressure status for monitoring and telemetry.
Returns:
Dict with utilization, threshold levels, and recommended action.
"""
util = self._buffer.utilization_ratio
if util < self.WARN_THRESHOLD:
recommended = "none"
elif util < self.TRUNCATE_THRESHOLD:
recommended = "log_warning_and_truncate"
elif util < self.SUMMARIZE_THRESHOLD:
recommended = "truncate_oldest_messages"
else:
recommended = "trigger_summarization"
return {
"utilization": round(util, 3),
"dynamic_tokens": self._buffer.dynamic_token_count,
"max_tokens": self._buffer._max_tokens,
"message_count": self._buffer.message_count,
"recommended_action": recommended,
}
Checkpoint: After compression, verify the buffer's utilization_ratio has dropped below the target threshold. If summarization was triggered, confirm the summary message includes a source: auto_summary metadata tag for auditing purposes.
Step 3: Implement Auto-Summarization of Old Turns
When the buffer crosses the summarize threshold, compress older conversation turns into a condensed summary that preserves goals, facts, pending actions, and constraints. The summarizer should be lightweight — typically using a smaller model or distilled prompt — to avoid introducing significant latency into the agent's main loop.
"""
Auto-summarization engine for compressing old conversation turns.
Detects memory pressure from bounded buffer utilization and produces
structured summaries of older messages while preserving conversational
continuity through recent turns kept in raw form.
The summarizer preserves four categories of information:
1. User goals and objectives stated during the session
2. Facts discovered or confirmed (data points, configurations)
3. Pending actions and commitments made by the agent
4. Constraints and preferences that shaped earlier decisions
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Any, Optional
@dataclass
class SummaryResult:
"""Structured output from summarizing old conversation turns."""
compressed_text: str
turns_compressed: int
tokens_before: int
tokens_after: int
preserved_categories: list[str]
@property
def compression_ratio(self) -> float:
"""Ratio of before/after token count (>1.0 means reduction)."""
if self.tokens_after == 0:
return 0.0
return self.tokens_before / self.tokens_after
class ConversationSummarizer:
"""Compresses older conversation turns into structured summaries.
Preserves conversational continuity by maintaining recent messages in raw form
while replacing older turns with a condensed summary. The summary is organized
into four categories for efficient retrieval and parsing by downstream agents.
BAD vs GOOD example:
❌ BAD: Naive truncation that drops all context before index N,
destroying goals discovered mid-conversation and any
constraints the user stated early on.
✅ GOOD: Structured compression that preserves semantic intent —
the "what" and "why" of each conversation segment — while
discarding verbose exchanges that don't affect ongoing tasks.
"""
# Minimum turns needed before summarization is meaningful
MIN_TURNS_BEFORE_SUMMARIZE: int = 6
def __init__(
self,
llm_summarizer: Optional[Any] = None,
fallback_max_summary_length: int = 500,
) -> None:
"""Initialize the summarizer.
Args:
llm_summarizer: An LLM client with a chat() method that accepts
[messages: list[dict], model: str] and returns a string.
If None, uses rule-based fallback summarization.
fallback_max_summary_length: Maximum character length for the
rule-based fallback summary when no LLM is available.
"""
self._llm_summarizer = llm_summarizer
self._fallback_max_length = fallback_max_summary_length
def summarize(
self,
older_messages: list[dict[str, Any]],
tokenizer: Optional[callable[[str], int]] = None,
) -> SummaryResult:
"""Compress older conversation turns into a structured summary.
Uses the configured LLM summarizer when available; falls back to
rule-based extraction that preserves key categories from the text.
Args:
older_messages: List of message dicts from older turns (oldest first).
These are the messages to compress. Recent messages should be
kept separate and passed to append() directly on the buffer.
tokenizer: Optional token counter for measuring compression ratio.
Returns:
SummaryResult with compressed text, turn count, and metrics.
Raises:
ValueError: If fewer than MIN_TURNS_BEFORE_SUMMARIZE messages provided.
"""
if len(older_messages) < self.MIN_TURNS_BEFORE_SUMMARIZE:
raise ValueError(
f"Need at least {self.MIN_TURNS_BEFORE_SUMMARIZE} turns to summarize, "
f"got {len(older_messages)}"
)
token_count_fn = tokenizer or (lambda text: len(text) // 4)
# Try LLM-based summarization first
if self._llm_summarizer is not None:
return self._llm_based_summary(older_messages, token_count_fn)
# Rule-based fallback
return self._rule_based_summary(older_messages, token_count_fn)
def _llm_based_summary(
self,
messages: list[dict[str, Any]],
token_count_fn: callable[[str], int],
) -> SummaryResult:
"""Use an LLM to produce a high-fidelity summary of old turns.
The prompt explicitly requests four preservation categories so the
downstream agent can parse structured context from the compressed form.
"""
conversation_text = "\n".join(
f"[{msg['role']}] {msg['content']}" for msg in messages
)
summary_prompt = (
"Summarize the following conversation turns. Preserve exactly four categories:\n"
"1. GOALS: User's stated objectives and tasks to accomplish\n"
"2. FACTS: Confirmed data points, configurations, preferences discovered\n"
"3. ACTIONS: Pending actions or commitments the agent agreed to perform\n"
"4. CONSTRAINTS: Constraints, preferences, or rules established during conversation\n\n"
f"Conversation turns:\n{conversation_text}\n\n"
"Return only the four categories with bullet points under each."
)
try:
result = self._llm_summarizer.chat(
messages=[{"role": "user", "content": summary_prompt}],
model="summary-tiny-model", # Use smallest/fastest model available
max_tokens=512,
)
compressed_text = str(result).strip()
except Exception as exc:
logger.error("LLM summarization failed: %s — using fallback", exc)
return self._rule_based_summary(messages, token_count_fn)
tokens_before = sum(token_count_fn(msg["content"]) for msg in messages)
tokens_after = token_count_fn(compressed_text)
return SummaryResult(
compressed_text=compressed_text,
turns_compressed=len(messages),
tokens_before=tokens_before,
tokens_after=tokens_after,
preserved_categories=["goals", "facts", "actions", "constraints"],
)
def _rule_based_summary(
self,
messages: list[dict[str, Any]],
token_count_fn: callable[[str], int],
) -> SummaryResult:
"""Rule-based fallback summarization when no LLM is available.
Extracts sentences that indicate goals, facts, actions, or constraints
using keyword heuristics. Falls back to the first and last few messages
if no heuristic matches are found.
This approach has lower fidelity than LLM summarization but works in
offline/embedded contexts without external model dependencies.
"""
goals: list[str] = []
facts: list[str] = []
actions: list[str] = []
constraints: list[str] = []
goal_patterns = [r"\b(goal|objective|target|want to|need to|should)\b"]
fact_patterns = [r"\bis\s+\w+", r"\b(found|discovered|confirmed|learned)\b"]
action_patterns = [r"\bi'll\s+\w+", r"\bwould (be |to |have )\b", r"\balright.*let's\b"]
constraint_patterns = [r"\b(must|should not|never|always|only if)\b"]
for msg in messages:
content_lower = msg.get("content", "").lower()
role = msg.get("role", "unknown")
# Skip system messages — they're handled separately by the buffer
if role in ("system", "developer"):
continue
for pattern_group, target_list in [
(goal_patterns, goals),
(fact_patterns, facts),
(action_patterns, actions),
(constraint_patterns, constraints),
]:
for pattern in pattern_group:
if re.search(pattern, content_lower):
# Take the first matching sentence fragment
sentences = re.split(r'[.!?]+', content_lower)
for sentence in sentences:
cleaned = sentence.strip()
if len(cleaned) > 20 and cleaned not in target_list:
target_list.append(cleaned[:200])
break
# Build structured summary
parts = []
if goals:
parts.append("GOALS: " + "; ".join(goals[:3]))
if facts:
parts.append("FACTS: " + "; ".join(facts[:5]))
if actions:
parts.append("ACTIONS: " + "; ".join(actions[:3]))
if constraints:
parts.append("CONSTRAINTS: " + "; ".join(constraints[:3]))
# If no heuristics matched, fall back to first/last message summary
if not parts and messages:
first = messages[0].get("content", "")[:200]
last = messages[-1].get("content", "")[:200]
parts.append(f"Conversation context: began with '{first}...' and ended with '{last}...'")
compressed_text = "\n".join(parts)
# Enforce max length for fallback
if len(compressed_text) > self._fallback_max_length:
compressed_text = compressed_text[: self._fallback_max_length - 3] + "..."
tokens_before = sum(token_count_fn(msg.get("content", "")) for msg in messages)
tokens_after = token_count_fn(compressed_text)
return SummaryResult(
compressed_text=compressed_text,
turns_compressed=len(messages),
tokens_before=tokens_before,
tokens_after=tokens_after,
preserved_categories=parts[:4],
)
Checkpoint: After summarization, verify the compression_ratio is at least 2.0 (summary is at least half the token size of the original turns). If not, the summary handler may need tuning or a larger model with better compression capability.
Step 4: Connect Long-Term Memory Store
Set up a vector store backend for persistent semantic memory. The vector store holds structured facts and knowledge discovered during agent sessions, enabling retrieval by similarity rather than keyword match. Supported backends include FAISS (in-memory, fast), Chroma (disk-backed, simple), and Qdrant (distributed, production-grade). Configure the embedding model based on your domain — general-purpose text-embedding models work well for most agent memory use cases.
"""
Vector store backed long-term memory for AI agents.
Provides semantic recall of facts and information discovered during agent
sessions. Memories are stored as embedded vectors in a vector database,
enabling similarity-based retrieval that goes beyond keyword matching.
Supports three backends: FAISS (fast in-memory), Chroma (simple disk-backed),
and Qdrant (production-grade distributed). Uses a unified interface so the
agent code doesn't need to know which backend is active.
"""
from __future__ import annotations
import logging
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional
logger = logging.getLogger(__name__)
@dataclass
class StoredMemory:
"""A single memory entry in the vector store.
Memories are text chunks that have been embedded and stored with metadata
for filtering and retrieval. Each memory has a unique ID, creation timestamp,
access count (for recency bias), and associated tags.
"""
memory_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
content: str = ""
metadata: dict[str, Any] = field(default_factory=dict)
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
last_accessed: Optional[datetime] = None
access_count: int = 0
@property
def age_hours(self) -> float:
"""Hours since this memory was created."""
return (datetime.now(timezone.utc) - self.created_at).total_seconds() / 3600
def record_access(self) -> None:
"""Record a retrieval access for recency scoring."""
self.access_count += 1
self.last_accessed = datetime.now(timezone.utc)
@dataclass
class RetrievalResult:
"""A single result from querying the vector store.
Attributes:
memory: The stored memory matched to the query.
similarity_score: Floating point score (0.0-1.0) indicating relevance.
rank: Position in the sorted results (1-based).
"""
memory: StoredMemory
similarity_score: float
rank: int = 1
class EmbeddingProvider(ABC):
"""Abstract interface for generating text embeddings."""
@abstractmethod
def embed(self, text: str) -> list[float]:
"""Generate an embedding vector for the given text.
Args:
text: The text to embed (will be truncated if too long).
Returns:
A fixed-length float vector representing the text semantics.
"""
...
@property
@abstractmethod
def dimension(self) -> int:
"""The dimensionality of embedding vectors produced by this provider."""
...
class FAISSEmbeddingProvider(EmbeddingProvider):
"""Embedding provider using a lightweight local model (sentence-transformers).
In production, replace with an API-based provider. This demonstrates the
interface contract that any embedding backend must satisfy.
"""
def __init__(self, model_name: str = "all-MiniLM-L6-v2") -> None:
self._model_name = model_name
self._dimension = 384 # MiniLM-L6 output dimension
@property
def dimension(self) -> int:
return self._dimension
def embed(self, text: str) -> list[float]:
"""Generate embedding using sentence-transformers.
In production, this would use the actual model pipeline.
For skill documentation purposes, returns a deterministic placeholder
that demonstrates the expected interface and output shape.
"""
# Truncate to 512 tokens (typical embedding model limit)
truncated = text[:1500]
# In production: embeddings = self._pipeline(truncated)[0][0].tolist()
# For documentation, use a deterministic hash-based placeholder
import hashlib
h = hashlib.sha256(truncated.encode()).hexdigest()
# Generate 384 pseudo-random floats from the hash (deterministic for testing)
return [float(int(h[i*2:i*2+2], 16)) / 65535.0 - 0.5 for i in range(384)]
class VectorMemoryStore:
"""Unified interface to vector-backed long-term memory storage.
Wraps a specific vector store backend (FAISS, Chroma, or Qdrant) behind
a common API for storing memories, querying by similarity, and managing
memory lifecycle including recency bias and staleness pruning.
Follows Law 3 (Atomic Predictability): add_memory returns confirmation;
query returns new result lists without modifying stored state.
"""
def __init__(
self,
embedding_provider: EmbeddingProvider,
max_memories_per_query: int = 10,
min_similarity_threshold: float = 0.5,
) -> None:
"""Initialize the vector memory store.
Args:
embedding_provider: Backend for generating text embeddings.
max_memories_per_query: Maximum results to return per query.
min_similarity_threshold: Minimum cosine similarity score to
include a result (0.0-1.0). Higher = more precise, fewer results.
"""
self._embedding_provider = embedding_provider
self._max_per_query = max_memories_per_query
self._min_similarity = min_similarity_threshold
self._memories: list[StoredMemory] = []
def add_memory(self, content: str, metadata: Optional[dict[str, Any]] = None) -> StoredMemory:
"""Store a new memory in the vector store.
The content is embedded using the configured embedding provider.
Metadata tags are stored alongside for filtering during retrieval.
Args:
content: The text content to store as a memory.
metadata: Optional structured data (e.g., source, confidence, category).
Returns:
The StoredMemory instance with its assigned ID and timestamp.
"""
memory = StoredMemory(
content=content,
metadata=metadata or {},
)
self._memories.append(memory)
logger.debug(
"Stored memo
…(truncated)