AWS Bedrock を使用した AI 交通渋滞予測器のデプロイ: 完全な概要

私たちは皆、交通が大好きですよね?自分のプレゼンテーションを完全に台無しにしてしまったことについて考えるのは、このときだけです (考えすぎるのは面倒です)。

冗談はさておき、将来さらに強化するために、PoC としてリアルタイムでトラフィックを検索できるプロジェクトを作成したいと考えていました。渋滞予測機能をご紹介します。

AWS Bedrock を使用して交通渋滞予測をデプロイする手順を説明します。 AWS Bedrock は基盤モデル用のフルマネージド サービスを提供し、AI アプリケーションのデプロイに最適です。初期セットアップから最終的な展開とテストまですべてをカバーします。

さて、前提条件

• 適切な権限を持つ AWS アカウント (一定の制限内は無料で使用できると思っていたため、検証にデビット カードを使用する必要がありました。痛かったです)。
• Python 3.8
• 渋滞予測コード (以前の開発からのもの)
• AWS CLI のインストールと構成
• Python と AWS サービスの基本的な知識があれば十分です。

ステップ 1: 環境を準備する

まず、開発環境をセットアップします。

# 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

ステップ 2: AWS Bedrock のセットアップ

1. AWS コンソールに移動し、AWS Bedrock を有効にします

2. Bedrock で新しいモデルを作成します:

• AWS Bedrock コンソールに移動します
• 「モデルアクセス」を選択します
• クロード モデル ファミリへのアクセスをリクエスト
• 承認を待ちます (通常は即時ですが、何かが起こる可能性があります)

ステップ 3: Bedrock 統合用にコードを変更する

新しいファイルを作成します"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)

ステップ 4: FastAPI バックエンドを作成する

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

ステップ 5: AWS インフラストラクチャを作成する

"infrastructor.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)

ステップ 6: アプリケーションをコンテナ化する

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

「requirements.txt:」
を作成します

fastapi
uvicorn
boto3
pandas
numpy
scikit-learn

ステップ 7: AWS にデプロイする

次のコマンドを実行します:

# 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

ステップ 8: Streamlit フロントエンドを更新する

API に接続するために "app.py" を変更します:

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

ステップ 9: テストとモニタリング

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

AWS CloudWatch を使用して監視する:

• CloudWatch ダッシュボードをセットアップする
• エラー率と遅延のアラームを作成する
• API の使用量とコストを監視します

すべてがうまくいけば。おめでとう！交通渋滞予測機能が正常に展開されました。そのために背中を押してください！コストとパフォーマンスを監視し、モデルを定期的に更新し、CI/CD パイプラインを実装するようにしてください。次のステップは、ユーザー認証の追加、監視とアラートの強化、モデルのパフォーマンスの最適化、ユーザーのフィードバックに基づいた機能の追加です。

読んでいただきありがとうございます。ご意見、ご質問、ご意見がございましたらお聞かせください。

以上がAWS Bedrock を使用した AI 交通渋滞予測器のデプロイ: 完全な概要の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。