私たちは皆、交通が大好きですよね?自分のプレゼンテーションを完全に台無しにしてしまったことについて考えるのは、このときだけです (考えすぎるのは面倒です)。
冗談はさておき、将来さらに強化するために、PoC としてリアルタイムでトラフィックを検索できるプロジェクトを作成したいと考えていました。渋滞予測機能をご紹介します。
AWS Bedrock を使用して交通渋滞予測をデプロイする手順を説明します。 AWS Bedrock は基盤モデル用のフルマネージド サービスを提供し、AI アプリケーションのデプロイに最適です。初期セットアップから最終的な展開とテストまですべてをカバーします。
まず、開発環境をセットアップします。
# 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
AWS コンソールに移動し、AWS Bedrock を有効にします
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)
「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)
「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
次のコマンドを実行します:
# 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
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()
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 を使用して監視する:
すべてがうまくいけば。おめでとう！交通渋滞予測機能が正常に展開されました。そのために背中を押してください！コストとパフォーマンスを監視し、モデルを定期的に更新し、CI/CD パイプラインを実装するようにしてください。次のステップは、ユーザー認証の追加、監視とアラートの強化、モデルのパフォーマンスの最適化、ユーザーのフィードバックに基づいた機能の追加です。
読んでいただきありがとうございます。ご意見、ご質問、ご意見がございましたらお聞かせください。
以上がAWS Bedrock を使用した AI 交通渋滞予測器のデプロイ: 完全な概要の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
