Implementing Semantic Cache Coherency in Multi-Node LLM Deployments using Redis Raft — code snippets

snippet-1.py

import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.http import models

class SemanticVectorResolver:
    def __init__(self, host: str, port: int, collection_name: str, threshold: float = 0.88):
        self.client = QdrantClient(host=host, port=port)
        self.collection_name = collection_name
        self.threshold = threshold

    def find_nearest_cache_id(self, query_vector: list[float]) -> str | None:
        """
        Queries Qdrant for the closest semantic match. Returns the Redis key UUID
        if the similarity score is above the strict production threshold.
        """
        try:
            results = self.client.search(
                collection_name=self.collection_name,
                query_vector=query_vector,
                limit=1,
                with_payload=True,
                with_vectors=False
            )
            if not results:
                return None
            
            best_match = results[0]
            # Use cosine similarity threshold matching
            if best_match.score >= self.threshold:
                # The payload contains the unique key pointing to the Redis Raft store
                return best_match.payload.get("redis_cache_key")
        except Exception as e:
            # In production, log this to a metrics pipeline (e.g., Prometheus)
            # and fall back to a cache miss to prevent gateway blocking
            print(f"Vector search failed: {str(e)}")
        return None

snippet-2.go

package cache

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

type GatewayCoordinator struct {
	rdb          *redis.Client
	vectorClient *SemanticVectorResolver // Interfaced python/gRPC wrapper
	llmClient    *LLMProviderClient
}

type CacheResult struct {
	Payload    string
	FromCache  bool
	DurationMs int64
}

func (g *GatewayCoordinator) ExecutePrompt(ctx context.Context, prompt string, embedding []float32) (*CacheResult, error) {
	startTime := time.Now()

	// 1. Search the Vector Plane
	cacheUUID := g.vectorClient.FindNearestCacheID(embedding)
	if cacheUUID != "" {
		// 2. Linearizable Read from Redis Raft
		// We use a strongly consistent read command to ensure we aren't reading split-brain state
		val, err := g.rdb.Get(ctx, fmt.Sprintf("cache:%s", cacheUUID)).Result()
		if err == nil {
			return &CacheResult{
				Payload:    val,
				FromCache:  true,
				DurationMs: time.Since(startTime).Milliseconds(),
			}, nil
		}
		if !errors.Is(err, redis.Nil) {
			// Log connection errors but continue to LLM to preserve uptime
			fmt.Printf("Raft read failure: %v\n", err)
		}
	}

	// 3. Cache Miss: Acquire Distributed Lock (Lease) on Redis Raft
	// Raft guarantees this lock is distributed and strictly consistent
	lockKey := fmt.Sprintf("lock:%s", cacheUUID)
	if cacheUUID == "" {
		cacheUUID = GenerateUUID()
		lockKey = fmt.Sprintf("lock:%s", cacheUUID)
	}

	acquired, err := g.rdb.SetNX(ctx, lockKey, "locked", 10*time.Second).Result()
	if err != nil {
		return nil, err
	}

	if !acquired {
		// Herd mitigation: Wait and poll the consistent state
		return g.pollForCacheValue(ctx, cacheUUID, startTime)
	}
	defer g.rdb.Del(ctx, lockKey)

	// 4. Invoke the LLM
	llmResponse, err := g.llmClient.Call(ctx, prompt)
	if err != nil {
		return nil, err
	}

	// 5. Populate Cache Planes
	// Write to Raft first to ensure consensus commit before writing to Vector index
	err = g.commitToRaftAndVector(ctx, cacheUUID, embedding, llmResponse)
	if err != nil {
		return nil, err
	}

	return &CacheResult{
		Payload:    llmResponse,
		FromCache:  false,
		DurationMs: time.Since(startTime).Milliseconds(),
	}, nil
}

func (g *GatewayCoordinator) pollForCacheValue(ctx context.Context, uuid string, start time.Time) (*CacheResult, error) {
	ticker := time.NewTicker(200 * time.Millisecond)
	defer ticker.Stop()
	timeout := time.After(8 * time.Second)

	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-timeout:
			return nil, errors.New("timeout waiting for concurrent cache write")
		case <-ticker.C:
			val, err := g.rdb.Get(ctx, fmt.Sprintf("cache:%s", uuid)).Result()
			if err == nil {
				return &CacheResult{
					Payload:    val,
					FromCache:  true,
					DurationMs: time.Since(start).Milliseconds(),
				}, nil
			}
		}
	}
}

snippet-3.go

package cache

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// WriteBackLuaScript ensures metadata validation and lock release are committed
// atomically across the Raft consensus cluster.
const WriteBackLuaScript = `
local cache_key = KEYS[1]
local lock_key = KEYS[2]
local payload = ARGV[1]
local ttl = tonumber(ARGV[2])

-- Write the cache payload with linearizable consistency
redis.call('SET', cache_key, payload, 'EX', ttl)

-- Release the execution lock
redis.call('DEL', lock_key)

return 1
`

func (g *GatewayCoordinator) commitToRaftAndVector(ctx context.Context, uuid string, embedding []float32, response string) error {
	cacheKey := fmt.Sprintf("cache:%s", uuid)
	lockKey := fmt.Sprintf("lock:%s", uuid)

	// Execute Lua script on Redis Raft Leader
	// The command propagates through the Raft log to all followers before returning success
	_, err := g.rdb.Eval(ctx, WriteBackLuaScript, []string{cacheKey, lockKey}, response, 3600).Result()
	if err != nil {
		return fmt.Errorf("failed committing state to Raft: %w", err)
	}

	// Index the vector. If this fails, we log it. The next cache lookup will miss,
	// but our metadata layer remains clean.
	err = g.vectorClient.UpsertVector(uuid, embedding)
	if err != nil {
		fmt.Printf("Vector plane write failed (eventual consistency recovery required): %v\n", err)
	}

	return nil
}

snippet-4.yaml

version: '3.8'

services:
  raft-node-1:
    image: redislabs/redisraft:latest
    container_name: raft-node-1
    command: >
      redis-server 
      --loadmodule /usr/lib/redis/modules/redisraft.so
      addr=raft-node-1:5001
      bootstrap=true
      raft.log-filename=/data/raft.log
      raft.election-timeout=2000
      raft.heartbeat-interval=400
      --port 5001
      --dir /data
      --appendonly yes
    ports:
      - "5001:5001"
    volumes:
      - raft_data_1:/data

  raft-node-2:
    image: redislabs/redisraft:latest
    container_name: raft-node-2
    command: >
      redis-server 
      --loadmodule /usr/lib/redis/modules/redisraft.so
      addr=raft-node-2:5002
      join=raft-node-1:5001
      raft.log-filename=/data/raft.log
      raft.election-timeout=2000
      raft.heartbeat-interval=400
      --port 5002
      --dir /data
      --appendonly yes
    ports:
      - "5002:5002"
    depends_on:
      - raft-node-1
    volumes:
      - raft_data_2:/data

  raft-node-3:
    image: redislabs/redisraft:latest
    container_name: raft-node-3
    command: >
      redis-server 
      --loadmodule /usr/lib/redis/modules/redisraft.so
      addr=raft-node-3:5003
      join=raft-node-1:5001
      raft.log-filename=/data/raft.log
      raft.election-timeout=2000
      raft.heartbeat-interval=400
      --port 5003
      --dir /data
      --appendonly yes
    ports:
      - "5003:5003"
    depends_on:
      - raft-node-1
    volumes:
      - raft_data_3:/data

volumes:
  raft_data_1:
  raft_data_2:
  raft_data_3:

snippet-5.py

import redis
from qdrant_client import QdrantClient

class SemanticInvalidator:
    def __init__(self, raft_host: str, raft_port: int, vector_host: str, vector_port: int, collection: str):
        # Connect to the Redis Raft cluster leader (which handles writes)
        self.raft = redis.Redis(host=raft_host, port=raft_port, decode_responses=True)
        self.vector_store = QdrantClient(host=vector_host, port=vector_port)
        self.collection = collection

    def invalidate_concept(self, target_embedding: list[float], threshold: float = 0.90) -> int:
        """
        Locates all cached queries that are semantically close to target_embedding,
        and invalidates them atomically in the Redis Raft consensus cluster.
        """
        # 1. Query vector space for candidates
        candidates = self.vector_store.search(
            collection_name=self.collection,
            query_vector=target_embedding,
            limit=100,
            with_payload=True
        )
        
        keys_to_delete = []
        vector_ids_to_purge = []
        
        for candidate in candidates:
            if candidate.score >= threshold:
                redis_key = candidate.payload.get("redis_cache_key")
                if redis_key:
                    keys_to_delete.append(f"cache:{redis_key}")
                    vector_ids_to_purge.append(candidate.id)
                    
        if not keys_to_delete:
            return 0

        # 2. Atomic Delete on Redis Raft
        # Because we run under Raft, this commit propagates to all replicas before acknowledging.
        # This prevents any gateway node from serving stale data.
        pipeline = self.raft.pipeline()
        for k in keys_to_delete:
            pipeline.delete(k)
        pipeline.execute()
        
        # 3. Asynchronously clean up the vector store plane
        try:
            self.vector_store.delete(
                collection_name=self.collection,
                points_selector=vector_ids_to_purge
            )
        except Exception as ve:
            # Log for the background reconciliation engine to retry
            print(f"Eventual consistency: vector cleanup deferred: {str(ve)}")
            
        return len(keys_to_delete)