🧠 Distributed AI Inference Processing System

A production-grade, high-throughput asynchronous AI inference pipeline built with Spring Boot 3.x, RabbitMQ, Redis, FastAPI (Python), Docker, and AWS.

This system implements task queues for model workers, result caching to minimize repeated inference latency, dead-letter-queue retry mechanics, task batching, and a real-time monitoring dashboard with glassmorphism dark-mode aesthetics.

---

🏗️ System Architecture

graph TB
    subgraph "Client Layer"
        C[REST Client / Dashboard UI]
    end

    subgraph "API Gateway - Spring Boot"
        API[REST API Controller]
        TS[Task Service]
        RC[Redis Cache]
        RP[RabbitMQ Publisher]
        SC[Status Controller]
    end

    subgraph "Message Broker"
        RMQ[RabbitMQ]
        WQ[inference.tasks Queue]
        DLX[Dead Letter Exchange]
        DLQ[inference.tasks.dlq]
        RQ[inference.results Queue]
    end

    subgraph "AI Workers - FastAPI x3"
        W1[Worker 1 - Sentiment]
        W2[Worker 2 - Classification]
        W3[Worker 3 - General]
    end

    subgraph "Data Layer"
        RED[(Redis Cache + Task Store)]
    end

    C -->|POST /api/v1/inference| API
    C -->|GET /api/v1/inference/:id/status| SC
    API --> TS
    TS -->|Check Cache| RC
    RC -->|Cache Hit| TS
    TS -->|Cache Miss| RP
    RP -->|Publish Task| WQ
    WQ -->|Consume| W1 & W2 & W3
    W1 & W2 & W3 -->|Publish Result| RQ
    RQ -->|Consume Result| TS
    TS -->|Store Result + Cache| RED
    WQ -->|On Failure| DLX
    DLX --> DLQ
    DLQ -->|Retry| WQ

Loading

1. Client Submission: Clients send JSON prompts to the Gateway API (single request or batch).
2. Deterministic Caching: Gateway hashes the input text + model configuration. If the key exists in Redis, it serves the result instantly (~0-2ms cache hit vs 1.8s uncached).
3. Broker Dispatch: On a cache miss, the task is stored in Redis under a PENDING status and published onto a RabbitMQ queue with specific priorities.
4. Parallel Processing: FastAPI Python workers consume the tasks, run TF-IDF vectors + Logistic Regression / Naive Bayes predictions (scikit-learn), and publish the results to a result queue.
5. Gateway Consumption: The Gateway consumes the result queue, updates task records in Redis to COMPLETED, and saves the prediction to the cache with a 10-minute TTL.

---

✨ Features

• Asynchronous Task Queue: Fair dispatching of tasks to python inference replicas via RabbitMQ.
• Dead Letter Exchange (DLQ): Retries failing requests up to 3 times with exponential backoff before sending them to a DLQ for inspection.
• Task Batching: Allows submitting multiple requests concurrently with single status endpoint tracking.
• Redis Cache Layer: Stores hashed inputs, dropping repeat execution latency to near-zero.
• Real-time Monitoring UI: High-fidelity dark mode dashboard displaying latency analytics on HTML5 Canvas, queue stats, and live execution status.
• Load Testing Tool: Standalone python client script testing parallel capacity of 500+ requests.
• Infrastructure as Code: Ready-to-deploy Docker Compose and AWS CloudFormation templates.

---

🚀 Quick Start

Prerequisites

• Docker & Docker Compose installed.
• Python 3.11+ (optional, for local load testing).

Running with Docker Compose

To start the entire network (Gateway, Redis, RabbitMQ, and 3 workers):

docker-compose up --build -d

Verify that all containers are healthy:

docker-compose ps

Access Points

• Dashboard UI: http://localhost:8080/
• Gateway Actuator Health: http://localhost:8080/actuator/health
• RabbitMQ Management Console: http://localhost:15672/ (Credentials: admin / admin123)
• FastAPI Workers Docs:
  • Worker 1: http://localhost:8001/docs
  • Worker 2: http://localhost:8002/docs
  • Worker 3: http://localhost:8003/docs

---

📡 API Reference

1. Submit Single Inference

curl -X POST http://localhost:8080/api/v1/inference \
  -H "Content-Type: application/json" \
  -d '{
    "input": "This product exceeded my expectations! Incredible quality.",
    "modelType": "SENTIMENT",
    "priority": 1,
    "useCache": true
  }'

Response (200 OK):

{
  "taskId": "7f9e8a8b-4b11-482a-bc9e-e31464fb56a8",
  "status": "QUEUED",
  "message": "Task submitted successfully"
}

2. Submit Batch Inference

curl -X POST http://localhost:8080/api/v1/inference/batch \
  -H "Content-Type: application/json" \
  -d '[
    {"input": "I absolute hate the delay in shipping.", "modelType": "SENTIMENT"},
    {"input": "New CPU microarchitecture achieves double efficiency.", "modelType": "CLASSIFICATION"}
  ]'

3. Check Task Status

curl http://localhost:8080/api/v1/inference/7f9e8a8b-4b11-482a-bc9e-e31464fb56a8

Response (200 OK - Finished):

{
  "taskId": "7f9e8a8b-4b11-482a-bc9e-e31464fb56a8",
  "status": "COMPLETED",
  "result": "positive",
  "confidence": 0.985,
  "latencyMs": 142,
  "cached": false,
  "message": "Task status: COMPLETED"
}

---

📊 Parallel Load Testing

Run the included load testing utility to assert system capabilities under concurrency:

1. Install requirements:

pip install -r inference-worker/requirements.txt

2. Run test (500 requests, 50 concurrency):

python scripts/load_test.py --requests 500 --concurrency 50

---

🛠️ Tech Stack & Ports

Technology	Role	Port
Spring Boot 3.3	Gateway, Task Distributor & Cache Manager	8080
FastAPI / Python 3.11	ML Workers (Sentiment & Classification)	8001, 8002, 8003
RabbitMQ 3.13	Task Broker & Dead-letter Management	5672, Management: 15672
Redis 7.2	Execution Result Store & TTL Caching	6379
Docker	Orchestrator & Multi-agent scaling	-

---

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.