Semantic Kernel vector store tutorial: Redis memory store

This semantic kernel redis memory store tutorial walks you through building a vector-backed memory layer using Redis Stack’s native vector search. You’ll wire up Semantic Kernel’s abstract memory interface to a real Redis instance, ingest documents with embeddings, and run similarity queries with metadata filtering. The code targets Semantic Kernel Python 1.28+ and Redis Stack 7.2+.

Prerequisites

You need a working Redis Stack instance with the RediSearch and RedisJSON modules. The easiest path is Docker:

docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

Verify the modules are loaded:

docker exec redis-stack redis-cli MODULE LIST

You should see search and json in the output.

Python dependencies:

pip install semantic-kernel[redis] openai redis

The semantic-kernel[redis] extra pulls in redis-py and the SK Redis memory store connector. You also need an OpenAI API key for embeddings — set it as OPENAI_API_KEY in your environment.

Create the kernel and configure the memory store

Start by instantiating the kernel and registering the Redis memory store. The store requires a Redis client, an index name, and an embedding generator.

import asyncio
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding
from semantic_kernel.memory import RedisMemoryStore

async def main():
    kernel = Kernel()

    # Embedding generator — text-embedding-3-small is 1536 dims, cheap, fast
    embedding_service = OpenAITextEmbedding(
        ai_model_id="text-embedding-3-small",
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    kernel.add_service(embedding_service)

    # Redis client pointing at the local stack instance
    import redis.asyncio as redis
    redis_client = redis.Redis(
        host="localhost",
        port=6379,
        decode_responses=True,
    )

    # The memory store wraps the client and handles index creation
    memory_store = RedisMemoryStore(
        redis_client=redis_client,
        index_name="sk-tutorial-index",
        vector_dimensions=1536,  # must match the embedding model
        distance_function="cosine",
    )

    # Register so other SK components can resolve IMemoryStore
    kernel.add_service(memory_store)

    # Verify connection and create index if missing
    await memory_store.create_collection("sk-tutorial-index")
    print("Index ready")

    return kernel, memory_store, redis_client

if __name__ == "__main__":
    asyncio.run(main())

Expected output:

Index ready

The create_collection call is idempotent — it creates the RediSearch index with the correct schema (vector field, metadata fields, JSON payload) on first run and no-ops thereafter.

Ingest documents with embeddings

Semantic Kernel’s MemoryRecord holds the text, its embedding, and arbitrary metadata. The memory store batches embeddings and writes them to Redis as JSON documents with a vector field.

from semantic_kernel.memory import MemoryRecord

async def ingest_documents(memory_store: RedisMemoryStore, embedding_service: OpenAITextEmbedding):
    docs = [
        {
            "id": "doc-1",
            "text": "Redis Stack adds native vector search via the RediSearch module, supporting HNSW and FLAT indexes.",
            "metadata": {"source": "redis-docs", "topic": "vector-search", "version": "7.2"},
        },
        {
            "id": "doc-2",
            "text": "Semantic Kernel's memory abstraction lets you swap vector stores without changing application code.",
            "metadata": {"source": "sk-docs", "topic": "architecture", "version": "1.28"},
        },
        {
            "id": "doc-3",
            "text": "HNSW indexes provide approximate nearest neighbor search with sub-millisecond latency at scale.",
            "metadata": {"source": "redis-docs", "topic": "performance", "version": "7.2"},
        },
        {
            "id": "doc-4",
            "text": "The Redis memory store persists embeddings as JSON documents with a dedicated vector field for RediSearch.",
            "metadata": {"source": "sk-docs", "topic": "implementation", "version": "1.28"},
        },
    ]

    records = []
    for doc in docs:
        # Generate embedding for the text
        embedding = await embedding_service.generate_embeddings([doc["text"]])
        vector = embedding[0]

        record = MemoryRecord(
            id=doc["id"],
            text=doc["text"],
            embedding=vector,
            description=doc["metadata"]["topic"],
            additional_metadata=doc["metadata"],
        )
        records.append(record)

    # Batch upsert — more efficient than individual calls
    await memory_store.upsert_batch("sk-tutorial-index", records)
    print(f"Ingested {len(records)} documents")

    # Verify count
    count = await memory_store.get_collection_size("sk-tutorial-index")
    print(f"Collection size: {count}")

Run it:

kernel, memory_store, redis_client = await main()
embedding_service = kernel.get_service(type=OpenAITextEmbedding)
await ingest_documents(memory_store, embedding_service)

Expected output:

Ingested 4 documents
Collection size: 4

Each document is now stored as a Redis JSON document at key sk-tutorial-index:{id} with a vector field that RediSearch can query.

Similarity search

The memory store exposes get_nearest_matches for vector similarity search. It accepts a query vector, a limit, and an optional minimum relevance score (cosine similarity threshold).

async def search_similar(memory_store: RedisMemoryStore, embedding_service: OpenAITextEmbedding, query: str, limit: int = 3, min_score: float = 0.7):
    query_embedding = await embedding_service.generate_embeddings([query])
    query_vector = query_embedding[0]

    results = await memory_store.get_nearest_matches(
        collection_name="sk-tutorial-index",
        embedding=query_vector,
        limit=limit,
        min_relevance_score=min_score,
    )

    print(f"\nQuery: '{query}'")
    print(f"Results (min_score={min_score}, limit={limit}):")
    for i, (record, score) in enumerate(results, 1):
        print(f"  {i}. [{score:.4f}] {record.id} — {record.text[:80]}...")
        print(f"      Metadata: {record.additional_metadata}")

    return results

Test it:

await search_similar(
    memory_store,
    embedding_service,
    "How does Semantic Kernel integrate with vector databases?",
    limit=3,
    min_score=0.65,
)

Expected output (scores will vary slightly):

Query: 'How does Semantic Kernel integrate with vector databases?'
Results (min_score=0.65, limit=3):
  1. [0.8234] doc-2 — Semantic Kernel's memory abstraction lets you swap vector stores without changing application code.
      Metadata: {'source': 'sk-docs', 'topic': 'architecture', 'version': '1.28'}
  2. [0.7121] doc-4 — The Redis memory store persists embeddings as JSON documents with a dedicated vector field for RediSearch.
      Metadata: {'source': 'sk-docs', 'topic': 'implementation', 'version': '1.28'}
  3. [0.6892] doc-1 — Redis Stack adds native vector search via the RediSearch module, supporting HNSW and FLAT indexes.
      Metadata: {'source': 'redis-docs', 'topic': 'vector-search', 'version': '7.2'}

The scores are cosine similarities — higher is closer. The min_relevance_score filters out weak matches before they reach your application logic.

Metadata filtering with RediSearch queries

The get_nearest_matches method doesn’t yet expose a filter parameter in the public API. For production workloads requiring metadata filters (e.g., source == "redis-docs"), you have two options:

1. Post-filter in Python — fetch more results than you need, then filter locally. Simple but wastes bandwidth.
2. Use the Redis client directly — construct a RediSearch FT.SEARCH query with a filter clause. This pushes filtering to the engine.

Here’s the direct approach using the async Redis client:

async def filtered_search(redis_client: redis.Redis, embedding_service: OpenAITextEmbedding, query: str, filter_expr: str, limit: int = 5):
    """
    filter_expr examples:
      - '@source:{redis-docs}'
      - '@topic:{vector-search|performance}'
      - '@version:{7.2} @source:{redis-docs}'
    """
    query_embedding = await embedding_service.generate_embeddings([query])
    query_vector = query_embedding[0]

    # Convert to bytes for RediSearch VECTOR parameter
    import numpy as np
    vector_bytes = np.array(query_vector, dtype=np.float32).tobytes()

    # Build the RediSearch query
    # KNN vector search with FILTER clause
    search_query = f"*=>[KNN {limit} @vector $vec AS score]"

    results = await redis_client.execute_command(
        "FT.SEARCH",
        "sk-tutorial-index",
        search_query,
        "PARAMS", "2", "vec", vector_bytes,
        "DIALECT", "2",
        "LIMIT", "0", str(limit),
        "FILTER", filter_expr,
        "RETURN", "3", "id", "text", "metadata", "score",
        "SORTBY", "score",
    )

    # Parse results: [total, [key, [field1, val1, ...]], ...]
    total = results[0]
    print(f"\nFiltered query: '{query}' | Filter: {filter_expr}")
    print(f"Total matches: {total}")

    for i in range(1, len(results), 2):
        key = results[i]
        fields = results[i + 1]
        field_dict = dict(zip(fields[::2], fields[1::2]))
        print(f"  {key}: score={field_dict.get('score')}")
        print(f"    Text: {field_dict.get('text')[:80]}...")
        print(f"    Metadata: {field_dict.get('metadata')}")

# Only redis-docs, any topic
await filtered_search(
    redis_client,
    embedding_service,
    "vector index performance",
    filter_expr="@source:{redis-docs}",
    limit=3,
)

# Only architecture topic
await filtered_search(
    redis_client,
    embedding_service,
    "memory abstraction design",
    filter_expr="@topic:{architecture}",
    limit=3,
)

Expected output:

Filtered query: 'vector index performance' | Filter: @source:{redis-docs}
Total matches: 2
  sk-tutorial-index:doc-1: score=0.8123
    Text: Redis Stack adds native vector search via the RediSearch module, supporting HNSW and FLAT indexes.
    Metadata: {"source": "redis-docs", "topic": "vector-search", "version": "7.2"}
  sk-tutorial-index:doc-3: score=0.7941
    Text: HNSW indexes provide approximate nearest neighbor search with sub-millisecond latency at scale.
    Metadata: {"source": "redis-docs", "topic": "performance", "version": "7.2"}

Filtered query: 'memory abstraction design' | Filter: @topic:{architecture}
Total matches: 1
  sk-tutorial-index:doc-2: score=0.8234
    Text: Semantic Kernel's memory abstraction lets you swap vector stores without changing application code.
    Metadata: {"source": "sk-docs", "topic": "architecture", "version": "1.28"}

The FILTER clause uses RediSearch’s tag query syntax. Tag fields are indexed for exact-match filtering with minimal overhead. The metadata field in the index is a JSON blob — we also index source, topic, and version as separate tag fields during collection creation (handled automatically by RedisMemoryStore).

Delete and update records

Memory stores need CRUD, not just create and read. The store exposes remove and upsert for single-record operations.

async def update_document(memory_store: RedisMemoryStore, embedding_service: OpenAITextEmbedding, doc_id: str, new_text: str, new_metadata: dict):
    embedding = await embedding_service.generate_embeddings([new_text])
    vector = embedding[0]

    record = MemoryRecord(
        id=doc_id,
        text=new_text,
        embedding=vector,
        description=new_metadata.get("topic", "updated"),
        additional_metadata=new_metadata,
    )

    await memory_store.upsert("sk-tutorial-index", record)
    print(f"Updated {doc_id}")

async def delete_document(memory_store: RedisMemoryStore, doc_id: str):
    await memory_store.remove("sk-tutorial-index", doc_id)
    print(f"Deleted {doc_id}")

# Update doc-3 with new content
await update_document(
    memory_store,
    embedding_service,
    "doc-3",
    "HNSW indexes in Redis 7.4 add on-disk storage tier for cost-efficient large-scale vector search.",
    {"source": "redis-docs", "topic": "performance", "version": "7.4"},
)

# Verify the update
await search_similar(memory_store, embedding_service, "on-disk vector storage tier", limit=2)

# Delete doc-4
await delete_document(memory_store, "doc-4")

# Verify deletion
count = await memory_store.get_collection_size("sk-tutorial-index")
print(f"\nCollection size after delete: {count}")

Expected output:

Updated doc-3

Query: 'on-disk vector storage tier'
Results (min_score=0.65, limit=2):
  1. [0.8412] doc-3 — HNSW indexes in Redis 7.4 add on-disk storage tier for cost-efficient large-scale vector search.
      Metadata: {'source': 'redis-docs', 'topic': 'performance', 'version': '7.4'}

Deleted doc-4

Collection size after delete: 3

Clean up resources

Close the Redis client and any kernel resources when your application shuts down.

async def cleanup(redis_client: redis.Redis):
    await redis_client.aclose()
    print("Redis connection closed")

await cleanup(redis_client)

Production considerations

Index configuration

The default index uses HNSW with M=16, EF_CONSTRUCTION=200, and EF_RUNTIME=10. For higher recall at the cost of build time and memory, increase EF_CONSTRUCTION. For lower latency, decrease EF_RUNTIME. You can pass these via RedisMemoryStore constructor arguments:

memory_store = RedisMemoryStore(
    redis_client=redis_client,
    index_name="prod-index",
    vector_dimensions=1536,
    distance_function="cosine",
    hnsw_m=32,
    hnsw_ef_construction=500,
    hnsw_ef_runtime=50,
)

Connection pooling

The redis.asyncio.Redis client uses a connection pool by default (max 10 connections). Tune max_connections for your concurrency profile:

redis_client = redis.Redis(
    host="localhost",
    port=6379,
    decode_responses=True,
    max_connections=50,
)

Namespace isolation

Use distinct index names per environment or tenant: sk-prod-tenantA, sk-staging-tenantB. The memory store has no built-in multi-tenancy — the index name is your namespace.

Monitoring

RediSearch exposes FT.INFO index-name for index stats (document count, memory usage, index size). Hook this into your observability stack:

docker exec redis-stack redis-cli FT.INFO sk-tutorial-index

Embedding model consistency

The vector_dimensions parameter must match your embedding model’s output dimension. Changing models requires a new index — RediSearch cannot alter vector dimensions in place. Plan for re-indexing during model migrations.

Full runnable script

Save this as sk_redis_memory_tutorial.py and run with python sk_redis_memory_tutorial.py:

import asyncio
import os
import numpy as np
import redis.asyncio as redis
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding
from semantic_kernel.memory import RedisMemoryStore, MemoryRecord

async def main():
    # --- Setup ---
    kernel = Kernel()
    embedding_service = OpenAITextEmbedding(
        ai_model_id="text-embedding-3-small",
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    kernel.add_service(embedding_service)

    redis_client = redis.Redis(
        host="localhost",
        port=6379,
        decode_responses=True,
    )

    memory_store = RedisMemoryStore(
        redis_client=redis_client,
        index_name="sk-tutorial-index",
        vector_dimensions=1536,
        distance_function="cosine",
    )
    kernel.add_service(memory_store)

    await memory_store.create_collection("sk-tutorial-index")
    print("Index ready")

    # --- Ingest ---
    docs = [
        {"id": "doc-1", "text": "Redis Stack adds native vector search via the RediSearch module, supporting HNSW and FLAT indexes.", "metadata": {"source": "redis-docs", "topic": "vector-search", "version": "7.2"}},
        {"id": "doc-2", "text": "Semantic Kernel's memory abstraction lets you swap vector stores without changing application code.", "metadata": {"source": "sk-docs", "topic": "architecture", "version": "1.28"}},
        {"id": "doc-3", "text": "HNSW indexes provide approximate nearest neighbor search with sub-millisecond latency at scale.", "metadata": {"source": "redis-docs", "topic": "performance", "version": "7.2"}},
        {"id": "doc-4", "text": "The Redis memory store persists embeddings as JSON documents with a dedicated vector field for RediSearch.", "metadata": {"source": "sk-docs", "topic": "implementation", "version": "1.28"}},
    ]

    records = []
    for doc in docs:
        embedding = await embedding_service.generate_embeddings([doc["text"]])
        records.append(MemoryRecord(
            id=doc["id"],
            text=doc["text"],
            embedding=embedding[0],
            description=doc["metadata"]["topic"],
            additional_metadata=doc["metadata"],
        ))

    await memory_store.upsert_batch("sk-tutorial-index", records)
    print(f"Ingested {len(records)} documents")

    # --- Search ---
    async def search(query: str, limit: int = 3, min_score: float = 0.65):
        q_emb = await embedding_service.generate_embeddings([query])
        results = await memory_store.get_nearest_matches("sk-tutorial-index", q_emb[0], limit, min_score)
        print(f"\nQuery: '{query}'")
        for i, (rec, score) in enumerate(results, 1):
            print(f"  {i}. [{score:.4f}] {rec.id} — {rec.text[:80]}...")

    await search("How does Semantic Kernel integrate with vector databases?")
    await search("vector index performance characteristics")

    # --- Filtered search via raw RediSearch ---
    async def filtered_search(query: str, filter_expr: str, limit: int = 5):
        q_emb = await embedding_service.generate_embeddings([query])
        vec_bytes = np.array(q_emb[0], dtype=np.float32).tobytes()
        res = await redis_client.execute_command(
            "FT.SEARCH", "sk-tutorial-index",
            f"*=>[KNN {limit} @vector $vec AS score]",
            "PARAMS", "2", "vec", vec_bytes,
            "DIALECT", "2", "LIMIT", "0", str(limit),
            "FILTER", filter_expr,
            "RETURN", "3", "id", "text", "metadata", "score",
            "SORTBY", "score",
        )
        total = res[0]
        print(f"\nFiltered: '{query}' | {filter_expr} | Total: {total}")
        for i in range(1, len(res), 2):
            fields = dict(zip(res[i+1][::2], res[i+1][1::2]))
            print(f"  {res[i]}: score={fields.get('score')} | {fields.get('text')[:60]}...")

    await filtered_search("vector index performance", "@source:{redis-docs}")
    await filtered_search("memory abstraction", "@topic:{architecture}")

    # --- Update & Delete ---
    new_emb = await embedding_service.generate_embeddings(["HNSW in Redis 7.4 adds on-disk tier for cost-efficient large-scale vector search."])
    await memory_store.upsert("sk-tutorial-index", MemoryRecord(
        id="doc-3",
        text="HNSW in Redis 7.4 adds on-disk tier for cost-efficient large-scale vector search.",
        embedding=new_emb[0],
        description="performance",
        additional_metadata={"source": "redis-docs", "topic": "performance", "version": "7.4"},
    ))
    print("\nUpdated doc-3")
    await search("on-disk vector storage")

    await memory_store.remove("sk-tutorial-index", "doc-4")
    print("Deleted doc-4")
    print(f"Final collection size: {await memory_store.get_collection_size('sk-tutorial-index')}")

    # --- Cleanup ---
    await redis_client.aclose()
    print("Done")

if __name__ == "__main__":
    asyncio.run(main())

Next steps

You now have a working Semantic Kernel vector store backed by Redis Stack. From here:

• Add a planner or agent that uses the memory store as a retrieval tool via kernel.add_function and a custom skill.
• Implement hybrid search by combining the vector query with a full-text FT.SEARCH on the text field and merging results with reciprocal rank fusion.
• Enable semantic caching — store prompt/response pairs with embeddings and serve repeated queries from Redis at sub-millisecond latency. This is where a gateway like n4n.ai can help by metering token usage across cached and live calls.
• Scale horizontally — Redis Cluster with RediSearch (Redis Enterprise) shards the index across nodes. The SK memory store points at the cluster endpoint unchanged.

The abstraction holds: your application code stays the same whether the index lives on a single Docker container or a 50-node cluster.