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.

Building a Lightweight Debugging Agent

 

                                                             image generated by meta ai

Building a Lightweight Debugging Agent: Python, Perl, and Awk

In modern development, especially when managing massive log files from platforms like GitLab, sending raw data directly to an LLM is inefficient. This post explores how to build a terminal-level tool that pre-processes logs and searches codebases using a combination of classic Unix concepts and Python.

1. The Quick & Dirty: Unix Tools

For simple recursive string searches, Unix tools remain the fastest starting point.

Using Grep + Awk

You can use grep for the search and awk for filtering or formatting the output.

Bash

grep -rn "your_string" /path/to/project | awk -F: '{print "File: "$1", Line: "$2", Match: "$3}'

• -r: Recursive search

• -n: Show line numbers

Using Awk Alone

Awk is powerful for text processing but does not handle directory recursion well on its own. You must combine it with find:

Bash

find . -type f -exec awk '/your_string/ {print FILENAME":"NR": "$0}' {} +

---

2. The Power Players: Perl and Python

Perl: The Regex Powerhouse

Perl is excellent for file scanning and complex recursion using the File::Find module.

Perl

use File::Find;
my $search = "your_string";
find(sub {
    return unless -f;
    open my $fh, '<', $_ or return;
    while (<$fh>) {
        if (/$search/) {
            print "$File::Find::name: $. $_";
        }
    }
    close $fh;
}, ".");

Python: The CLI Builder

If you want to build a full terminal application, Python is the best choice. It allows for easy argument parsing and integration.

Pro Tip: For maximum terminal productivity, consider using ripgrep (rg). It is significantly faster than grep and built specifically for developer workflows.

---

3. "CodeBugAgent": A Minimal Debug Assistant

By combining these concepts, we can build a "CodeBugAgent" script that identifies common bug indicators like error, exception, TODO, or FIXME.

Core Architecture

• Python: Acts as the main agent and CLI controller.

• Regex (Perl-like): Handles sophisticated pattern matching.

• Awk-style Filtering: Used for post-processing stream results.

• LLM Integration: Formats the output specifically for AI debugging prompts.

---

4. Optimized Log Processing for LLMs

The most effective way to use AI for debugging is to pre-process large files locally to reduce token noise and cost.

The Recommended Workflow

1. Extract: Pull only error blocks and stack traces.

2. Deduplicate: Remove repeating identical error messages.

3. Summarize: Create a high-signal context block.

4. Analyze: Feed the clean, small context into an LLM (like GPT or Windsurf).

Log Analyzer Implementation

An improved agent can extract error sections with surrounding context lines to give the LLM enough information to understand the root cause:

Python

# Improved logic for extracting errors with 3 lines of context
def extract_errors(file_path, context=3):
    pattern = re.compile(r"ERROR|Exception|Traceback|FATAL", re.IGNORECASE)
    # ... logic to capture context and deduplicate ...

---

5. Final Takeaway

Building your own local pre-processing agent is a production-grade approach that is both efficient and scalable. By extracting only the top 1-5% of useful lines, you ensure much better debugging results from your LLM.