教程：使用 Azure 托管 Redis 生成代理内存

在本教程中，你将使用 Azure Managed Redis 为 AI 代理实现内存。 代理记忆并不是 Azure Managed Redis 的一项独立托管功能。 它是一种应用程序模式，它结合了 Redis 数据结构、过期策略和矢量搜索。

该模式有两个层：

• 短期或会话记忆：最近的聊天轮次、会话状态、工具输出、当前计划以及会自动过期或被裁剪的检查点。
• 长期语义记忆：长期保存的事实、偏好、摘要和从交互中提取的观察结果，连同嵌入和元数据一起存储，并通过向量相似度和筛选条件进行检索。

当缓存具有所需的 RediSearch/vector 搜索功能时，Azure Managed Redis 支持矢量搜索。 有关详细信息，请参阅Azure托管 Redis 中的矢量嵌入和矢量搜索。 RediSearch 和 RedisJSON 是 Redis 模块，创建Azure托管 Redis 实例时必须启用模块。 有关详细信息，请参阅将 Redis 模块与 Azure 托管 Redis 配合使用。

先决条件

• 一个Azure Managed Redis 实例。 如果需要创建一个，请参阅快速入门：创建Azure托管 Redis 实例。
• 短期内存（列表、哈希、流）无需任何模块即可工作。 长期语义记忆需要 RediSearch，它在哈希或 JSON 文档中存储和搜索向量，如 Azure 托管 Redis 向量搜索 和 Redis 向量搜索概念 中所述。 本教程会同时构建这两层，因此，要从头到尾完成本教程，你需要使用 RediSearch。 如果将会话或内存记录存储为 JSON 文档而不是哈希，则还启用 RedisJSON。 只能在创建实例时启用模块，并且以后无法添加，因此请提前决定需要哪些模块。 请参阅具有 Azure 托管 Redis 的 Redis 模块。
• Python 3.10 或更高版本，以及诸如 redis、redis-entraid 和 azure-identity 等软件包，以及你的嵌入或代理框架的 SDK。

Architecture

以下概念体系结构将易失会话上下文与持久语义内存分开。

User request
    |
    v
Agent runtime
    |
    +--> Short-term memory in Azure Managed Redis
    |       - Lists: bounded chat history
    |       - Hashes or JSON: session state, current plan, checkpoints
    |       - Streams: tool events and intermediate outputs
    |       - TTL and trimming on every write
    |
    +--> Long-term semantic memory in Azure Managed Redis
            - Extracted facts, preferences, summaries
            - Embedding vector + tenant/user/session metadata
            - RediSearch vector index
            - KNN search with metadata filters

Redis 非常适合代理状态，因为代理需要快速内存、搜索和有状态工作流。 Redis AI 代理概念将短期内存、长期内存、规划和工具执行描述为代理体系结构的核心部分。 有关详细信息，请参阅 Redis AI 代理概念。

连接到 Azure 托管 Redis

在生产环境中使用Microsoft Entra ID身份验证和 TLS。 以下示例仅使用占位符。

import redis
from azure.identity import DefaultAzureCredential
from redis_entraid.cred_provider import create_from_default_azure_credential

credential_provider = create_from_default_azure_credential(
    ("https://redis.azure.com/.default",)
)

redis_client = redis.Redis(
    host="<azure-managed-redis-hostname>",
    port=10000,
    ssl=True,
    decode_responses=False,
    credential_provider=credential_provider,
)

添加短期会话内存

应绑定短期内存。 将最近的原始轮次存储在列表中，将状态和检查点存储在哈希表或 JSON 文档中，并将工具输出引用存储在流中。 在每次写入时执行 TTL 和裁剪，使非活动会话自动失效。

存储有限聊天历史记录

import json
import time

SESSION_TTL_SECONDS = 60 * 60
MAX_RECENT_TURNS = 40

def history_key(tenant_id: str, session_id: str) -> str:
    return f"agent:{tenant_id}:session:{session_id}:history"

def add_turn(tenant_id: str, session_id: str, role: str, content: str) -> None:
    turn = {
        "role": role,
        "content": content,
        "created_at": int(time.time()),
    }
    key = history_key(tenant_id, session_id)

    pipe = redis_client.pipeline()
    pipe.rpush(key, json.dumps(turn))
    pipe.ltrim(key, -MAX_RECENT_TURNS, -1)
    pipe.expire(key, SESSION_TTL_SECONDS)
    pipe.execute()

def load_recent_turns(tenant_id: str, session_id: str) -> list[dict]:
    return [
        json.loads(item)
        for item in redis_client.lrange(history_key(tenant_id, session_id), 0, -1)
    ]

存储会话状态、计划和检查点

对频繁更改的压缩状态使用哈希。 如果需要嵌套计划或复杂的检查点结构，请使用 RedisJSON。 RedisJSON 需要 RedisJSON 模块，必须在创建缓存时启用该模块。 有关模块可用性和创建时间要求，请参阅具有 Azure 托管 Redis 的 Redis 模块。

def state_key(tenant_id: str, session_id: str) -> str:
    return f"agent:{tenant_id}:session:{session_id}:state"

def save_checkpoint(
    tenant_id: str,
    session_id: str,
    *,
    summary: str,
    current_plan: list[str],
    checkpoint_name: str,
) -> None:
    key = state_key(tenant_id, session_id)
    redis_client.hset(
        key,
        mapping={
            "summary": summary,
            "current_plan": json.dumps(current_plan),
            "checkpoint": checkpoint_name,
            "updated_at": str(int(time.time())),
        },
    )
    redis_client.expire(key, SESSION_TTL_SECONDS)

存储工具事件和输出

将大型工具输出存储在应用程序存储或绑定的 Redis 密钥中，并将引用或摘要仅放在流中。 使用 MAXLEN，这样流就不会无限增长。

def tool_events_key(tenant_id: str, session_id: str) -> str:
    return f"agent:{tenant_id}:session:{session_id}:tool-events"

def record_tool_event(
    tenant_id: str,
    session_id: str,
    tool_name: str,
    status: str,
    output_ref: str,
) -> None:
    key = tool_events_key(tenant_id, session_id)
    redis_client.xadd(
        key,
        {
            "tool": tool_name,
            "status": status,
            "output_ref": output_ref,
            "created_at": str(int(time.time())),
        },
        maxlen=200,
        approximate=True,
    )
    redis_client.expire(key, SESSION_TTL_SECONDS)

添加长期语义内存

长期内存仅存储所选持久信息。 不要嵌入所有原始轮次。 相反，在完成提取步骤后，应提倡使用简洁的事实、偏好、摘要或观察。 使用元数据存储每个内存，以便检索可以按租户、用户、源、内存类型和保留策略进行筛选。

Redis 矢量搜索对哈希或 JSON 文档使用辅助索引，支持 FLAT 和 HNSW 矢量索引，并支持使用元数据筛选器进行矢量搜索。 更多信息请参见 Redis 向量搜索概念。 在 Azure Managed Redis 中，要使用矢量搜索，需要具备 RediSearch/矢量搜索功能。 有关详细信息，请参阅Azure托管 Redis 矢量搜索。

创建内存索引

为每个索引使用一个嵌入模型、维度和距离指标。 如果更改嵌入模型，请重新生成或创建新索引。

from redis.commands.search.field import NumericField, TagField, TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition, IndexType
from redis.exceptions import ResponseError

INDEX_NAME = "idx:agent_memories"
MEMORY_PREFIX = "agent:memory:"
VECTOR_DIMENSIONS = 1536

def create_memory_index() -> None:
    try:
        redis_client.ft(INDEX_NAME).create_index(
            fields=[
                TagField("tenant_id"),
                TagField("user_id"),
                TagField("memory_type"),
                TextField("text"),
                TextField("source"),
                NumericField("created_at"),
                VectorField(
                    "embedding",
                    "HNSW",
                    {
                        "TYPE": "FLOAT32",
                        "DIM": VECTOR_DIMENSIONS,
                        "DISTANCE_METRIC": "COSINE",
                    },
                ),
            ],
            definition=IndexDefinition(
                prefix=[MEMORY_PREFIX],
                index_type=IndexType.HASH,
            ),
        )
    except ResponseError as ex:
        if "Index already exists" not in str(ex):
            raise

将事实信息转入长期记忆

embed 和 extract_memories 这些函数是特定于应用程序的占位符。 提取步骤应仅返回持久且已获同意的记忆条目。 包括源归属，以便将来的响应可以解释为何使用内存。

import uuid
import numpy as np

def embed(text: str) -> list[float]:
    return [0.0] * VECTOR_DIMENSIONS  # Replace with your embedding model call.

def extract_memories(transcript: list[dict]) -> list[dict]:
    return []  # Replace with your extraction logic (LLM call or rules).

def vector_bytes(values: list[float]) -> bytes:
    return np.array(values, dtype=np.float32).tobytes()

def store_memory(
    tenant_id: str,
    user_id: str,
    memory_type: str,
    text: str,
    source: str,
    source_turn_ids: list[str],
) -> str:
    memory_id = f"{MEMORY_PREFIX}{tenant_id}:{user_id}:{uuid.uuid4()}"
    created_at = int(time.time())

    redis_client.hset(
        memory_id,
        mapping={
            "tenant_id": tenant_id,
            "user_id": user_id,
            "memory_type": memory_type,
            "text": text,
            "source": source,
            "source_turn_ids": json.dumps(source_turn_ids),
            "created_at": created_at,
            "embedding": vector_bytes(embed(text)),
        },
    )
    return memory_id

检索与请求相关的记忆

在运行矢量相似性之前，按租户和用户进行筛选。 仅返回代理提示所需的字段。

from redis.commands.search.query import Query

def escape_tag(value: str) -> str:
    return (
        value.replace("\\", "\\\\")
        .replace("{", "\\{")
        .replace("}", "\\}")
        .replace("|", "\\|")
    )

def retrieve_memories(tenant_id: str, user_id: str, user_message: str) -> list[dict]:
    tenant_filter = escape_tag(tenant_id)
    user_filter = escape_tag(user_id)

    query = (
        Query(
            f"(@tenant_id:{{{tenant_filter}}} @user_id:{{{user_filter}}})"
            "=>[KNN 5 @embedding $vector AS score]"
        )
        .sort_by("score")
        .return_fields("text", "memory_type", "source", "created_at", "score")
        .dialect(2)
    )

    results = redis_client.ft(INDEX_NAME).search(
        query,
        query_params={"vector": vector_bytes(embed(user_message))},
    )

    return [
        {
            "text": doc.text,
            "type": doc.memory_type,
            "source": doc.source,
            "score": doc.score,
        }
        for doc in results.docs
    ]

在代理循环中使用内存

在每次交互中，结合会话摘要、最近几轮对话和相关的长期记忆。 保持注入的记忆精简且可归因。

async def run_agent_turn(tenant_id: str, user_id: str, session_id: str, message: str):
    recent_turns = load_recent_turns(tenant_id, session_id)
    state = redis_client.hgetall(state_key(tenant_id, session_id))
    memories = retrieve_memories(tenant_id, user_id, message)

    prompt_context = {
        "session_summary": state.get(b"summary", b"").decode("utf-8"),
        "recent_turns": recent_turns,
        "relevant_memories": memories,
    }

    response = await call_agent_model(
        user_message=message,
        context=prompt_context,
    )

    add_turn(tenant_id, session_id, "user", message)
    add_turn(tenant_id, session_id, "assistant", response.text)
    return response

设计指南

删除会话和用户内存

在应用程序中内置删除功能。 在会话结束时或用户请求删除时删除短期会话密钥。 通过租户和用户密钥前缀或维护每用户内存 ID 集来删除长期密钥。

def delete_session(tenant_id: str, session_id: str) -> None:
    redis_client.delete(
        history_key(tenant_id, session_id),
        state_key(tenant_id, session_id),
        tool_events_key(tenant_id, session_id),
    )

def delete_user_memory_ids(memory_ids: list[str]) -> None:
    if memory_ids:
        redis_client.delete(*memory_ids)

相关内容

• 什么是Azure托管 Redis 中的矢量嵌入和矢量搜索？
• 使用 Redis 模块配合 Azure Managed Redis。
• 快速入门：创建一个Azure托管的Redis 实例
• 快速入门：使用 Azure 托管 Redis 创建Python应用