Enhance AI Solutions with Azure Managed Redis

1. Azure Managed Redis Tiers

Also: Azure Cache for Redis with Basic/Standard/Premium tiers — Basic is single-node dev/test; Standard adds replication; Premium adds geo-replication and VNet.

2. Connection: Port 10000 (Not 6379)

Azure Managed Redis uses port 10000 with mandatory TLS — not the default Redis port 6379. This is a common exam trap.

redis://myinstance.redis.azure.com:10000 (TLS required)

3. Caching Strategies — Which to Use When

1. Cache-aside (Lazy loading) — app checks cache first, on MISS fetches from DB and stores in cache. Most common for AI inference results. Requires cache invalidation on source data change.
2. Write-through — writes go to cache AND DB simultaneously. Cache always consistent. Higher write latency. Good for user profiles that must always be current.
3. Write-behind (Write-back) — writes go to cache, DB updated asynchronously. Low write latency. Risk of data loss on crash. Advanced scenario.
4. Read-through — cache fetches from DB on MISS automatically. Cache acts as transparent proxy. Simplifies application code.

1. Python: redis-py with Managed Identity

import redis

# Access key authentication
r = redis.Redis(
    host='your-instance.redis.azure.com',
    port=10000,
    ssl=True,
    decode_responses=True,
    password='your-access-key'
)

# Entra ID (Managed Identity) authentication
from redis_entraid.cred_provider import create_from_default_azure_credential
credential_provider = create_from_default_azure_credential(
    ("https://redis.azure.com/.default",)
)
r = redis.Redis(host='...', port=10000, ssl=True,
    decode_responses=True, credential_provider=credential_provider)

2. decode_responses=True

Set this in ALL clients. Without it, Redis returns bytes objects. With it, returns Python str. Nearly always what you want.

1. Strings — Most Common Type (Key→Value)

r.set("inference:result:user-123", '{"answer": "42", "confidence": 0.98}', ex=300)
result = r.get("inference:result:user-123")  # None if expired

r.setex("rate:user-123", 60, 1)  # Set with TTL of 60 seconds
r.incr("rate:user-123")           # Atomic increment — thread-safe counter
remaining = r.ttl("rate:user-123")  # Time to live in seconds

2. Hashes — Object with Multiple Fields

Use for session data, user profiles — avoids serializing/deserializing a full JSON blob when you only need one field.

r.hset("user:1001", mapping={"name": "Alice", "tier": "premium", "credits": "500"})
name = r.hget("user:1001", "name")        # Get one field
profile = r.hgetall("user:1001")          # Get all fields as dict
r.hincrby("user:1001", "credits", -10)   # Atomically decrement credits

3. Lists — Ordered Queue / Recent History

r.lpush("chat:session-abc", "Hello")   # Push to left
r.rpush("chat:session-abc", "World")   # Push to right
history = r.lrange("chat:session-abc", 0, -1)  # All items
r.ltrim("chat:session-abc", 0, 9)      # Keep only last 10 messages

4. Sets — Unique Members

r.sadd("active-sessions", "sess-001", "sess-002")
r.sismember("active-sessions", "sess-001")  # True/False
r.smembers("active-sessions")               # All members

5. Sorted Sets — Leaderboards / Priority Queues

r.zadd("model-latency", {"gpt-4o": 250.5, "gpt-4o-mini": 85.2})
fastest = r.zrange("model-latency", 0, -1, withscores=True)  # Ascending (fastest first)

6. TTL Strategy by Data Type

Create RediSearch Vector Index

import redis, struct

r = redis.Redis(
    host="myredis.redis.cache.windows.net",
    port=10000, password="key", ssl=True,
    decode_responses=False  # MUST be False for binary vectors
)

r.execute_command(
    "FT.CREATE", "idx:docs", "ON", "HASH",
    "PREFIX", "1", "doc:",
    "SCHEMA",
    "title", "TEXT",
    "embedding", "VECTOR", "HNSW", "6",
        "TYPE", "FLOAT32",
        "DIM", "1536",
        "DISTANCE_METRIC", "COSINE"
)

# Store document with embedding
def embed(text):
    vec = oai.embeddings.create(input=text,
        model="text-embedding-3-small").data[0].embedding
    return struct.pack(f"{len(vec)}f", *vec)

r.hset("doc:1", mapping={
    "title": "Azure Key Vault",
    "embedding": embed("Azure Key Vault secures secrets")
})

FT.SEARCH with KNN

def vector_search(query_text, top_k=5):
    q_emb = embed(query_text)
    results = r.execute_command(
        "FT.SEARCH", "idx:docs",
        f"*=>[KNN {top_k} @embedding $blob AS score]",
        "PARAMS", "2", "blob", q_emb,
        "RETURN", "3", "title", "score",
        "SORTBY", "score",
        "DIALECT", "2"
    )
    return results

Cache-Aside with Similarity Threshold

import hashlib

THRESHOLD = 0.95  # cosine similarity for cache hit

def semantic_cache_get(prompt):
    q_emb = embed(prompt)
    res = r.execute_command(
        "FT.SEARCH", "idx:cache",
        "*=>[KNN 1 @embedding $blob AS score]",
        "PARAMS", "2", "blob", q_emb,
        "RETURN", "2", "response", "score",
        "DIALECT", "2"
    )
    if res[0] > 0:
        distance = float(dict(zip(res[2][::2],
            res[2][1::2]))[b"score"])
        if (1 - distance) >= THRESHOLD:
            return res[2][res[2].index(b"response")+1]
    return None

def semantic_cache_set(prompt, response, ttl=3600):
    key = f"cache:{hashlib.md5(prompt.encode()).hexdigest()}"
    r.hset(key, mapping={
        "prompt": prompt,
        "response": response,
        "embedding": embed(prompt)
    })
    r.expire(key, ttl)