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)[source]#

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

AutoGen 的 Mem0 内存实现。

此组件与 Mem0.ai 的内存系统集成，提供了 AutoGen 内存接口的实现。它通过 mem0ai Python 包支持云端和本地后端。

要使用此组件，您需要为 autogen-ext 包安装 mem0（仅限云端）或 mem0-local（本地）附加项。

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

内存组件可以存储和检索代理在对话中需要记住的信息。它还为语言模型提供相关内存的上下文更新。

示例

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())

输出

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)))}

将其与 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())

输出

User likes the color blue.

参数:

user_id – 内存操作的可选用户 ID。如果未提供，将生成一个 UUID。
limit – 内存查询中返回结果的最大数量。
is_cloud – 是否使用云端 Mem0 客户端 (True) 或本地客户端 (False)。
api_key – 云端 Mem0 客户端的 API 密钥。如果未提供，将从环境变量 MEM0_API_KEY 读取。
config – 本地 Mem0 客户端的配置字典。如果 is_cloud=False，则为必需。

component_type: ClassVar[ComponentType] = 'memory'#: 组件的逻辑类型。

component_provider_override: ClassVar[str | None] = 'autogen_ext.memory.mem0.Mem0Memory'#: 覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

component_config_schema#: 别名 Mem0MemoryConfig

property user_id: str#: 获取用于内存操作的用户 ID。

property limit: int#: 获取内存查询中返回结果的最大数量。

property is_cloud: bool#: 检查 Mem0 客户端是否基于云端。

property config: Dict[str, Any] | None#: 获取 Mem0 客户端的配置。

async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None[source]#

将内容添加到内存。

参数:

content – 要添加的内存内容。
cancellation_token – 可选的取消操作令牌。

抛出:

Exception – 如果将内容添加到 mem0 内存时发生错误。

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

查询内存以获取相关内容。

参数:

query – 要搜索的查询，可以是字符串或 MemoryContent。
cancellation_token – 可选的取消操作令牌。
**kwargs – 要传递给 mem0 的附加查询参数。

返回:

包含搜索结果的 MemoryQueryResult。

async update_context(model_context: ChatCompletionContext) → UpdateContextResult[source]#

使用相关内存更新模型上下文。

此方法从模型上下文中检索对话历史记录，使用最后一条消息作为查询来查找相关内存，然后将这些内存作为系统消息添加到上下文中。

参数:: model_context – 要更新的模型上下文。
返回:: UpdateContextResult 包含添加到上下文的内存。

async clear() → None[source]#

清除当前用户内存中的所有内容。

抛出:: Exception – 如果清除 mem0 内存时发生错误。

async close() → None[source]#

如果需要，清理资源。

对于 Mem0 客户端，这是一个无操作，因为它们不需要显式清理。

pydantic model Mem0MemoryConfig[source]#

基类: BaseModel

Mem0Memory 组件的配置。

显示 JSON 模式

{
   "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"
      }
   }
}

字段:

api_key (str | None)
config (Dict[str, Any] | None)
is_cloud (bool)
limit (int)
user_id (str | None)

field user_id: str | None = None#: 用于内存操作的用户 ID。如果未提供，将生成一个 UUID。

field limit: int = 10#: 内存查询中返回结果的最大数量。

field is_cloud: bool = True#: 是否使用云端 Mem0 客户端 (True) 或本地客户端 (False)。

field api_key: str | None = None#: 云端 Mem0 客户端的 API 密钥。如果 is_cloud=True，则为必需。

field config: Dict[str, Any] | None = None#: 本地 Mem0 客户端的配置字典。如果 is_cloud=False，则为必需。