我們都喜歡交通,對吧?這是我唯一一次思考我是如何完全搞砸了我的演示(想太多是一種痛苦)。
拋開所有笑話,我想創建一個項目,可以作為 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 設定
導覽至 AWS 主控台並啟用 AWS 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 基礎架構
建立「基礎設施.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 前端
修改「app.py」以連接到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()
第 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中文網其他相關文章!

本文討論了版本3.10中介紹的Python的新“匹配”語句,該語句與其他語言相同。它增強了代碼的可讀性,並為傳統的if-elif-el提供了性能優勢

Python中的功能註釋將元數據添加到函數中,以進行類型檢查,文檔和IDE支持。它們增強了代碼的可讀性,維護,並且在API開發,數據科學和圖書館創建中至關重要。

本文討論了Python中的單位測試,其好處以及如何有效編寫它們。它突出顯示了諸如UNITSEST和PYTEST之類的工具進行測試。

文章討論了Python的\ _ \ _ Init \ _ \ _()方法和Self在初始化對象屬性中的作用。還涵蓋了其他類方法和繼承對\ _ \ _ Init \ _ \ _()的影響。

本文討論了python中@classmethod,@staticmethod和實例方法之間的差異,詳細介紹了它們的屬性,用例和好處。它說明瞭如何根據所需功能選擇正確的方法類型和DA

Inpython,YouAppendElementStoAlistusingTheAppend()方法。 1)useappend()forsingleelements:my_list.append(4).2)useextend()orextend()或= formultiplelements:my_list.extend.extend(emote_list)ormy_list = [4,5,6] .3)useInsert()forspefificpositions:my_list.insert(1,5).beaware


熱AI工具

Undresser.AI Undress
人工智慧驅動的應用程序,用於創建逼真的裸體照片

AI Clothes Remover
用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool
免費脫衣圖片

Clothoff.io
AI脫衣器

Video Face Swap
使用我們完全免費的人工智慧換臉工具,輕鬆在任何影片中換臉!

熱門文章

熱工具

ZendStudio 13.5.1 Mac
強大的PHP整合開發環境

mPDF
mPDF是一個PHP庫,可以從UTF-8編碼的HTML產生PDF檔案。原作者Ian Back編寫mPDF以從他的網站上「即時」輸出PDF文件,並處理不同的語言。與原始腳本如HTML2FPDF相比,它的速度較慢,並且在使用Unicode字體時產生的檔案較大,但支援CSS樣式等,並進行了大量增強。支援幾乎所有語言,包括RTL(阿拉伯語和希伯來語)和CJK(中日韓)。支援嵌套的區塊級元素(如P、DIV),

SecLists
SecLists是最終安全測試人員的伙伴。它是一個包含各種類型清單的集合,這些清單在安全評估過程中經常使用,而且都在一個地方。 SecLists透過方便地提供安全測試人員可能需要的所有列表,幫助提高安全測試的效率和生產力。清單類型包括使用者名稱、密碼、URL、模糊測試有效載荷、敏感資料模式、Web shell等等。測試人員只需將此儲存庫拉到新的測試機上,他就可以存取所需的每種類型的清單。

SAP NetWeaver Server Adapter for Eclipse
將Eclipse與SAP NetWeaver應用伺服器整合。

MinGW - Minimalist GNU for Windows
這個專案正在遷移到osdn.net/projects/mingw的過程中,你可以繼續在那裡關注我們。 MinGW:GNU編譯器集合(GCC)的本機Windows移植版本,可自由分發的導入函式庫和用於建置本機Windows應用程式的頭檔;包括對MSVC執行時間的擴展,以支援C99功能。 MinGW的所有軟體都可以在64位元Windows平台上運作。