Microsoft Agent Framework Redis

Redis-backed conversation history, memory retrieval, and vector storage for Microsoft Agent Framework.

Choose the right component

These components can be used together, but do not automatically share stored records. Import them from agent_framework.redis or agent_framework_redis.

Install

Requires Python 3.10 or later:

pip install agent-framework-redis --pre

Run Redis

Choose a managed service or run Redis locally:

• Azure Managed Redis: Run Redis as a managed Azure service. Follow the Azure Managed Redis quickstart to create an instance.

• Redis Cloud: Use Redis Cloud for a fully managed Redis database.

• Local Docker: Start Redis on your machine with:

  docker run --rm --name agent-framework-redis -p 127.0.0.1:6379:6379 redis:8.0.3
  

For managed services, choose a configuration with the Search/JSON capabilities required by your component below, and use the service's connection endpoint and authentication settings.

Requirements and configuration

Conversation history uses Redis Lists and does not require Search or RedisJSON. Context retrieval requires Redis Search. Vector collections require Redis 8.0.3 or later with Search, including INDEXMISSING and INDEXEMPTY support; JSON collections additionally require RedisJSON. Redis Search indexes require logical database 0, so vector-store URLs and clients must select database 0 (the default).

Choose storage_type="hash" for non-null records with binary vector storage, or "json" for nullable fields/vectors and native JSON data. HASH rejects None, including vector=None; JSON preserves explicit null. Both support multiple vector fields and FLAT/HNSW indexes.

RedisCollection and RedisStore use RedisSettings with the Agent Framework settings loader. Connection precedence is an explicit redis_url, then REDIS_URL in a supplied env_file_path, then the process environment, with redis://localhost:6379 as the default. Settings mask credential-bearing URLs with SecretString; use rediss:// when your server requires TLS.

RedisHistoryProvider can create its own client from a URL or credential provider, or borrow a standalone redis.asyncio.Redis client configured with decode_responses=True. A borrowed client lets the application configure connection policy such as bounded timeouts and health checks. It remains caller-owned and must not be combined with redis_url or credential_provider. Retry behavior also remains caller-controlled; transcript appends are not idempotent if a client replays a write after an ambiguous connection loss.

For vector storage, you may supply a standalone redis.asyncio.Redis client using decode_responses=False and RESP2. Both URL-created and supplied clients must use strict UTF-8 encoding (encoding="utf-8", encoding_errors="strict", the defaults) so Unicode keys and string fields round-trip without lossy conversions. Incompatible encoding or database settings are rejected before connecting. Supplied clients take precedence over URL settings and remain caller-owned. Closing a store closes its owned connection, not its stored data. Redis Cluster clients are not supported.

Vector collections support dense search and a subset of portable filters, not hybrid/full-text search or literal substring/prefix/suffix filters. JSON null checks are supported on numeric and boolean fields, not strings or arrays. Indexed strings cannot contain surrounding whitespace, NUL, or U+001F, and must fit Redis's 4096-byte TAG limit; these restrictions do not apply to unindexed payloads. Unsupported operations raise an error.

Isolate conversation history

RedisHistoryProvider uses scoped keys by default. Supply a stable application_id; also supply tenant_id and agent_id whenever those boundaries exist in your application. The provider's source_id and each non-empty session ID are included automatically:

from agent_framework.redis import RedisHistoryProvider

history_provider = RedisHistoryProvider(
    redis_url="redis://localhost:6379",
    application_id="support-app",
    tenant_id="contoso",
    agent_id="triage-agent",
)

To bound transcript I/O with a caller-configured client:

from redis.asyncio import Redis

redis_client = Redis.from_url(
    "redis://localhost:6379",
    decode_responses=True,
    socket_connect_timeout=3,
    socket_timeout=5,
)
history_provider = RedisHistoryProvider(
    redis_client=redis_client,
    application_id="support-app",
)

The application remains responsible for closing redis_client.

Scoped mode rejects missing application or session identifiers rather than placing unrelated conversations under a shared fallback key. Identifiers are encoded independently, so they do not need to be globally unique across tenants, applications, agents, and provider sources.

Releases that predate scoped keys used {key_prefix}:{session_id or "default"}. Existing deployments can temporarily retain that exact format by opting in explicitly:

legacy_history_provider = RedisHistoryProvider(
    redis_url="redis://localhost:6379",
    key_format="legacy",
)

Legacy mode does not accept scoped identifiers. Scoped mode never reads, rewrites, or deletes legacy keys. To migrate existing history, copy only the records belonging to a verified application, tenant, agent, provider source, and session into the corresponding scoped key using an application-owned migration process. After verifying the copied history, remove legacy keys separately according to the application's retention policy.

Store and search documents

This example uses precomputed vectors, so no embedding service is needed. It connects using REDIS_URL or the localhost default:

import asyncio
from dataclasses import dataclass
from typing import Annotated

from agent_framework import VectorStoreField, vectorstoremodel
from agent_framework.redis import RedisStore


@vectorstoremodel
@dataclass
class Document:
    id: Annotated[str, VectorStoreField("key")]
    text: Annotated[str, VectorStoreField("data")]
    vector: Annotated[
        list[float] | None,
        VectorStoreField("vector", dimensions=2, index_kind="flat"),
    ] = None


async def main() -> None:
    async with RedisStore(storage_type="json", namespace="my-app") as store:
        collection = store.get_collection(Document, collection_name="documents")
        await collection.ensure_collection_exists()
        await collection.upsert(
            [Document("one", "A guide to Redis", [1.0, 0.0])],
            generate_vectors=False,
        )
        results = await collection.search(vector=[1.0, 0.0], top=3)
        async for result in results:
            print(result["record"].text, result["score"])


asyncio.run(main())

CRUD operations accept batches. Provide an embedding generator to generate vectors automatically, or use generate_vectors=False to keep supplied vectors. Retrieval excludes vectors by default; use include_vectors=True to return them. Search scores are native distances (cosine distance by default), where lower is better; a nonnegative score_threshold sets a maximum distance before paging.

Documentation and examples

• Agent Framework documentation
• Redis documentation
• Redis vector search
• Redis conversation-history example
• Redis context-provider examples
• HASH and JSON vector-store example