Menggunakan Peramal Kesesakan Trafik AI menggunakan AWS Bedrock: Gambaran Keseluruhan Lengkap

Kita semua sukakan lalu lintas, bukan? Satu-satunya masa di mana saya berfikir tentang bagaimana saya benar-benar merosakkan pembentangan saya (terlalu berfikir adalah menyakitkan).

Ketepikan semua jenaka, saya ingin mencipta projek di mana saya boleh mencari trafik dalam masa nyata sebagai PoC supaya saya mempertingkatkannya lagi pada masa hadapan. Temui peramal kesesakan lalu lintas.

Saya akan meneruskan penggunaan Peramal Kesesakan Trafik menggunakan AWS Bedrock. AWS Bedrock menyediakan perkhidmatan terurus sepenuhnya untuk model asas, menjadikannya sempurna untuk menggunakan aplikasi AI. Kami akan merangkumi segala-galanya daripada persediaan awal hingga penggunaan dan ujian akhir.

Sekarang, prasyarat

• Akaun AWS dengan kebenaran yang sesuai (Terpaksa menggunakan kad debit saya untuk pengesahan kerana saya menganggap ia percuma untuk digunakan untuk had tertentu. Sakit).
• Python 3.8
• Kod Peramal Kesesakan Lalu Lintas (daripada perkembangan sebelumnya)
• AWS CLI dipasang dan dikonfigurasikan
• Pengetahuan asas perkhidmatan Python dan AWS akan berfungsi dengan baik.

Langkah 1: Menyediakan Persekitaran Anda

Pertama, sediakan persekitaran pembangunan anda:

# 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

Langkah 2: Persediaan Batuan Dasar AWS

1. Navigasi ke AWS Console dan dayakan AWS Bedrock

2. Buat model baharu dalam Batuan Dasar:

• Pergi ke konsol AWS Bedrock
• Pilih "Akses model"
• Minta akses kepada keluarga model Claude
• Tunggu kelulusan (biasanya segera tetapi apa-apa boleh berlaku)

Langkah 3: Ubah suai Kod untuk Integrasi Batuan Dasar

Buat fail baharu "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)

Langkah 4: Buat Bahagian Belakang FastAPI

Buat "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))

Langkah 5: Cipta Infrastruktur AWS

Buat "infrastruktur.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)

Langkah 6: Menyimpan Aplikasi

Buat "Fail Docker:"

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"]

Buat "requirements.txt:"

fastapi
uvicorn
boto3
pandas
numpy
scikit-learn

Langkah 7: Sebarkan ke AWS

Jalankan arahan ini:

# 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

Langkah 8: Kemas kini Streamlit Frontend

Ubah suai "app.py" untuk menyambung ke 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()

Langkah 9: Pengujian dan Pemantauan

Uji titik akhir 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}'

Pantau menggunakan AWS CloudWatch:

• Sediakan papan pemuka CloudWatch
• Buat penggera untuk kadar ralat dan kependaman
• Pantau penggunaan dan kos API

Jika semuanya berjalan lancar. tahniah! Anda telah berjaya menggunakan peramal kesesakan lalu lintas. Lapik diri anda di belakang untuk yang itu! Pastikan anda memantau kos dan prestasi, mengemas kini model secara kerap dan melaksanakan saluran paip CI/CD. Langkah seterusnya ialah menambah pengesahan pengguna, meningkatkan pemantauan dan makluman, mengoptimumkan prestasi model dan menambah lebih banyak ciri berdasarkan maklum balas pengguna.

Terima kasih kerana membaca ini. Beritahu saya apa-apa pemikiran, soalan atau pemerhatian!

Atas ialah kandungan terperinci Menggunakan Peramal Kesesakan Trafik AI menggunakan AWS Bedrock: Gambaran Keseluruhan Lengkap. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!