Important Setup Note: RedisAI and RedisML are no longer actively maintained as of 2025/2026. Therefore, Pattern B has evolved into standard deployment patterns using lightweight modern tools (like
ONNX RuntimeorFastAPIworkers coupled directly with Redis), ensuring your pipeline remains production-grade.
Pattern A: Asynchronous Parallel Processing (Redis Streams)
Best for: Heavy models (Deep Learning, large Random Forests) that need separate workers or hardware (GPUs) to compute in parallel without locking your main API.
Architecture Flow
[Client Request]
│
▼
┌───────────┐ XADD ┌───────────────┐
│ FastAPI │ ────────────────>│ Redis Stream │──┐ (Model 1 Worker)
│ Gateway │ │(ml:requests) │──┼─> [Model 2 Worker]
└───────────┘ └───────────────┘──┘ (Model 3 Worker)
│ │
│ (Polls responses via │ XADD
│ unique Request ID) ▼
│ ┌───────────────┐ ┌───────────────┐
└────────────────────────│ Redis Hash │<─│ Ensemble │
│(ml:results:ID)│ │ Aggregator │
└───────────────┘ └───────────────┘
Python Implementation
You will need redis and scikit-learn for this pattern. Run pip install redis scikit-learn.
Python
import time
import json
import uuid
import threading
import redis
from sklearn.linear_model import LogisticRegression
import numpy as np
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# 1. API GATEWAY FUNCTION (Pushes tasks, awaits final decision)
def mock_api_gateway(features):
request_id = str(uuid.uuid4())
payload = {"request_id": request_id, "features": json.dumps(features)}
# Broadcast to the stream
r.xadd("ml:requests", payload)
print(f"[API] Dispatched request {request_id}")
# Poll for the ensemble aggregator's final answer
result_key = f"ml:results:{request_id}"
while True:
final_decision = r.hgetall(result_key)
if final_decision:
r.delete(result_key) # Clean up
return final_decision
time.sleep(0.01) # 10ms polling backoff
# 2. MODEL WORKERS (Simulating 2 different models processing concurrently)
def run_model_worker(model_name, weight_offset):
# Dummy local model initialization
model = LogisticRegression()
while True:
# Read from stream (simulating listening for real-time events)
messages = r.xread({"ml:requests": "$"}, block=100)
if not messages:
continue
for stream, msgs in messages:
for msg_id, data in msgs:
req_id = data["request_id"]
features = json.loads(data["features"])
# Mock prediction: calculate probability
raw_prob = float(np.mean(features)) * weight_offset
prob = min(max(raw_prob, 0.0), 1.0)
# Write individual prediction back to a Redis Hash unique to this request
r.hset(f"ml:preds:{req_id}", model_name, prob)
# 3. ENSEMBLE AGGREGATOR WORKER
def run_ensemble_aggregator():
while True:
# Scan for active prediction hashes
for key in r.scan_iter("ml:preds:*"):
req_id = key.split(":")[-1]
predictions = r.hgetall(key)
# Check if both expected model workers have responded
if len(predictions) == 2:
prob1 = float(predictions["model_alpha"])
prob2 = float(predictions["model_beta"])
# Ensemble Logic: Simple Average
ensemble_prob = (prob1 + prob2) / 2
decision = "Approved" if ensemble_prob > 0.5 else "Rejected"
# Publish final outcome back to API
r.hset(f"ml:results:{req_id}", mapping={
"decision": decision,
"probability": ensemble_prob
})
r.delete(key) # Clear intermediate predictions
# --- TEST EXECUTION LOOP ---
if __name__ == "__main__":
# Flush old data for clean demo
r.flushdb()
# Start background threads for workers
threading.Thread(target=run_model_worker, args=("model_alpha", 1.2), daemon=True).start()
threading.Thread(target=run_model_worker, args=("model_beta", 0.8), daemon=True).start()
threading.Thread(target=run_ensemble_aggregator, daemon=True).start()
time.sleep(1) # Allow workers to boot
# Simulate a client checking credit features
mock_features = [0.4, 0.6, 0.55, 0.7]
output = mock_api_gateway(mock_features)
print(f"[API] Final Ensemble Response: {output}")
Pattern B: Synchronous In-Memory Processing (Pipeline)
Best for: Light-to-medium models (like Scikit-Learn or XGBoost) where raw speed matters most. Instead of threading out across streams, your API queries a centralized Redis backend using a pipeline to get all required features at once, then locally runs predictions.
Architecture Flow
[Client Request] ──> [FastAPI Server]
│
├─> Redis Pipeline (MGET Features)
│ │
│ ▼
│ [Redis In-Memory Key Store]
│ │ (Returns user & context features <1ms)
▼ ▼
Executes Model 1, 2, & 3 locally
│
▼
Ensemble Aggregation ──> [Return Decision]
Python Implementation
Python
import redis
import json
import numpy as np
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# Mock some feature store data inside Redis
r.hset("features:user_99", mapping={"risk_score": "0.34", "account_age_days": "120"})
r.hset("features:merchant_44", mapping={"fraud_velocity": "0.05", "trust_rating": "0.91"})
def predict_ensemble_sync(user_id, merchant_id):
# 1. Pipeline Feature Store Retrieval (Single network round-trip)
pipe = r.pipeline()
pipe.hgetall(f"features:{user_id}")
pipe.hgetall(f"features:{merchant_44}")
user_feats, merchant_feats = pipe.execute()
# Combine feature vectors
base_features = [
float(user_feats["risk_score"]),
float(user_feats["account_age_days"]) / 365.0, # normalized
float(merchant_feats["fraud_velocity"]),
float(merchant_feats["trust_rating"])
]
# 2. Synchronous Local Ensemble Execution
# Model 1: Linear rule
pred_1 = 1.0 if base_features[0] > 0.5 else 0.0
# Model 2: Statistical weight
pred_2 = (base_features[3] * 0.7) + (1 - base_features[2]) * 0.3
# 3. Aggregation (Ensemble Rule: Weighted Voting)
final_score = (pred_1 * 0.4) + (pred_2 * 0.6)
return "Flagged Fraud" if final_score < 0.5 else "Authorized"
# Test synchronous run
decision = predict_ensemble_sync("user_99", "merchant_44")
print(f"[Sync API] Ensemble Outcome: {decision}")
Pattern C: Caching Decisions (The Shield)
Regardless of whether you use Pattern A or B, you should shield your models from processing repetitive inputs using an exact signature cache lookup.
Python
import hashlib
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_ensemble_decision_with_cache(user_id, context_payload):
# Generate a deterministic cache key based on inputs
payload_string = json.dumps(context_payload, sort_keys=True)
payload_hash = hashlib.sha256(payload_string.encode()).hexdigest()
cache_key = f"cache:decision:{user_id}:{payload_hash}"
# 1. Hit the cache layer
cached_decision = r.get(cache_key)
if cached_decision:
print("[Cache] Match found! Zero model calculation needed.")
return json.loads(cached_decision)
# 2. Cache Miss: Run your actual ensemble pipeline (from Pattern A or B)
print("[Cache] Miss. Running full ensemble inference...")
computed_decision = {"status": "Approved", "confidence": 0.89}
# 3. Save to cache with a 60-second Time-To-Live (TTL)
r.setex(cache_key, 60, json.dumps(computed_decision))
return computed_decision
# Test caching execution
payload = {"amount": 250.00, "location": "NY"}
print(get_ensemble_decision_with_cache("user_99", payload))
print(get_ensemble_decision_with_cache("user_99", payload)) # Secondary call hits cache
Building a Real-Time Feature Store with Redis This video outlines Redis' architectures, illustrating how it manages massive operational throughput for contextual data architectures in modern live-serving applications.
