Multi-Modal AI System Design

# Define vision service
curl -i -X POST http://kong:8001/services \
  --data name=vision-service \
  --data url=https://vision-service:3000

# Define vision route
curl -i -X POST http://kong:8001/services/vision-service/routes \
  --data 'paths[]=/vision' \
  --data methods[]=POST

# Define NLP service
curl -i -X POST http://kong:8001/services \
  --data name=nlp-service \
  --data url=https://nlp-service:3000

# Define NLP route
curl -i -X POST http://kong:8001/services/nlp-service/routes \
  --data 'paths[]=/nlp' \
  --data methods[]=POST

# Enable JWT plugin for vision service
curl -i -X POST http://kong:8001/services/vision-service/plugins \
  --data name=jwt

# Enable JWT plugin for NLP service
curl -i -X POST http://kong:8001/services/nlp-service/plugins \
  --data name=jwt

# Enable rate-limiting plugin for vision service
curl -i -X POST http://kong:8001/services/vision-service/plugins \
  --data name=rate-limiting \
  --data config.second=10 \
  --data config.hour=2000 \
  --data config.policy=redis \
  --data config.redis_host=redis-host

# Enable rate-limiting plugin for NLP service
curl -i -X POST http://kong:8001/services/nlp-service/plugins \
  --data name=rate-limiting \
  --data config.second=10 \
  --data config.hour=2000 \
  --data config.policy=redis \
  --data config.redis_host=redis-host

# Enable Prometheus plugin
curl -i -X POST http://kong:8001/plugins \
  --data name=prometheus
                

from flask import Flask, request, jsonify
import jwt
import redis
from pymongo import MongoClient
import tensorflow as tf
import numpy as np
import os
import requests

app = Flask(__name__)
JWT_SECRET = os.getenv('JWT_SECRET', 'your-secret-key')
REDIS_HOST = 'redis://redis-host:6379'
MONGO_URI = 'mongodb://mongo:27017'
VISION_SERVICE_URL = 'https://vision-service:3000/vision'
NLP_SERVICE_URL = 'https://nlp-service:3000/nlp'
MODEL_PATH = '/models/fusion_model/1'

# Initialize clients
redis_client = redis.Redis.from_url(REDIS_HOST)
mongo_client = MongoClient(MONGO_URI)
db = mongo_client['multi_modal']
fusion_model = tf.saved_model.load(MODEL_PATH)

def check_rbac(required_role):
    def decorator(f):
        def wrapper(*args, **kwargs):
            auth_header = request.headers.get('Authorization')
            if not auth_header or not auth_header.startswith('Bearer '):
                return jsonify({'error': 'Unauthorized'}), 401
            token = auth_header.split(' ')[1]
            try:
                decoded = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
                if decoded.get('role') != required_role:
                    return jsonify({'error': 'Insufficient permissions'}), 403
                return f(*args, **kwargs)
            except jwt.InvalidTokenError:
                return jsonify({'error': 'Invalid token'}), 403
        return wrapper
    return decorator

@app.route('/decision', methods=['POST'])
@check_rbac('decision')
def make_decision():
    data = request.json
    session_id = data['session_id']
    image_data = data.get('image_data')
    text_data = data.get('text_data')

    # Check cache
    cache_key = f'decision:{session_id}'
    cached = redis_client.get(cache_key)
    if cached:
        return jsonify({'decision': cached.decode('utf-8')})

    # Call Vision Service
    vision_output = None
    if image_data:
        vision_response = requests.post(
            VISION_SERVICE_URL,
            json={'image': image_data},
            headers={'Authorization': request.headers.get('Authorization')}
        )
        if vision_response.status_code == 200:
            vision_output = vision_response.json()['features']

    # Call NLP Service
    nlp_output = None
    if text_data:
        nlp_response = requests.post(
            NLP_SERVICE_URL,
            json={'text': text_data},
            headers={'Authorization': request.headers.get('Authorization')}
        )
        if nlp_response.status_code == 200:
            nlp_output = nlp_response.json()['features']

    # Combine outputs in fusion model
    inputs = {
        'vision': tf.convert_to_tensor(vision_output or np.zeros((1, 512))),
        'nlp': tf.convert_to_tensor(nlp_output or np.zeros((1, 768)))
    }
    decision = fusion_model(inputs).numpy().tolist()

    # Cache and store decision
    redis_client.setex(cache_key, 3600, str(decision))
    db['decisions'].update_one(
        {'session_id': session_id},
        {'$set': {
            'vision_output': vision_output,
            'nlp_output': nlp_output,
            'decision': decision,
            'updated_at': datetime.now()
        }},
        upsert=True
    )

    return jsonify({'decision': decision})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, ssl_context=('server-cert.pem', 'server-key.pem'))
                

from flask import Flask, request, jsonify
import jwt
import redis
import cv2
import numpy as np
import os
from yolov5 import YOLOv5

app = Flask(__name__)
JWT_SECRET = os.getenv('JWT_SECRET', 'your-secret-key')
REDIS_HOST = 'redis://redis-host:6379'
MODEL_PATH = '/models/yolov5s.pt'

# Initialize clients
redis_client = redis.Redis.from_url(REDIS_HOST)
yolo_model = YOLOv5(MODEL_PATH, device='cuda')

def check_rbac(required_role):
    def decorator(f):
        def wrapper(*args, **kwargs):
            auth_header = request.headers.get('Authorization')
            if not auth_header or not auth_header.startswith('Bearer '):
                return jsonify({'error': 'Unauthorized'}), 401
            token = auth_header.split(' ')[1]
            try:
                decoded = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
                if decoded.get('role') != required_role:
                    return jsonify({'error': 'Insufficient permissions'}), 403
                return f(*args, **kwargs)
            except jwt.InvalidTokenError:
                return jsonify({'error': 'Invalid token'}), 403
        return wrapper
    return decorator

@app.route('/vision', methods=['POST'])
@check_rbac('vision')
def process_image():
    data = request.json
    image_data = data['image']  # Base64-encoded image
    session_id = data['session_id']

    # Check cache
    cache_key = f'vision:{session_id}'
    cached = redis_client.get(cache_key)
    if cached:
        return jsonify({'features': cached.decode('utf-8')})

    # Decode and process image
    image = cv2.imdecode(np.frombuffer(base64.b64decode(image_data), np.uint8), cv2.IMREAD_COLOR)
    results = yolo_model.predict(image)

    # Extract features (e.g., bounding boxes, class probabilities)
    features = results.pandas().xyxy[0].to_dict()

    # Cache results
    redis_client.setex(cache_key, 3600, str(features))

    return jsonify({'features': features})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000, ssl_context=('server-cert.pem', 'server-key.pem'))