autogen_ext.memory.mem0

class Mem0Memory(user_id: str | None = None, limit: int = 10, is_cloud: bool = True, api_key: str | None = None, config: Dict[str, Any] | None = None)

Bases: Memory, Component[Mem0MemoryConfig], ComponentBase[Mem0MemoryConfig]

Mem0 memory implementation for AutoGen.

This component integrates with Mem0.ai’s memory system, providing an implementation of AutoGen’s Memory interface. It supports both cloud and local backends through the mem0ai Python package.

To use this component, you need to have the mem0 (for cloud-only) or mem0-local (for local) extra installed for the autogen-ext package:

pip install -U "autogen-ext[mem0]" # For cloud-based Mem0
pip install -U "autogen-ext[mem0-local]" # For local Mem0

The memory component can store and retrieve information that agents need to remember across conversations. It also provides context updating for language models with relevant memories.

Examples

import asyncio
from autogen_ext.memory.mem0 import Mem0Memory
from autogen_core.memory import MemoryContent


async def main() -> None:
    # Create a local Mem0Memory (no API key required)
    memory = Mem0Memory(
        is_cloud=False,
        config={"path": ":memory:"},  # Use in-memory storage for testing
    )
    print("Memory initialized successfully!")

    # Add something to memory
    test_content = "User likes the color blue."
    await memory.add(MemoryContent(content=test_content, mime_type="text/plain"))
    print(f"Added content: {test_content}")

    # Retrieve memories with a search query
    results = await memory.query("What color does the user like?")
    print(f"Query results: {len(results.results)} found")

    for i, result in enumerate(results.results):
        print(f"Result {i+1}: {result}")


asyncio.run(main())

Output:

Memory initialized successfully!
Added content: User likes the color blue.
Query results: 1 found
Result 1: content='User likes the color blue' mime_type='text/plain' metadata={'score': 0.6977155806281953, 'created_at': datetime.datetime(2025, 7, 6, 17, 25, 18, 754725, tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=61200)))}

Using it with an AssistantAgent:

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_core.memory import MemoryContent
from autogen_ext.memory.mem0 import Mem0Memory
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    # Create a model client
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    # Create a Mem0 memory instance
    memory = Mem0Memory(
        user_id="user123",
        is_cloud=False,
        config={"path": ":memory:"},  # Use in-memory storage for testing
    )

    # Add something to memory
    test_content = "User likes the color blue."
    await memory.add(MemoryContent(content=test_content, mime_type="text/plain"))

    # Create an assistant agent with Mem0 memory
    agent = AssistantAgent(
        name="assistant",
        model_client=model_client,
        memory=[memory],
        system_message="You are a helpful assistant that remembers user preferences.",
    )

    # Run a sample task
    result = await agent.run(task="What color does the user like?")
    print(result.messages[-1].content)  # type: ignore


asyncio.run(main())

Output:

User likes the color blue.

Parameters:

• user_id – Optional user ID for memory operations. If not provided, a UUID will be generated.

• limit – Maximum number of results to return in memory queries.

• is_cloud – Whether to use cloud Mem0 client (True) or local client (False).

• api_key – API key for cloud Mem0 client. It will read from the environment MEM0_API_KEY if not provided.

• config – Configuration dictionary for local Mem0 client. Required if is_cloud=False.

component_type: ClassVar[ComponentType] = 'memory'

The logical type of the component.

component_provider_override: ClassVar[str | None] = 'autogen_ext.memory.mem0.Mem0Memory'

Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

component_config_schema

alias of Mem0MemoryConfig

property user_id: str

Get the user ID for memory operations.

property limit: int

Get the maximum number of results to return in memory queries.

property is_cloud: bool

Check if the Mem0 client is cloud-based.

property config: Dict[str, Any] | None

Get the configuration for the Mem0 client.

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

Add content to memory.

Parameters:

• content – The memory content to add.

• cancellation_token – Optional token to cancel operation.

Raises:

Exception – If there’s an error adding content to mem0 memory.

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

Query memory for relevant content.

Parameters:

• query – The query to search for, either as string or MemoryContent.

• cancellation_token – Optional token to cancel operation.

• **kwargs – Additional query parameters to pass to mem0.

Returns:

MemoryQueryResult containing search results.

async update_context(model_context: ChatCompletionContext) → UpdateContextResult

Update the model context with relevant memories.

This method retrieves the conversation history from the model context, uses the last message as a query to find relevant memories, and then adds those memories to the context as a system message.

Parameters:

model_context – The model context to update.

Returns:

UpdateContextResult containing memories added to the context.

async clear() → None

Clear all content from memory for the current user.

Raises:

Exception – If there’s an error clearing mem0 memory.

async close() → None

Clean up resources if needed.

This is a no-op for Mem0 clients as they don’t require explicit cleanup.

pydantic model Mem0MemoryConfig

Bases: BaseModel

Configuration for Mem0Memory component.

Show JSON schema

{
   "title": "Mem0MemoryConfig",
   "description": "Configuration for Mem0Memory component.",
   "type": "object",
   "properties": {
      "user_id": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "User ID for memory operations. If not provided, a UUID will be generated.",
         "title": "User Id"
      },
      "limit": {
         "default": 10,
         "description": "Maximum number of results to return in memory queries.",
         "title": "Limit",
         "type": "integer"
      },
      "is_cloud": {
         "default": true,
         "description": "Whether to use cloud Mem0 client (True) or local client (False).",
         "title": "Is Cloud",
         "type": "boolean"
      },
      "api_key": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "API key for cloud Mem0 client. Required if is_cloud=True.",
         "title": "Api Key"
      },
      "config": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Configuration dictionary for local Mem0 client. Required if is_cloud=False.",
         "title": "Config"
      }
   }
}

Fields:

• api_key (str | None)

• config (Dict[str, Any] | None)

• is_cloud (bool)

• limit (int)

• user_id (str | None)

field user_id: str | None = None

User ID for memory operations. If not provided, a UUID will be generated.

field limit: int = 10

Maximum number of results to return in memory queries.

field is_cloud: bool = True

Whether to use cloud Mem0 client (True) or local client (False).

field api_key: str | None = None

API key for cloud Mem0 client. Required if is_cloud=True.

field config: Dict[str, Any] | None = None

Configuration dictionary for local Mem0 client. Required if is_cloud=False.