Skip to content

Memory System

The Memory System in Bedrock Swarm provides context management and message history for agents. It enables agents to maintain state and recall previous interactions.

BaseMemory

bedrock_swarm.memory.base.BaseMemory

Base class for memory implementations.

Functions

add_message(message: Message) -> None

Add a message to memory.

PARAMETER DESCRIPTION
message

Message to add

TYPE: Message

Source code in src/bedrock_swarm/memory/base.py 
42
43
44
45
46
47
48
def add_message(self, message: Message) -> None:
    """Add a message to memory.

    Args:
        message: Message to add
    """
    raise NotImplementedError

get_messages(thread_id: Optional[str] = None) -> List[Message]

Get messages from memory.

PARAMETER DESCRIPTION
thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
List[Message]

List of messages in chronological order
Source code in src/bedrock_swarm/memory/base.py 
50
51
52
53
54
55
56
57
58
59
def get_messages(self, thread_id: Optional[str] = None) -> List[Message]:
    """Get messages from memory.

    Args:
        thread_id: Optional thread ID to filter messages

    Returns:
        List of messages in chronological order
    """
    raise NotImplementedError

get_last_message(thread_id: Optional[str] = None) -> Optional[Message]

Get the most recent message.

PARAMETER DESCRIPTION
thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Optional[Message]

Most recent message or None if no messages
Source code in src/bedrock_swarm/memory/base.py 
61
62
63
64
65
66
67
68
69
70
def get_last_message(self, thread_id: Optional[str] = None) -> Optional[Message]:
    """Get the most recent message.

    Args:
        thread_id: Optional thread ID to filter messages

    Returns:
        Most recent message or None if no messages
    """
    raise NotImplementedError

clear() -> None

Clear all messages from memory.

Source code in src/bedrock_swarm/memory/base.py 
72
73
74
def clear(self) -> None:
    """Clear all messages from memory."""
    raise NotImplementedError

SimpleMemory

bedrock_swarm.memory.base.SimpleMemory(max_size: int = 1000)

Bases: BaseMemory

Simple in-memory implementation with thread support and shared state.

PARAMETER DESCRIPTION
max_size

Maximum number of messages to store per thread

TYPE: int DEFAULT: 1000

Source code in src/bedrock_swarm/memory/base.py 
80
81
82
83
84
85
86
87
88
def __init__(self, max_size: int = 1000) -> None:
    """Initialize SimpleMemory.

    Args:
        max_size: Maximum number of messages to store per thread
    """
    self._messages: Dict[str, List[Message]] = {}  # thread_id -> messages
    self._max_size = max_size
    self.shared_state = SharedState()

Functions

add_message(message: Message) -> None

Add a message to memory.

If max_size is reached, oldest messages are removed.

PARAMETER DESCRIPTION
message

Message to add

TYPE: Message

Source code in src/bedrock_swarm/memory/base.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def add_message(self, message: Message) -> None:
    """Add a message to memory.

    If max_size is reached, oldest messages are removed.

    Args:
        message: Message to add
    """
    thread_id = message.thread_id or "default"
    if thread_id not in self._messages:
        self._messages[thread_id] = []

    self._messages[thread_id].append(message)

    # Enforce size limit per thread
    if len(self._messages[thread_id]) > self._max_size:
        self._messages[thread_id] = self._messages[thread_id][-self._max_size :]

get_messages(thread_id: Optional[str] = None) -> List[Message]

Get messages from memory.

PARAMETER DESCRIPTION
thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
List[Message]

List of messages in chronological order
Source code in src/bedrock_swarm/memory/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get_messages(self, thread_id: Optional[str] = None) -> List[Message]:
    """Get messages from memory.

    Args:
        thread_id: Optional thread ID to filter messages

    Returns:
        List of messages in chronological order
    """
    if thread_id:
        return self._messages.get(thread_id, []).copy()

    # If no thread_id, return all messages sorted by timestamp
    all_messages = []
    for messages in self._messages.values():
        all_messages.extend(messages)
    return sorted(all_messages, key=lambda m: m.timestamp)

get_last_message(thread_id: Optional[str] = None) -> Optional[Message]

Get the most recent message.

PARAMETER DESCRIPTION
thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Optional[Message]

Most recent message or None if no messages
Source code in src/bedrock_swarm/memory/base.py 
126
127
128
129
130
131
132
133
134
135
136
def get_last_message(self, thread_id: Optional[str] = None) -> Optional[Message]:
    """Get the most recent message.

    Args:
        thread_id: Optional thread ID to filter messages

    Returns:
        Most recent message or None if no messages
    """
    messages = self.get_messages(thread_id)
    return messages[-1] if messages else None

get_messages_by_type(message_type: str, thread_id: Optional[str] = None) -> List[Message]

Get messages of a specific type from memory.

PARAMETER DESCRIPTION
message_type

Type of messages to retrieve (e.g., 'tool_result', 'user_message')

TYPE: str

thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
List[Message]

List of messages of the specified type in chronological order
Source code in src/bedrock_swarm/memory/base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def get_messages_by_type(
    self, message_type: str, thread_id: Optional[str] = None
) -> List[Message]:
    """Get messages of a specific type from memory.

    Args:
        message_type: Type of messages to retrieve (e.g., 'tool_result', 'user_message')
        thread_id: Optional thread ID to filter messages

    Returns:
        List of messages of the specified type in chronological order
    """
    messages = self.get_messages(thread_id)
    return [
        msg
        for msg in messages
        if msg.metadata and msg.metadata.get("type") == message_type
    ]

get_tool_results(thread_id: Optional[str] = None, limit: Optional[int] = None) -> List[Message]

Get tool execution results from memory.

PARAMETER DESCRIPTION
thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

limit

Optional limit on number of results to return (most recent first)

TYPE: Optional[int] DEFAULT: None

RETURNS DESCRIPTION
List[Message]

List of tool result messages in chronological order
Source code in src/bedrock_swarm/memory/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def get_tool_results(
    self, thread_id: Optional[str] = None, limit: Optional[int] = None
) -> List[Message]:
    """Get tool execution results from memory.

    Args:
        thread_id: Optional thread ID to filter messages
        limit: Optional limit on number of results to return (most recent first)

    Returns:
        List of tool result messages in chronological order
    """
    tool_results = self.get_messages_by_type("tool_result", thread_id)
    if limit:
        return tool_results[-limit:]
    return tool_results

get_conversation_summary(thread_id: Optional[str] = None, limit: int = 5) -> List[Dict[str, str]]

Get a summary of the recent conversation.

PARAMETER DESCRIPTION
thread_id

Optional thread ID to filter messages

TYPE: Optional[str] DEFAULT: None

limit

Number of most recent message pairs to include

TYPE: int DEFAULT: 5

RETURNS DESCRIPTION
List[Dict[str, str]]

List of message pairs (user/assistant) with their content
Source code in src/bedrock_swarm/memory/base.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def get_conversation_summary(
    self, thread_id: Optional[str] = None, limit: int = 5
) -> List[Dict[str, str]]:
    """Get a summary of the recent conversation.

    Args:
        thread_id: Optional thread ID to filter messages
        limit: Number of most recent message pairs to include

    Returns:
        List of message pairs (user/assistant) with their content
    """
    messages = self.get_messages(thread_id)
    summary = []

    # Process messages from newest to oldest, up to limit pairs
    i = len(messages) - 1
    while i >= 0 and len(summary) < limit:
        msg = messages[i]

        if msg.role == "user":
            # Look ahead for assistant response
            if i + 1 < len(messages) and messages[i + 1].role == "assistant":
                summary.insert(
                    0,
                    {
                        "user": msg.content,
                        "assistant": messages[i + 1].content,
                        "has_tool_calls": bool(
                            messages[i + 1].metadata
                            and messages[i + 1].metadata.get("type")
                            == "tool_call_intent"
                        ),
                    },
                )
            else:
                # No assistant response found
                summary.insert(0, {"user": msg.content, "assistant": None})
        i -= 1

    return summary[-limit:]

clear() -> None

Clear all messages from memory.

Source code in src/bedrock_swarm/memory/base.py 
216
217
218
219
def clear(self) -> None:
    """Clear all messages from memory."""
    self._messages.clear()
    self.shared_state.clear()

clear_thread(thread_id: str) -> None

Clear all messages from a specific thread.

PARAMETER DESCRIPTION
thread_id

ID of the thread to clear

TYPE: str

Source code in src/bedrock_swarm/memory/base.py 
221
222
223
224
225
226
227
228
def clear_thread(self, thread_id: str) -> None:
    """Clear all messages from a specific thread.

    Args:
        thread_id: ID of the thread to clear
    """
    if thread_id in self._messages:
        del self._messages[thread_id]

get_thread_ids() -> List[str]

Get list of all thread IDs in memory.

RETURNS DESCRIPTION
List[str]

List of thread IDs
Source code in src/bedrock_swarm/memory/base.py 
230
231
232
233
234
235
236
def get_thread_ids(self) -> List[str]:
    """Get list of all thread IDs in memory.

    Returns:
        List of thread IDs
    """
    return list(self._messages.keys())

Message

bedrock_swarm.memory.base.Message(role: str, content: str, timestamp: datetime, thread_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) dataclass

Message class for storing conversation history.

Usage Example

from bedrock_swarm.memory.base import SimpleMemory

# Create a memory instance
memory = SimpleMemory()

# Add messages to memory
memory.add_message("user", "What's the weather like?")
memory.add_message("assistant", "It's sunny today.")

# Get message history
history = memory.get_history()

Best Practices

  1. Use SimpleMemory for basic message history tracking
  2. Extend BaseMemory for custom memory implementations
  3. Implement message filtering and context window management
  4. Consider memory persistence for long-running agents
  5. Handle message types consistently across the system