Redis AI Quick Reference - Integration Patterns

# Simple implementation
def cached_llm_call(prompt, model="gpt-4"):
    cache_key = f"llm:{hash(prompt + model)}"

    # Try cache first
    cached = redis.get(cache_key)
    if cached:
        return cached

    # Call LLM
    response = llm_api.complete(prompt, model)

    # Cache for 1 hour
    redis.setex(cache_key, 3600, response)
    return response

# Agent state in Redis Hash
def save_agent_state(agent_id, session_id, state):
    key = f"agent:{agent_id}:session:{session_id}"
    redis.hset(key, mapping={
        "context": json.dumps(state.context),
        "turn": state.turn_count,
        "updated": datetime.now().isoformat()
    })
    redis.expire(key, 86400)  # 24 hours

def load_agent_state(agent_id, session_id):
    key = f"agent:{agent_id}:session:{session_id}"
    return redis.hgetall(key)

# Using RediSearch for vector search
def store_embedding(doc_id, text, embedding):
    redis.json().set(f"doc:{doc_id}", "$", {
        "text": text,
        "embedding": embedding
    })

def search_similar(query_embedding, top_k=5):
    query = Query(f"*=>[KNN {top_k} @embedding $vec]") \
        .return_fields("text", "score") \
        .dialect(2)

    return redis.ft("idx").search(
        query,
        query_params={"vec": query_embedding}
    )

# Pub/Sub for agent communication
def agent_send_message(from_agent, to_channel, message):
    redis.publish(to_channel, json.dumps({
        "from": from_agent,
        "message": message,
        "timestamp": time.time()
    }))

def agent_listen(channel, callback):
    pubsub = redis.pubsub()
    pubsub.subscribe(channel)

    for msg in pubsub.listen():
        if msg['type'] == 'message':
            callback(json.loads(msg['data']))

# RedisAI for model serving
def run_inference(model_name, input_data):
    # Store input tensor
    redis.ai_tensorset(f"input:{req_id}", input_data)

    # Run model
    redis.ai_modelexecute(
        model_name,
        inputs=[f"input:{req_id}"],
        outputs=[f"output:{req_id}"]
    )

    # Get result
    result = redis.ai_tensorget(f"output:{req_id}")

    # Cleanup
    redis.delete(f"input:{req_id}", f"output:{req_id}")

    return result

import redis
from redis.cluster import RedisCluster

# Single instance
r = redis.Redis(
    host='localhost',
    port=6379,
    decode_responses=True,
    socket_timeout=5,
    socket_connect_timeout=5
)

# Cluster
cluster = RedisCluster(
    host='localhost',
    port=6379,
    decode_responses=True
)

# With connection pool
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=50
)
r = redis.Redis(connection_pool=pool)

const redis = require('redis');

// Single instance
const client = redis.createClient({
    host: 'localhost',
    port: 6379
});

await client.connect();

// Cluster
const { createCluster } = require('redis');
const cluster = createCluster({
    rootNodes: [
        { url: 'redis://node1:6379' },
        { url: 'redis://node2:6379' }
    ]
});

await cluster.connect();

import (
    "github.com/go-redis/redis/v8"
    "context"
)

// Single instance
rdb := redis.NewClient(&redis.Options{
    Addr: "localhost:6379",
    DB: 0,
})

// Cluster
rdb := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{
        "localhost:6379",
        "localhost:6380",
    },
})

# Check memory usage
redis-cli INFO memory

# See slow queries
redis-cli SLOWLOG GET 10

# Monitor commands in real-time
redis-cli MONITOR

# Check connected clients
redis-cli CLIENT LIST

# Database statistics
redis-cli INFO stats

# Key space analysis
redis-cli --bigkeys

# Memory analysis
redis-cli --memkeys

# redis.conf optimizations for LLM caching
maxmemory 8gb
maxmemory-policy allkeys-lru
maxmemory-samples 10

# Persistence (adjust based on needs)
save 900 1
save 300 10
save 60 10000

# Network
timeout 300
tcp-keepalive 60

# Focus on durability
appendonly yes
appendfsync everysec
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

# Memory
maxmemory-policy noeviction  # Don't evict agent state

# Performance optimization
maxmemory 16gb
maxmemory-policy noeviction

# Load RediSearch module
loadmodule /path/to/redisearch.so

# Increase search timeout
timeout 300

Need AI/LLM/Agent integration with Redis?
│
├─ Need to reduce LLM API costs?
│  └─ Use: Response caching pattern
│
├─ Need semantic search?
│  └─ Use: Vector similarity with RediSearch
│
├─ Need agent memory/state?
│  └─ Use: Hash + JSON for state management
│
├─ Need real-time AI processing?
│  └─ Use: Redis Streams + RedisAI
│
├─ Need multi-agent coordination?
│  └─ Use: Pub/Sub + shared state
│
├─ Need model serving?
│  └─ Use: RedisAI for inference
│
└─ Need RAG (Retrieval-Augmented Generation)?
   └─ Use: Vector search + LLM caching

# 1. Use strong authentication
redis = Redis(
    host='redis.example.com',
    password='strong-random-password',
    ssl=True
)

# 2. Implement ACLs
# redis-cli: ACL SETUSER myapp on >password ~myapp:* +get +set +del

# 3. Encrypt sensitive data
from cryptography.fernet import Fernet

key = Fernet.generate_key()
f = Fernet(key)

encrypted = f.encrypt(b"sensitive data")
redis.set("secure:key", encrypted)

decrypted = f.decrypt(redis.get("secure:key"))

# 4. Use separate Redis instances for different security levels
redis_public = Redis(host='public-redis')  # Non-sensitive
redis_secure = Redis(host='secure-redis')  # Sensitive data