Déploiement d'un prédicteur de congestion du trafic IA à l'aide d'AWS Bedrock : un aperçu complet

Nous aimons tous le trafic, n'est-ce pas ? La seule fois où je pense à quel point j'ai complètement raté ma présentation (trop réfléchir est pénible).

Blague à part, j'ai voulu créer un projet dans lequel je peux rechercher du trafic en temps réel en tant que PoC afin de l'améliorer encore à l'avenir. Rencontrez le prédicteur d’embouteillages.

Je vais vous expliquer le déploiement de Traffic Congestion Predictor à l'aide d'AWS Bedrock. AWS Bedrock fournit un service entièrement géré pour les modèles de base, ce qui le rend parfait pour le déploiement d'applications d'IA. Nous couvrirons tout, de la configuration initiale au déploiement final et aux tests.

Maintenant, les prérequis

• Compte AWS avec les autorisations appropriées (J'ai dû utiliser ma carte de débit pour la vérification car je pensais que son utilisation était gratuite jusqu'à une certaine limite. Douleur).
• Python 3.8
• Code de prévision des embouteillages (issu d'un développement précédent)
• AWS CLI installé et configuré
• Une connaissance de base des services Python et AWS fera très bien l'affaire.

Étape 1 : Préparer votre environnement

Tout d'abord, configurez votre environnement de développement :

# Create a new virtual environment
python -m venv bedrock-env
source bedrock-env/bin/activate  # On Windows use: bedrock-env\Scripts\activate

# Install required packages
pip install boto3 pandas numpy scikit-learn streamlit plotly

Étape 2 : Configuration d'AWS Bedrock

1. Accédez à la console AWS et activez AWS Bedrock

2. Créer un nouveau modèle dans Bedrock :

• Accédez à la console AWS Bedrock
• Sélectionnez "Accès au modèle"
• Demander l'accès à la famille modèle Claude
• Attendez l'approbation (généralement instantanée mais tout peut arriver)

Étape 3 : Modifier le code pour l'intégration de Bedrock

Créez un nouveau fichier "bedrock_integration.py" :

import boto3
import json
import numpy as np
import pandas as pd
from typing import Dict, Any

class TrafficPredictor:
    def __init__(self):
        self.bedrock = boto3.client(
            service_name='bedrock-runtime',
            region_name='us-east-1'  # Change to your region
        )

    def prepare_features(self, input_data: Dict[str, Any]) -> pd.DataFrame:
        # Convert input data to model features
        hour = input_data['hour']
        day = input_data['day']

        features = pd.DataFrame({
            'hour_sin': [np.sin(2 * np.pi * hour/24)],
            'hour_cos': [np.cos(2 * np.pi * hour/24)],
            'day_sin': [np.sin(2 * np.pi * day/7)],
            'day_cos': [np.cos(2 * np.pi * day/7)],
            'temperature': [input_data['temperature']],
            'precipitation': [input_data['precipitation']],
            'special_event': [input_data['special_event']],
            'road_work': [input_data['road_work']],
            'vehicle_count': [input_data['vehicle_count']]
        })
        return features

    def predict(self, input_data: Dict[str, Any]) -> float:
        features = self.prepare_features(input_data)

        # Prepare prompt for Claude
        prompt = f"""
        Based on the following traffic conditions, predict the congestion level (0-10):
        - Time: {input_data['hour']}:00
        - Day of week: {input_data['day']}
        - Temperature: {input_data['temperature']}°C
        - Precipitation: {input_data['precipitation']}mm
        - Special event: {'Yes' if input_data['special_event'] else 'No'}
        - Road work: {'Yes' if input_data['road_work'] else 'No'}
        - Vehicle count: {input_data['vehicle_count']}

        Return only the numerical prediction.
        """

        # Call Bedrock
        response = self.bedrock.invoke_model(
            modelId='anthropic.claude-v2',
            body=json.dumps({
                "prompt": prompt,
                "max_tokens": 10,
                "temperature": 0
            })
        )

        # Parse response
        response_body = json.loads(response['body'].read())
        prediction = float(response_body['completion'].strip())

        return np.clip(prediction, 0, 10)

Étape 4 : Créer un backend FastAPI

Créer "api.py :"

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from bedrock_integration import TrafficPredictor
from typing import Dict, Any

app = FastAPI()
predictor = TrafficPredictor()

class PredictionInput(BaseModel):
    hour: int
    day: int
    temperature: float
    precipitation: float
    special_event: bool
    road_work: bool
    vehicle_count: int

@app.post("/predict")
async def predict_traffic(input_data: PredictionInput) -> Dict[str, float]:
    try:
        prediction = predictor.predict(input_data.dict())
        return {"congestion_level": prediction}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Étape 5 : Créer une infrastructure AWS

Créer "infrastructure.py" :

import boto3
import json

def create_infrastructure():
    # Create ECR repository
    ecr = boto3.client('ecr')
    try:
        ecr.create_repository(repositoryName='traffic-predictor')
    except ecr.exceptions.RepositoryAlreadyExistsException:
        pass

    # Create ECS cluster
    ecs = boto3.client('ecs')
    ecs.create_cluster(clusterName='traffic-predictor-cluster')

    # Create task definition
    task_def = {
        'family': 'traffic-predictor',
        'containerDefinitions': [{
            'name': 'traffic-predictor',
            'image': f'{ecr.describe_repositories()["repositories"][0]["repositoryUri"]}:latest',
            'memory': 512,
            'cpu': 256,
            'essential': True,
            'portMappings': [{
                'containerPort': 8000,
                'hostPort': 8000,
                'protocol': 'tcp'
            }]
        }],
        'requiresCompatibilities': ['FARGATE'],
        'networkMode': 'awsvpc',
        'cpu': '256',
        'memory': '512'
    }

    ecs.register_task_definition(**task_def)

Étape 6 : Conteneuriser l’application

Créer "Dockerfile :"

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

Créer "requirements.txt :"

fastapi
uvicorn
boto3
pandas
numpy
scikit-learn

Étape 7 : Déployer sur AWS

Exécutez ces commandes :

# Build and push Docker image
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com
docker build -t traffic-predictor .
docker tag traffic-predictor:latest $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/traffic-predictor:latest
docker push $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/traffic-predictor:latest

# Create infrastructure
python infrastructure.py

Étape 8 : Mettre à jour l’interface Streamlit

Modifiez "app.py" pour vous connecter à l'API :

import streamlit as st
import requests
import plotly.graph_objects as go
import plotly.express as px

API_ENDPOINT = "your-api-endpoint"

def predict_traffic(input_data):
    response = requests.post(f"{API_ENDPOINT}/predict", json=input_data)
    return response.json()["congestion_level"]

# Rest of the Streamlit code remains the same, but replace direct model calls
# with API calls using predict_traffic()

Étape 9 : Tests et surveillance

Testez le point de terminaison de l'API :

curl -X POST "your-api-endpoint/predict" \
     -H "Content-Type: application/json" \
     -d '{"hour":12,"day":1,"temperature":25,"precipitation":0,"special_event":false,"road_work":false,"vehicle_count":1000}'

Surveiller à l'aide d'AWS CloudWatch :

• Configurer le tableau de bord CloudWatch
• Créer des alarmes pour les taux d'erreur et la latence
• Surveiller l'utilisation et les coûts de l'API

Si tout se passe bien. Félicitations! Vous avez déployé avec succès un prédicteur d'embouteillages. Rembourrez-vous le dos pour celui-là ! Assurez-vous de surveiller les coûts et les performances, de mettre régulièrement à jour le modèle et de mettre en œuvre un pipeline CI/CD. Les prochaines étapes consistent à ajouter l'authentification des utilisateurs, à améliorer la surveillance et les alertes, à optimiser les performances du modèle et à ajouter davantage de fonctionnalités basées sur les commentaires des utilisateurs.

Merci d'avoir lu ceci. Faites-moi part de vos réflexions, questions ou observations !

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!