License
- OSI Approved :: MIT License
Development Status
- 4 - Beta
Intended Audience
- Developers
Programming Language
- Python :: 3
- Python :: 3.10
- Python :: 3.11
- Python :: 3.12
- Python :: 3.13
- Python :: 3.14
Typing
- Typed
Microsoft Agent Framework Redis
Redis-backed conversation history, memory retrieval, and vector storage for Microsoft Agent Framework.
Choose the right component
| Your goal | Component |
|---|---|
| Persist ordered conversation messages across runs and application restarts | RedisHistoryProvider |
| Retrieve relevant memories and add them to an agent's context | RedisContextProvider |
| Store and search your own documents or embeddings | RedisCollection (experimental) |
| Manage vector collections with a shared connection and namespace | RedisStore (experimental) |
Resolve vector-store connection configuration from arguments, .env, or environment variables |
RedisSettings |
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.
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",
)
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.