How to use cloud redis cache with python in my livekit project to reduce latency?

Thanks for reaching out, Sabbir. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

STT/LLM inference responses

Thanks — you want a focused, practical guide for caching STT and LLM inference responses with Redis in a Python LiveKit project to reduce latency. I’ll keep this actionable and show safe patterns you can adopt in your agent that calls LiveKit Inference or external AI services. Before code: one quick question • Do you call STT and LLM as separate steps (STT → LLM), or do you send raw audio directly to an LLM/ multimodal endpoint? This affects how you build keys (audio fingerprint vs transcript text). If unsure, I’ll assume STT produces text first and then the LLM consumes that text. Recommended approach (high level) • Cache STT outputs keyed by the audio fingerprint (hash of normalized audio bytes + STT model + parameters). • Cache LLM outputs keyed by normalized text + model + prompt/template + relevant params. • Use deterministic keys, TTLs appropriate for your use case, and a lock to avoid cache stampedes. • For larger audio or intermediate files, store them in object storage and keep references in Redis rather than the raw blob. • Run Redis in the same cloud region as your LiveKit agent/inference service to avoid adding network latency. Cache key strategy • STT key: stt:<sha256(audio_bytes_normalized)|model|params_hash> • LLM key: llm:<sha256(normalized_prompt_text)|model|params_hash> • Normalize inputs: trim whitespace, consistent sample rate/encoding for audio, canonicalize JSON for params (sorted keys). • Include model version and decoding/config params so different configs don’t collide. TTL guidance • STT results: short-medium TTL (e.g., 5–60 minutes) depending on how often the same audio is likely to repeat. • LLM results: medium TTL (e.g., 1 hour–24 hours) if outputs are reusable. Use shorter TTL if the content is user-specific or time-sensitive. • Consider no TTL for deterministic system prompts that never change (or very long TTL). Stampede protection and concurrency • Use a small Redis lock on misses (SET NX with an expiry) so only one worker computes the result. • Other workers poll for the cached value with short sleeps, then either return the cached result or compute locally after timeout. • Optionally use “populate-as-you-go” with background refresh to avoid blocking live requests. Storage of payloads • Small JSON/text outputs: store directly in Redis (JSON is fine). • Large LLM outputs or rich artifacts: store in object storage (S3/GCS) and save the URL/metadata in Redis. • For STT audio outputs (if you store audio segments), prefer object storage + reference in Redis. Practical Python example • This example uses the redis-py client. Replace

compute_stt

and

call_llm

with your actual calls to the STT and LLM endpoints (LiveKit Inference or third-party). 1) Install: • pip install redis 2) Example code:

import hashlib
import json
import redis
import time
from typing import Callable

# Configure Redis connection (update host/port/password)
r = redis.Redis(host='REDIS_HOST', port=6379, db=0, password='REDIS_PASS', decode_responses=True)

def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def make_key(prefix: str, parts: list) -> str:
    # parts should be strings already normalized
    joined = "|".join(parts)
    return f"{prefix}:" + sha256_hex(joined.encode())

def json_dumps_canonical(obj):
    return json.dumps(obj, sort_keys=True, separators=(',', ':'))

def cached_compute(key: str, compute_fn: Callable, ttl: int = 3600):
    """
    Generic cache wrapper with simple lock to avoid stampedes.
    compute_fn should return JSON-serializable result.
    """
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    lock_key = key + ":lock"
    lock_acquired = r.set(lock_key, "1", nx=True, ex=30)  # lock expires in 30s
    if lock_acquired:
        try:
            result = compute_fn()
            r.set(key, json.dumps(result), ex=ttl)
            return result
        finally:
            r.delete(lock_key)
    else:
        # Wait for the result to appear
        for _ in range(20):
            time.sleep(0.1)
            cached = r.get(key)
            if cached:
                return json.loads(cached)
        # Fallback: compute locally to avoid long waits
        result = compute_fn()
        # best-effort cache
        try:
            r.set(key, json.dumps(result), ex=ttl)
        except Exception:
            pass
        return result

# Example: caching STT
def compute_stt_from_audio(audio_bytes: bytes, stt_model: str, stt_params: dict):
    # Replace this with actual STT call to LiveKit Inference or your STT service.
    # Should return a dict, e.g., {"text": "...", "confidence": 0.98}
    raise NotImplementedError

def get_stt(audio_bytes: bytes, stt_model: str, stt_params: dict, ttl=1800):
    # Normalize audio: ensure consistent sample rate/encoding outside this snippet
    audio_hash = sha256_hex(audio_bytes)
    params_hash = sha256_hex(json_dumps_canonical(stt_params).encode())
    key = f"stt:{audio_hash}:{stt_model}:{params_hash}"
    return cached_compute(key, lambda: compute_stt_from_audio(audio_bytes, stt_model, stt_params), ttl=ttl)

# Example: caching LLM responses
def compute_llm_response(prompt_text: str, llm_model: str, llm_params: dict):
    # Replace this with actual LLM call (LiveKit Inference or external provider).
    # Return JSON-serializable dict like {"text": "...", "tokens": [...], ...}
    raise NotImplementedError

def get_llm(prompt_text: str, llm_model: str, llm_params: dict, ttl=3600):
    prompt_norm = " ".join(prompt_text.strip().split())  # simple normalization
    params_hash = sha256_hex(json_dumps_canonical(llm_params).encode())
    key = f"llm:{sha256_hex(prompt_norm.encode())}:{llm_model}:{params_hash}"
    return cached_compute(key, lambda: compute_llm_response(prompt_norm, llm_model, llm_params), ttl=ttl)

Integrating with your LiveKit agent • In your agent flow: when you receive audio, call

get_stt(...)

. If STT result exists, proceed directly to LLM; otherwise STT will be computed and cached. • For LLM calls, call

get_llm(...)

with normalized prompt and reused model params. • If you perform STT→LLM pipeline often with the same transcript and prompt template, consider composing a combined cache key for the final LLM output as well (e.g., hash of transcript + prompt template + model). Advanced improvements you can add • Use Redis Hashes to store additional metadata (timestamp, model_version, source_url). • Use Redis streams or task queue patterns if you want async background population of cache. • Implement versioned keys (e.g., include app version or model-version tag) so you can invalidate broad sets of keys after upgrades. • Use Redis Bloom / LFU counters to identify hot keys and tune TTLs. • Use Redis Cluster and connection pooling for production scale. Next steps I can help with • Provide a concrete snippet that wires

compute_stt_from_audio

to your actual LiveKit Inference STT API or to your STT provider — tell me which provider or paste the code you currently use. • Show an example that uploads audio to S3 and stores references in Redis for larger assets. • Help decide TTL values based on your usage patterns (streaming live audio vs repeated identical clips). Which of these would you like next: a LiveKit-specific STT/LLM wiring example, or wiring to a third-party STT/LLM provider you already use?