AutoGen智能体开发：autogen_ext.memory.redis包

autogen_ext.memory.redis

王国平

325人浏览 · 2025-12-08 06:46:53

王国平 · 2025-12-08 06:46:53 发布

pydantic 模型 RedisMemoryConfig[源]

基类: BaseModel

基于 Redis 的向量内存配置。

此类定义了使用 Redis 作为向量内存存储（支持语义内存）的配置选项。它允许自定义 Redis 连接、索引设置、相似性搜索参数和嵌入模型。

显示 JSON 模式

{
   "title": "RedisMemoryConfig",
   "description": "Configuration for Redis-based vector memory.\n\nThis class defines the configuration options for using Redis as a vector memory store,\nsupporting semantic memory. It allows customization of the Redis connection, index settings,\nsimilarity search parameters, and embedding model.",
   "type": "object",
   "properties": {
      "redis_url": {
         "default": "redis://:6379",
         "description": "url of the Redis instance",
         "title": "Redis Url",
         "type": "string"
      },
      "index_name": {
         "default": "chat_history",
         "description": "Name of the Redis collection",
         "title": "Index Name",
         "type": "string"
      },
      "prefix": {
         "default": "memory",
         "description": "prefix of the Redis collection",
         "title": "Prefix",
         "type": "string"
      },
      "sequential": {
         "default": false,
         "description": "ignore semantic similarity and simply return memories in sequential order",
         "title": "Sequential",
         "type": "boolean"
      },
      "distance_metric": {
         "default": "cosine",
         "enum": [
            "cosine",
            "ip",
            "l2"
         ],
         "title": "Distance Metric",
         "type": "string"
      },
      "algorithm": {
         "default": "flat",
         "enum": [
            "flat",
            "hnsw"
         ],
         "title": "Algorithm",
         "type": "string"
      },
      "top_k": {
         "default": 10,
         "description": "Number of results to return in queries",
         "title": "Top K",
         "type": "integer"
      },
      "datatype": {
         "default": "float32",
         "enum": [
            "uint8",
            "int8",
            "float16",
            "float32",
            "float64",
            "bfloat16"
         ],
         "title": "Datatype",
         "type": "string"
      },
      "distance_threshold": {
         "default": 0.7,
         "description": "Minimum similarity score threshold",
         "title": "Distance Threshold",
         "type": "number"
      },
      "model_name": {
         "default": "sentence-transformers/all-mpnet-base-v2",
         "description": "Embedding model name",
         "title": "Model Name",
         "type": "string"
      }
   }
}

字段:

算法 (Literal['flat', 'hnsw'])
数据类型 (Literal['uint8', 'int8', 'float16', 'float32', 'float64', 'bfloat16'])
距离度量 (Literal['cosine', 'ip', 'l2'])
距离阈值 (float)
索引名称 (str)
模型名称 (str)
前缀 (str)
redis_url (str)
顺序 (bool)
top_k (int)

字段 redis_url: str = 'redis://:6379'

Redis 实例的 URL

字段 index_name: str = 'chat_history'

Redis 集合的名称

字段 prefix: str = 'memory'

Redis 集合的前缀

字段 sequential: bool = False

忽略语义相似性，仅按顺序返回内存

字段 distance_metric: Literal['cosine', 'ip', 'l2'] = 'cosine'

字段 algorithm: Literal['flat', 'hnsw'] = 'flat'

字段 top_k: int = 10

查询中返回的结果数量

字段 datatype: Literal['uint8', 'int8', 'float16', 'float32', 'float64', 'bfloat16'] = 'float32'

字段 distance_threshold: float = 0.7

最小相似度分数阈值

字段 model_name: str = 'sentence-transformers/all-mpnet-base-v2'

嵌入模型名称

class RedisMemory(config: RedisMemoryConfig | None = None)[源]

基类：Memory, Component[RedisMemoryConfig]

使用 RedisVL 支持的向量相似性搜索存储和检索内存。

RedisMemory 提供了一种基于向量的内存实现，它使用 RedisVL 根据语义相似性或顺序存储和检索内容。它通过利用向量嵌入来查找相似内容，从而增强代理在对话中回忆相关信息的能力。

此实现需要安装 RedisVL 扩展。安装命令如下：

pip install "autogen-ext[redisvl]"

此外，您还需要访问 Redis 实例。要在 Docker 中运行本地 Redis 实例，请执行以下命令：

docker run -d --name redis -p 6379:6379 redis:8

要下载并本地运行 Redis：

curl -fsSL https://packages.redis.io/gpg | sudo gpg --dearmor -o /usr/share/keyrings/redis-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/redis-archive-keyring.gpg] https://packages.redis.io/deb $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/redis.list
sudo apt-get update  > /dev/null 2>&1
sudo apt-get install redis-server  > /dev/null 2>&1
redis-server --daemonize yes

参数:

config (RedisMemoryConfig | None) – Redis 内存的配置。如果为 None，则默认为带有推荐设置的 RedisMemoryConfig。

示例

from logging import WARNING, getLogger

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core.memory import MemoryContent, MemoryMimeType
from autogen_ext.memory.redis import RedisMemory, RedisMemoryConfig
from autogen_ext.models.openai import OpenAIChatCompletionClient

logger = getLogger()
logger.setLevel(WARNING)


# Define tool to use
async def get_weather(city: str, units: str = "imperial") -> str:
    if units == "imperial":
        return f"The weather in {city} is 73 °F and Sunny."
    elif units == "metric":
        return f"The weather in {city} is 23 °C and Sunny."
    else:
        return f"Sorry, I don't know the weather in {city}."


async def main():
    # Initailize Redis memory
    redis_memory = RedisMemory(
        config=RedisMemoryConfig(
            redis_url="redis://:6379",
            index_name="chat_history",
            prefix="memory",
        )
    )

    # Add user preferences to memory
    await redis_memory.add(
        MemoryContent(
            content="The weather should be in metric units",
            mime_type=MemoryMimeType.TEXT,
            metadata={"category": "preferences", "type": "units"},
        )
    )

    await redis_memory.add(
        MemoryContent(
            content="Meal recipe must be vegan",
            mime_type=MemoryMimeType.TEXT,
            metadata={"category": "preferences", "type": "dietary"},
        )
    )

    model_client = OpenAIChatCompletionClient(
        model="gpt-4o",
    )

    # Create assistant agent with ChromaDB memory
    assistant_agent = AssistantAgent(
        name="assistant_agent",
        model_client=model_client,
        tools=[get_weather],
        memory=[redis_memory],
    )

    stream = assistant_agent.run_stream(task="What is the weather in New York?")
    await Console(stream)

    await model_client.close()
    await redis_memory.close()


asyncio.run(main())

输出

---------- TextMessage (user) ----------
What is the weather in New York?
---------- MemoryQueryEvent (assistant_agent) ----------
[MemoryContent(content='The weather should be in metric units', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata={'category': 'preferences', 'type': 'units'})]
---------- ToolCallRequestEvent (assistant_agent) ----------
[FunctionCall(id='call_tyCPvPPAV4SHWhtfpM6UMemr', arguments='{"city":"New York","units":"metric"}', name='get_weather')]
---------- ToolCallExecutionEvent (assistant_agent) ----------
[FunctionExecutionResult(content='The weather in New York is 23 °C and Sunny.', name='get_weather', call_id='call_tyCPvPPAV4SHWhtfpM6UMemr', is_error=False)]
---------- ToolCallSummaryMessage (assistant_agent) ----------
The weather in New York is 23 °C and Sunny.

component_config_schema

的别名 RedisMemoryConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.memory.redis_memory.RedisMemory'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async update_context(model_context: ChatCompletionContext) → UpdateContextResult [源]

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

此方法检索与上下文中最后一条消息相关的内存内容，并将其作为系统消息添加。此实现使用上下文中最后一条消息作为查询，查找语义相似的内存，并将它们全部作为单个系统消息添加到上下文中。

参数:

model_context (ChatCompletionContext) – 要用相关内存更新的模型上下文。

UpdateContextResult – 包含用于更新上下文的内存的对象。

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

将内存内容对象添加到 Redis。

注意

如果 RedisMemoryConfig 未设置为“顺序”，为了对存储的内存执行语义搜索，RedisMemory 会从 MemoryContent 对象的 content 字段创建向量嵌入。此内容假定为文本、JSON 或 Markdown，并传递给 RedisMemoryConfig 中指定的向量嵌入模型。

参数:

content (MemoryContent) – 要存储在 Redis 中的内存内容。
cancellation_token (CancellationToken) – 传递的令牌以停止操作。未使用。

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

根据语义向量相似性查询内存内容。

注意

RedisMemory.query() 支持额外的关键字参数以提高查询性能。top_k (int): 要包含的相关内存的最大数量。默认为 10。distance_threshold (float): 在执行余弦相似度搜索时，考虑内存语义相似的最大向量空间距离。默认为 0.7。sequential (bool): 忽略语义相似性并返回 top_k 条最新内存。

参数:

query (str | MemoryContent) – 用于执行向量相似性搜索的查询。如果传递字符串，则使用 RedisMemoryConfig 中指定的模型从该字符串创建向量嵌入。如果传递 MemoryContent 对象，则提取此对象的 content 字段，并使用 RedisMemoryConfig 中指定的模型从该字段创建向量嵌入。
cancellation_token (CancellationToken) – 传递的令牌以停止操作。未使用。

memoryQueryResult – 包含与所提供查询相关的内存的对象。

async clear() → None [源]

清除内存中的所有条目，同时保留 RedisMemory 资源。

async close() → None [源]

清除内存中的所有条目，并清理 Redis 客户端、索引和资源。

《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能。《AI提示工程必知必会》主要内容包括各类提示词的应用，如问答式、指令式、状态类、建议式、安全类和感谢类提示词，以及如何通过实战演练掌握提示词的使用技巧；使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务，以及在数据挖掘、程序开发等领域的应用；AI在绘画创作上的应用，百度文心一言和阿里通义大模型这两大智能平台的特性与功能，以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》，读者可掌握如何有效利用AI提示工程提升工作效率，创新工作流程，并在职场中脱颖而出。