Memory Module

Memory management utilities for LLM applications.

This module provides comprehensive memory management for conversational AI:

Core Classes:

ConversationBuffer - Main class for managing conversation memory Entity - Represents an extracted entity ConversationSummary - Represents a conversation summary

Common Functions:

create_sliding_window() - Recent messages window create_token_limited_window() - Token-limited window create_alternating_window() - Alternating user/assistant pairs

Submodules:

buffers - Conversation buffer management and sliding windows summaries - Summary-based memory functions entities - Entity extraction and tracking utils - Utility functions (format, filter, merge, persistence) patterns - Advanced memory patterns (semantic, episodic) classes - Data classes (Entity, ConversationSummary)

Examples

>>> # Common usage - top-level imports
>>> from kerb.memory import ConversationBuffer, Entity
>>> buffer = ConversationBuffer()
>>> buffer.add_message("user", "Hello!")

>>> # Specialized usage - submodule imports
>>> from kerb.memory.summaries import create_progressive_summary
>>> from kerb.memory.entities import extract_entities
>>> from kerb.memory.patterns import create_semantic_memory

class kerb.memory.ConversationBuffer(max_messages=100, window_size=10, enable_summaries=True, enable_entity_tracking=True)

Bases: object

Manages conversation history with multiple memory strategies.

__init__(max_messages=100, window_size=10, enable_summaries=True, enable_entity_tracking=True)

Initialize conversation buffer.

Parameters:

• max_messages (int) – Maximum messages to store

• window_size (int) – Size of sliding window

• enable_summaries (bool) – Whether to create summaries

• enable_entity_tracking (bool) – Whether to track entities

add_message(role, content, metadata=None)

Add a message to the buffer.

Parameters:

• role (str) – Message role (“user”, “assistant”, “system”)

• content (str) – Message content

• metadata (Optional[Dict]) – Optional metadata

Returns:

The created message

Return type:

Message

get_recent_messages(count=None)

Get recent messages (sliding window).

Parameters:

count (Optional[int]) – Number of messages (defaults to window_size)

Returns:

Recent messages

Return type:

List[Message]

get_context(max_tokens=None, include_summary=True)

Get conversation context as formatted string.

Parameters:

• max_tokens (Optional[int]) – Maximum tokens (if None, use all recent messages)

• include_summary (bool) – Whether to include summaries of old messages

Returns:

Formatted conversation context

Return type:

str

get_entities(min_mentions=1)

Get tracked entities.

Parameters:

min_mentions (int) – Minimum mentions to include

Returns:

List of entities

Return type:

List[Entity]

search_messages(query, max_results=10)

Search messages by content.

Parameters:

• query (str) – Search query

• max_results (int) – Maximum results to return

Returns:

Matching messages

Return type:

List[Message]

clear(keep_summaries=True)

Clear the buffer.

Parameters:

keep_summaries (bool) – Whether to keep summaries

to_dict()

Export buffer to dictionary.

Return type:

Dict[str, Any]

from_dict(data)

Load buffer from dictionary.

save(filepath)

Save conversation buffer to file.

Parameters:

filepath (str) – Path to save to (JSON format)

Return type:

None

Example

>>> buffer.save("conversation.json")

classmethod load(filepath)

Load conversation buffer from file.

Parameters:

filepath (str) – Path to load from

Returns:

Loaded buffer

Return type:

ConversationBuffer

Example

>>> buffer = ConversationBuffer.load("conversation.json")

prune(strategy='oldest', keep_count=None, keep_percentage=None)

Prune messages from buffer using various strategies.

Parameters:

• strategy (Union[PruneStrategy, str]) – Pruning strategy (PruneStrategy enum or string: “oldest”, “newest”, “least_relevant”, “most_relevant”, “token_limit”)

• keep_count (Optional[int]) – Number of messages to keep

• keep_percentage (Optional[float]) – Percentage of messages to keep (0-1)

Returns:

Self (for method chaining, modifies in place)

Return type:

ConversationBuffer

Examples

>>> from kerb.core.enums import PruneStrategy
>>> buffer.prune(strategy=PruneStrategy.OLDEST, keep_count=50)

class kerb.memory.Entity(name, type, mentions=1, first_seen=None, last_seen=None, context=<factory>, metadata=<factory>)

Bases: object

Represents an extracted entity with metadata.

name: str

type: str

mentions: int = 1

first_seen: str | None = None

last_seen: str | None = None

context: List[str]

metadata: Dict[str, Any]

to_dict()

Convert entity to dictionary.

Return type:

Dict[str, Any]

classmethod from_dict(data)

Create entity from dictionary.

Return type:

Entity

__init__(name, type, mentions=1, first_seen=None, last_seen=None, context=<factory>, metadata=<factory>)

class kerb.memory.ConversationSummary(summary, message_count, start_time, end_time, key_points=<factory>, entities=<factory>, metadata=<factory>)

Bases: object

Represents a summary of conversation history.

summary: str

message_count: int

start_time: str

end_time: str

key_points: List[str]

entities: List[str]

metadata: Dict[str, Any]

to_dict()

Convert summary to dictionary.

Return type:

Dict[str, Any]

classmethod from_dict(data)

Create summary from dictionary.

Return type:

ConversationSummary

__init__(summary, message_count, start_time, end_time, key_points=<factory>, entities=<factory>, metadata=<factory>)

kerb.memory.create_sliding_window(messages, window_size=10, include_system=True)

Create a sliding window of recent messages.

Parameters:

• messages (List[Message]) – List of conversation messages

• window_size (int) – Number of recent messages to keep

• include_system (bool) – Whether to include system messages in the window

Returns:

Most recent messages within window

Return type:

List[Message]

Example

>>> messages = [Message("user", "Hello"), Message("assistant", "Hi")]
>>> recent = create_sliding_window(messages, window_size=5)

kerb.memory.create_token_limited_window(messages, max_tokens=2000, token_estimator=None)

Create a sliding window limited by token count.

Parameters:

• messages (List[Message]) – List of conversation messages

• max_tokens (int) – Maximum total tokens

• token_estimator (Optional[Callable[[str], int]]) – Function to estimate tokens (defaults to word count / 0.75)

Returns:

Most recent messages that fit within token limit

Return type:

List[Message]

Example

>>> window = create_token_limited_window(messages, max_tokens=1000)

kerb.memory.create_alternating_window(messages, pairs=5)

Create a window with alternating user/assistant pairs.

Parameters:

• messages (List[Message]) – List of conversation messages

• pairs (int) – Number of user-assistant pairs to keep

Returns:

Recent alternating message pairs

Return type:

List[Message]

Example

>>> window = create_alternating_window(messages, pairs=3)

kerb.memory.create_progressive_summary(messages, existing_summary=None, summary_length='medium')

Create a progressive summary of conversation messages.

Parameters:

• messages (List[Message]) – Messages to summarize

• existing_summary (str) – Previous summary to build upon

• summary_length (str) – “short”, “medium”, or “long”

Returns:

Summary of the messages

Return type:

str

Example

>>> summary = create_progressive_summary(messages, summary_length="short")

kerb.memory.summarize_conversation(messages, summary_strategy='extractive', key_points=5)

Create a comprehensive conversation summary.

Parameters:

• messages (List[Message]) – Messages to summarize

• summary_strategy (Union[SummaryStrategy, str]) – Summary strategy (SummaryStrategy enum or string: “extractive”, “abstractive”, “combined”)

• key_points (int) – Number of key points to extract

Returns:

Structured summary of conversation

Return type:

ConversationSummary

Examples

>>> from kerb.core.enums import SummaryStrategy
>>> summary = summarize_conversation(messages, summary_strategy=SummaryStrategy.EXTRACTIVE)

kerb.memory.create_hierarchical_summary(messages, chunk_size=10)

Create hierarchical summaries of conversation chunks.

Parameters:

• messages (List[Message]) – Messages to summarize

• chunk_size (int) – Number of messages per chunk

Returns:

Summary for each chunk

Return type:

List[ConversationSummary]

Example

>>> summaries = create_hierarchical_summary(messages, chunk_size=5)

kerb.memory.save_conversation(buffer, filepath)

Save conversation buffer to a JSON file.

Parameters:

• buffer (ConversationBuffer) – Conversation buffer to save

• filepath (Union[str, Path]) – Path to save file

Return type:

None

Example

>>> save_conversation(buffer, "conversation.json")

kerb.memory.load_conversation(filepath)

Load conversation buffer from a JSON file.

Parameters:

filepath (Union[str, Path]) – Path to load file from

Returns:

Loaded buffer

Return type:

ConversationBuffer

Example

>>> buffer = load_conversation("conversation.json")

Conversation memory and entity tracking for stateful applications.