Bereitstellung eines KI-Traffic-Congestion-Predictor mit AWS Bedrock: Ein vollständiger Überblick

Wir alle lieben den Verkehr, oder? Das einzige Mal, dass ich darüber nachdenke, wie ich meine Präsentation völlig durcheinander gebracht habe (zu viel nachzudenken ist eine Qual).

Spaß beiseite, ich wollte ein Projekt erstellen, bei dem ich als PoC in Echtzeit nach Traffic suchen kann, um es in Zukunft weiter zu verbessern. Lernen Sie den Verkehrsstau-Vorhersager kennen.

Ich werde die Bereitstellung des Traffic Congestion Predictor mit AWS Bedrock durchgehen. AWS Bedrock bietet einen vollständig verwalteten Service für Foundation-Modelle und eignet sich daher perfekt für die Bereitstellung von KI-Anwendungen. Wir decken alles von der Ersteinrichtung bis zur endgültigen Bereitstellung und zum Testen ab.

Nun zu den Voraussetzungen

• AWS-Konto mit entsprechenden Berechtigungen (Ich musste zur Verifizierung meine Debitkarte verwenden, da ich davon ausgegangen bin, dass die Nutzung für ein bestimmtes Limit kostenlos ist. Schmerz).
• Python 3.8
• Verkehrsstau-Vorhersagecode (aus früherer Entwicklung)
• AWS CLI installiert und konfiguriert
• Grundkenntnisse in Python und AWS-Diensten reichen völlig aus.

Schritt 1: Vorbereiten Ihrer Umgebung

Richten Sie zunächst Ihre Entwicklungsumgebung ein:

# 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

Schritt 2: AWS Bedrock-Setup

1. Navigieren Sie zur AWS-Konsole und aktivieren Sie AWS Bedrock

2. Erstellen Sie ein neues Modell in Bedrock:

• Gehen Sie zur AWS Bedrock-Konsole
• Wählen Sie „Modellzugriff“
• Zugriff auf die Claude-Modellfamilie anfordern
• Warten Sie auf die Genehmigung (normalerweise sofort, aber alles kann passieren)

Schritt 3: Code für die Bedrock-Integration ändern

Erstellen Sie eine neue Datei "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)

Schritt 4: Erstellen Sie ein FastAPI-Backend

Erstellen Sie "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))

Schritt 5: AWS-Infrastruktur erstellen

Erstellen Sie "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)

Schritt 6: Containerisieren Sie die Anwendung

Erstellen Sie „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"]

Erstellen Sie "requirements.txt:"

fastapi
uvicorn
boto3
pandas
numpy
scikit-learn

Schritt 7: Bereitstellung in AWS

Führen Sie diese Befehle aus:

# 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

Schritt 8: Streamlit-Frontend aktualisieren

Ändern Sie "app.py", um eine Verbindung zur API herzustellen:

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()

Schritt 9: Testen und Überwachen

Testen Sie den API-Endpunkt:

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}'

Überwachen mit AWS CloudWatch:

• CloudWatch-Dashboard einrichten
• Erstellen Sie Alarme für Fehlerraten und Latenz
• API-Nutzung und -Kosten überwachen

Wenn alles gut geht. Glückwunsch! Sie haben einen Verkehrsstauvorhersager erfolgreich bereitgestellt. Machen Sie sich bereit dafür! Stellen Sie sicher, dass Sie Kosten und Leistung überwachen, das Modell regelmäßig aktualisieren und eine CI/CD-Pipeline implementieren. Die nächsten Schritte bestehen darin, die Benutzerauthentifizierung hinzuzufügen, die Überwachung und Warnung zu verbessern, die Modellleistung zu optimieren und weitere Funktionen basierend auf Benutzerfeedback hinzuzufügen.

Danke, dass Sie das gelesen haben. Teilen Sie mir Ihre Gedanken, Fragen oder Beobachtungen mit!

Das obige ist der detaillierte Inhalt vonBereitstellung eines KI-Traffic-Congestion-Predictor mit AWS Bedrock: Ein vollständiger Überblick. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!