Redis Stack for Enterprise AI: Architecture & Integration

import redis
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
import numpy as np
import os

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

INDEX_NAME = "semantic_cache_idx"
VECTOR_DIM = 1536

def create_redis_vector_index():
  """Create HNSW RediSearch index over JSON documents."""
  try:
      r.ft(INDEX_NAME).info()
      print("RediSearch Vector Index already exists.")
  except Exception:
      schema = (
          TextField("$.prompt", as_name="prompt"),
          TextField("$.response", as_name="response"),
          VectorField("$.embedding", "HNSW", {
              "TYPE": "FLOAT32",
              "DIM": VECTOR_DIM,
              "DISTANCE_METRIC": "COSINE"
          }, as_name="embedding")
      )
      definition = IndexDefinition(prefix=["cache:"], index_type=IndexType.JSON)
      r.ft(INDEX_NAME).create_index(fields=schema, definition=definition)
      print("Created RediSearch HNSW Vector Index successfully.")

def check_semantic_cache(query_vector: list[float], threshold: float = 0.95):
  """Query semantic cache for similar cached LLM response."""
  vector_bytes = np.array(query_vector, dtype=np.float32).tobytes()
  q = Query(f"*=>[KNN 1 @embedding $vec AS score]").return_fields("prompt", "response", "score").dialect(2)
  
  res = r.ft(INDEX_NAME).search(q, query_params={"vec": vector_bytes})
  if res.docs:
      doc = res.docs[0]
      score = 1.0 - float(doc.score)  # Convert cosine distance to similarity
      if score >= threshold:
          print(f"Semantic Cache HIT! Similarity: {score:.4f}")
          return doc.response
  print("Semantic Cache MISS.")
  return None

if __name__ == "__main__":
  create_redis_vector_index()
  dummy_vec = [0.012] * 1536
  cached_output = check_semantic_cache(dummy_vec)