使用 AWS Bedrock 部署 AI 交通拥堵预测器：完整概述

<code class="bash">python -m venv bedrock-env
source bedrock-env/bin/activate  # Windows 系统使用：bedrock-env\Scripts\activate

pip install boto3 pandas numpy scikit-learn streamlit plotly</code>

<code class="python">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'  # 请替换为您的区域
        )

    def prepare_features(self, input_data: Dict[str, Any]) -> pd.DataFrame:
        # 将输入数据转换为模型特征
        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)

        # 为 Claude 准备提示
        prompt = f"""
        根据以下交通状况，预测拥堵程度 (0-10)：
        - 时间: {input_data['hour']}:00
        - 星期几: {input_data['day']}
        - 温度: {input_data['temperature']}°C
        - 降水量: {input_data['precipitation']}mm
        - 特殊事件: {'是' if input_data['special_event'] else '否'}
        - 道路施工: {'是' if input_data['road_work'] else '否'}
        - 车辆数量: {input_data['vehicle_count']}

        只返回数值预测结果。
        """

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

        # 解析响应
        response_body = json.loads(response['body'].read())
        prediction = float(response_body['completion'].strip())

        return np.clip(prediction, 0, 10)</code>

<code class="python">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))</code>