memory

class agentopera.memory.Memory

Bases: ABC

Protocol defining the interface for memory implementations.

A memory is the storage for data that can be used to enrich or modify the model context.

A memory implementation can use any storage mechanism, such as a list, a database, or a file system. It can also use any retrieval mechanism, such as vector search or text search. It is up to the implementation to decide how to store and retrieve data.

It is also a memory implementation’s responsibility to update the model context with relevant memory content based on the current model context and querying the memory store.

See ListMemory for an example implementation.

abstract async update_context(model_context: ChatCompletionContext) → UpdateContextResult

Update the provided model context using relevant memory content.

Parameters:

model_context – The context to update.

Returns:

UpdateContextResult containing relevant memories

abstract async query(query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any) → MemoryQueryResult

Query the memory store and return relevant entries.

Parameters:

• query – Query content item

• cancellation_token – Optional token to cancel operation

• **kwargs – Additional implementation-specific parameters

Returns:

MemoryQueryResult containing memory entries with relevance scores

abstract async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None

Add a new content to memory.

Parameters:

• content – The memory content to add

• cancellation_token – Optional token to cancel operation

abstract async clear() → None

Clear all entries from memory.

abstract async close() → None

Clean up any resources used by the memory implementation.

class agentopera.memory.MemoryContent(*, content: str | bytes | Dict[str, Any] | Image, mime_type: MemoryMimeType | str, metadata: Dict[str, Any] | None = None)

Bases: BaseModel

A memory content item.

content: str | bytes | Dict[str, Any] | Image

The content of the memory item. It can be a string, bytes, dict, or Image.

mime_type: MemoryMimeType | str

The MIME type of the memory content.

metadata: Dict[str, Any] | None

Metadata associated with the memory item.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

serialize_mime_type(mime_type: MemoryMimeType | str) → str

Serialize the MIME type to a string.

class agentopera.memory.MemoryQueryResult(*, results: List[MemoryContent])

Bases: BaseModel

Result of a memory query() operation.

results: List[MemoryContent]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentopera.memory.UpdateContextResult(*, memories: MemoryQueryResult)

Bases: BaseModel

Result of a memory update_context() operation.

memories: MemoryQueryResult

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentopera.memory.MemoryMimeType(value)

Bases: Enum

Supported MIME types for memory content.

TEXT = 'text/plain'

JSON = 'application/json'

MARKDOWN = 'text/markdown'

IMAGE = 'image/*'

BINARY = 'application/octet-stream'

class agentopera.memory.ListMemory(name: str | None = None, memory_contents: List[MemoryContent] | None = None)

Bases: Memory

Simple chronological list-based memory implementation.

This memory implementation stores contents in a list and retrieves them in chronological order. It has an update_context method that updates model contexts by appending all stored memories.

The memory content can be directly accessed and modified through the content property, allowing external applications to manage memory contents directly.

Example

import asyncio
from agentopera.core.memory import ListMemory, MemoryContent
from agentopera.core.model_context import BufferedChatCompletionContext


async def main() -> None:
    # Initialize memory
    memory = ListMemory(name="chat_history")

    # Add memory content
    content = MemoryContent(content="User prefers formal language", mime_type="text/plain")
    await memory.add(content)

    # Directly modify memory contents
    memory.content = [MemoryContent(content="New preference", mime_type="text/plain")]

    # Create a model context
    model_context = BufferedChatCompletionContext(buffer_size=10)

    # Update a model context with memory
    await memory.update_context(model_context)

    # See the updated model context
    print(await model_context.get_messages())


asyncio.run(main())

Parameters:

name – Optional identifier for this memory instance

property name: str

Get the memory instance identifier.

Returns:

Memory instance name

Return type:

str

property content: List[MemoryContent]

Get the current memory contents.

Returns:

List of stored memory contents

Return type:

List[MemoryContent]

async update_context(model_context: ChatCompletionContext) → UpdateContextResult

Update the model context by appending memory content.

This method mutates the provided model_context by adding all memories as a SystemMessage.

Parameters:

model_context – The context to update. Will be mutated if memories exist.

Returns:

UpdateContextResult containing the memories that were added to the context

async query(query: str | MemoryContent = '', cancellation_token: CancellationToken | None = None, **kwargs: Any) → MemoryQueryResult

Return all memories without any filtering.

Parameters:

• query – Ignored in this implementation

• cancellation_token – Optional token to cancel operation

• **kwargs – Additional parameters (ignored)

Returns:

MemoryQueryResult containing all stored memories

async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None

Add new content to memory.

Parameters:

• content – Memory content to store

• cancellation_token – Optional token to cancel operation

async clear() → None

Clear all memory content.

async close() → None

Cleanup resources if needed.

• memory.graphrag