Python code examples for implementing an ML ensemble using Redis

 

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 Runtime or FastAPI workers 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.