iT邦幫忙

2025 iThome 鐵人賽

DAY 29
0

前言

經過 28 天的學習旅程,我們從 AWS AI 服務的基礎概念,一路實作到複雜的應用場景。
今天,我想整理這段時間累積的經驗,分享在使用 Amazon Bedrock 和 SageMaker 時的最佳實務,
希望能幫助大家少走彎路,更有效率地打造 AI 應用。
[這篇章主要是以總結為主]

架構設計的最佳實務

選擇合適的服務
Bedrock vs SageMaker:何時使用哪個服務?

使用 Bedrock 的情境:

需要快速原型開發和驗證想法
應用場景符合基礎模型的能力範圍
團隊缺乏深度 ML 專業知識
需要多模型比較和切換的靈活性
重視開發速度和 Time-to-Market

使用 SageMaker 的情境:

需要高度客製化的模型
有特定領域的訓練資料
需要完整掌控模型的訓練過程
對模型性能有極致要求
需要處理敏感資料,無法使用第三方模型

實務建議:對於大多數企業應用,我建議採用「混合架構」:

用 Bedrock 處理通用 AI 任務(如對話、摘要、翻譯)
用 SageMaker 訓練專屬的業務模型(如欺詐檢測、需求預測)
透過 API Gateway 統一對外服務介面

架構考量

考量面建立在幾個面向

  1. 架構設計最佳實務 - 服務選擇、成本優化
  2. 安全性最佳實務 - 資料保護、IAM 權限
  3. 效能優化 - Prompt Engineering、批次處理
  4. 監控與可觀測性 - CloudWatch 整合
  5. 開發流程 - 環境管理、版本控制
  6. 常見陷阱與解決方案 - Token 限制、錯誤處理
  7. 團隊協作 - API 文檔、測試策略
  8. 持續改進 - A/B 測試框架

快速回顧

成本優化 (參考過去文章)

Bedrock

  • 使用快取機制
  • 使用便宜的模型
  • 控制 token 的數量

SageMaker

  • 使用 Spot Instances
  • 使用 SageMaker Savings Plans
  • 定期清理未使用的 Endpoints

安全實務

  • 遮罩 Email
  • IAM 權限最小化原則
  • 內容審核 (AWS Guardrail)

效能最佳化 (參考過去文章)

  • Prompt Engineering 技巧
  • 批次處理優化

監控與可觀測性

  • 建立完整的監控系統 CloudWatch
  • 建立 CloudWatch Dashboard

開發流程實務 (參考過去文章)

  • configManager 環境管理
  • 版本控制

陷阱與解決方案

  • Token 限制處理

團隊協作

  • API 文件標準

例如 :

from typing import Dict, List, Optional
from pydantic import BaseModel, Field

class BedrockRequest(BaseModel):
    """Bedrock API 請求模型"""
    prompt: str = Field(..., description="使用者輸入的提示詞")
    model_id: str = Field(
        default="anthropic.claude-3-sonnet-20240229-v1:0",
        description="模型 ID"
    )
    max_tokens: int = Field(default=1024, ge=1, le=4096, description="最大 token 數")
    temperature: float = Field(default=1.0, ge=0, le=1, description="溫度參數")
    
    class Config:
        schema_extra = {
            "example": {
                "prompt": "請解釋什麼是機器學習",
                "model_id": "anthropic.claude-3-sonnet-20240229-v1:0",
                "max_tokens": 1024,
                "temperature": 0.7
            }
        }

class BedrockResponse(BaseModel):
    """Bedrock API 回應模型"""
    content: str = Field(..., description="生成的內容")
    model_id: str = Field(..., description="使用的模型 ID")
    tokens_used: int = Field(..., description="使用的 token 數量")
    latency_ms: float = Field(..., description="回應延遲(毫秒)")
  • 測試策略

例如 pytest

import pytest
from unittest.mock import Mock, patch

class TestBedrockIntegration:
    """Bedrock 整合測試"""
    
    @pytest.fixture
    def mock_bedrock_client(self):
        """Mock Bedrock 客戶端"""
        with patch('boto3.client') as mock_client:
            mock_instance = Mock()
            mock_client.return_value = mock_instance
            
            # 設定 mock 回應
            mock_instance.invoke_model.return_value = {
                'body': Mock(read=lambda: json.dumps({
                    'content': [{'text': '測試回應'}],
                    'usage': {'total_tokens': 100}
                }).encode())
            }
            
            yield mock_instance
    
    def test_invoke_bedrock_success(self, mock_bedrock_client):
        """測試成功調用 Bedrock"""
        result = invoke_bedrock("測試提示詞")
        assert result is not None
        assert 'content' in result
        
    def test_invoke_bedrock_with_cache(self, mock_bedrock_client):
        """測試快取機制"""
        prompt = "相同的提示詞"
        
        # 第一次調用
        result1 = invoke_bedrock_with_cache(prompt)
        
        # 第二次調用(應使用快取)
        result2 = invoke_bedrock_with_cache(prompt)
        
        # 驗證只調用一次 API
        assert mock_bedrock_client.invoke_model.call_count == 1

持續改進

A/B testing

import random

class ABTestingFramework:
    def __init__(self):
        self.dynamodb = boto3.resource('dynamodb')
        self.results_table = self.dynamodb.Table('ab-test-results')
    
    def get_variant(self, user_id, experiment_name):
        """為使用者分配測試變體"""
        # 使用一致性雜湊確保同一使用者總是看到相同變體
        hash_value = hash(f"{user_id}{experiment_name}") % 100
        
        if hash_value < 50:
            return 'A'  # 控制組
        else:
            return 'B'  # 實驗組
    
    def log_result(self, user_id, experiment_name, variant, metric_name, value):
        """記錄測試結果"""
        self.results_table.put_item(Item={
            'experiment_id': f"{experiment_name}_{user_id}",
            'experiment_name': experiment_name,
            'variant': variant,
            'user_id': user_id,
            'metric_name': metric_name,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })

# 使用範例:測試不同的 Prompt 策略
ab_test = ABTestingFramework()

def generate_response(user_id, user_query):
    variant = ab_test.get_variant(user_id, 'prompt_strategy_v1')
    
    if variant == 'A':
        # 控制組:基礎 Prompt
        prompt = f"請回答: {user_query}"
    else:
        # 實驗組:結構化 Prompt
        prompt = f"""作為專業助手,請依以下步驟回答:
1. 理解問題核心
2. 提供詳細解答
3. 給予實用建議

問題:{user_query}"""
    
    start_time = time.time()
    response = invoke_bedrock(prompt)
    latency = time.time() - start_time
    
    # 記錄指標
    ab_test.log_result(user_id, 'prompt_strategy_v1', variant, 'latency', latency)
    
    return response

資源與未來展望

  • Multi-Agent 系統:多個 AI Agent 協作處理複雜任務
  • Fine-tuning:基於業務資料微調專屬模型
  • 邊緣部署:將模型部署到邊緣裝置
  • 多模態整合:結合文字、圖像、語音的應用

And
明天見!


上一篇
部署到生產環境
系列文
從零開始的AWS AI之路:用Bedrock與SageMaker打造智慧應用的30天實戰29
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言