graph TD
    A[1. 実行環境<br>(Firecracker VM)の作成] --> B[2. ランタイムの起動]
    B --> C[3. 関数コードの<br>ダウンロード・展開]
    C --> D[4. グローバル変数・<br>import文の実行]
    style A fill:#f9f,stroke:#333,stroke-width:2px;
    style B fill:#bbf,stroke:#333,stroke-width:2px;
    style C fill:#dbf,stroke:#333,stroke-width:2px;
    style D fill:#fff,stroke:#333,stroke-width:2px;
    linkStyle 0 stroke-width:2px,fill:none,stroke:gray;
    linkStyle 1 stroke-width:2px,fill:none,stroke:gray;
    linkStyle 2 stroke-width:2px,fill:none,stroke:gray;

graph TD
    E[1. ハンドラー関数の実行] --> F[2. ビジネスロジックの処理]
    F --> G[3. レスポンスの返却]
    style E fill:#f9f,stroke:#333,stroke-width:2px;
    style F fill:#bbf,stroke:#333,stroke-width:2px;
    style G fill:#dbf,stroke:#333,stroke-width:2px;
    linkStyle 0 stroke-width:2px,fill:none,stroke:gray;
    linkStyle 1 stroke-width:2px,fill:none,stroke:gray;

# ❌ 改善前：不要なライブラリを大量インポート
import pandas as pd              # 100MB
import numpy as np              # 50MB
import matplotlib.pyplot as plt # 80MB
import seaborn as sns          # 30MB
import scipy                   # 40MB
# 合計パッケージサイズ: 300MB
# コールドスタート: 12秒
def lambda_handler(event, context):
# 実際にはpandasの基本機能しか使わない
df = pd.DataFrame(event['data'])
return df.to_dict()

# ✅ 改善後：必要最小限のライブラリのみ
import json
from typing import Dict, List, Any
# pandas の代わりに標準ライブラリで実装
def process_data(data: List[Dict]) -> Dict:
    """軽量なデータ処理関数"""
result = {}
for item in data:
key = item.get('category', 'unknown')
if key not in result:
result[key] = []
result[key].append(item)
return result
def lambda_handler(event, context):
# 合計パッケージサイズ: 5MB
# コールドスタート: 0.8秒
processed_data = process_data(event['data'])
return {
'statusCode': 200,
'body': json.dumps(processed_data)
}

import boto3
import json
from datetime import datetime, timedelta
def analyze_lambda_performance(function_name: str, days: int = 7):
    """Lambda関数のパフォーマンス分析"""
cloudwatch = boto3.client('cloudwatch')
# メトリクス取得
metrics = [
'Duration',
'InitDuration', 
'MemoryUtilization',
'Invocations'
]
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=days)
performance_data = {}
for metric in metrics:
response = cloudwatch.get_metric_statistics(
Namespace='AWS/Lambda',
MetricName=metric,
Dimensions=[
{
'Name': 'FunctionName',
'Value': function_name
}
],
StartTime=start_time,
EndTime=end_time,
Period=3600,  # 1時間
Statistics=['Average', 'Maximum']
)
performance_data[metric] = response['Datapoints']
# 最適メモリサイズの推奨
avg_memory_util = sum(
dp['Average'] for dp in performance_data['MemoryUtilization']
) / len(performance_data['MemoryUtilization'])
current_memory = get_function_memory(function_name)
optimal_memory = calculate_optimal_memory(avg_memory_util, current_memory)
return {
'current_memory': current_memory,
'memory_utilization': avg_memory_util,
'recommended_memory': optimal_memory,
'expected_cost_reduction': calculate_cost_reduction(
current_memory, optimal_memory
)
}
def calculate_optimal_memory(utilization: float, current_memory: int) -> int:
    """最適なメモリサイズを計算"""
# 使用率が80%以上の場合はメモリ増加を推奨
if utilization > 80:
return min(current_memory * 2, 3008)
# 使用率が30%以下の場合はメモリ削減を推奨
elif utilization < 30:
return max(current_memory // 2, 128)
# 適切な範囲内の場合は現状維持
else:
return current_memory

# VPC内Lambda関数のコールドスタート分析
vpc_performance = {
"VPC設定あり": {
"コールドスタート時間": "15-45秒",
"原因": "ENI（Elastic Network Interface）の作成・アタッチ",
"対策": "Provisioned Concurrency、VPC設定の見直し"
},
"VPC設定なし": {
"コールドスタート時間": "0.5-3秒", 
"制限": "VPCリソースへの直接アクセス不可",
"対策": "NAT Gateway、VPC Endpoint経由でのアクセス"
}
}

# ❌ 改善前：VPC内でRDS直接接続
import pymysql
def lambda_handler(event, context):
# VPC内のRDSに直接接続（コールドスタート: 30秒）
connection = pymysql.connect(
host='rds-instance.cluster-xxx.ap-northeast-1.rds.amazonaws.com',
user='admin',
password='password',
database='mydb'
)
# データベース処理
return result
# ✅ 改善後：RDS Proxy + 接続プール
import boto3
import json
# グローバル変数で接続を再利用
rds_client = None
def get_rds_client():
global rds_client
if rds_client is None:
rds_client = boto3.client('rds-data')
return rds_client
def lambda_handler(event, context):
# RDS Proxy経由でアクセス（コールドスタート: 2秒）
client = get_rds_client()
response = client.execute_statement(
resourceArn='arn:aws:rds:region:account:cluster:cluster-name',
secretArn='arn:aws:secretsmanager:region:account:cluster:secret-name',
database='mydb',
sql='SELECT * FROM users WHERE id = :id',
parameters=[
{
'name': 'id',
'value': {'longValue': event['user_id']}
}
]
)
return {
'statusCode': 200,
'body': json.dumps(response['records'])
}

# ❌ 改善前：同期的な逐次リクエスト
import requests
import time
def lambda_handler(event, context):
urls = event.get('urls', [])
results = []
start_time = time.time()
# APIに1つずつ逐次リクエスト（実行時間: 10秒）
for url in urls:
response = requests.get(url)
results.append(response.json())
execution_time = time.time() - start_time
print(f"Execution time: {execution_time:.2f} seconds")
return results
# ✅ 改善後：非同期I/Oによる並列リクエスト
import asyncio
import aiohttp
import time
from typing import List, Dict
async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict:
    """非同期でURLからデータを取得する"""
try:
async with session.get(url, timeout=10) as response:
response.raise_for_status()
return await response.json()
except Exception as e:
print(f"Failed to fetch {url}: {e}")
return {'url': url, 'error': str(e)}
async def fetch_all_urls(urls: List[str]) -> List[Dict]:
    """aiohttpを使ってすべてのURLに並列でリクエストを送信する"""
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
def lambda_handler(event, context):
urls = event.get('urls', [])
start_time = time.time()
# 非同期処理を実行（実行時間: 2秒 → 80%短縮）
results = asyncio.run(fetch_all_urls(urls))
execution_time = time.time() - start_time
print(f"Execution time: {execution_time:.2f} seconds")
return {
'statusCode': 200,
'body': json.dumps({
'results': results,
'execution_time_seconds': execution_time
})
}

# ❌ 改善前：毎回新しい接続を作成
import psycopg2
def lambda_handler(event, context):
# 毎回新しい接続（接続時間: 2秒）
conn = psycopg2.connect(
host='postgres-instance.xxx.rds.amazonaws.com',
database='mydb',
user='admin',
password='password'
)
cursor = conn.cursor()
# N+1問題のあるクエリ
user_ids = event['user_ids']
results = []
for user_id in user_ids:  # 100回のクエリ実行
cursor.execute(
"SELECT * FROM users WHERE id = %s", (user_id,)
)
user = cursor.fetchone()
cursor.execute(
"SELECT * FROM orders WHERE user_id = %s", (user_id,)
)
orders = cursor.fetchall()
results.append({
'user': user,
'orders': orders
})
conn.close()
return results
# ✅ 改善後：接続プール + バッチクエリ
import psycopg2.pool
from typing import List, Dict, Any
# グローバル接続プール
connection_pool = None
def get_connection_pool():
global connection_pool
if connection_pool is None:
connection_pool = psycopg2.pool.ThreadedConnectionPool(
minconn=1,
maxconn=5,
host='postgres-instance.xxx.rds.amazonaws.com',
database='mydb',
user='admin',
password='password'
)
return connection_pool
def fetch_users_with_orders(user_ids: List[int]) -> List[Dict[str, Any]]:
    """最適化されたバッチクエリ"""
pool = get_connection_pool()
conn = pool.getconn()
try:
cursor = conn.cursor()
# 1回のクエリで全ユーザー情報を取得
cursor.execute("""
            SELECT u.id, u.name, u.email,
                   o.id as order_id, o.amount, o.created_at
            FROM users u
            LEFT JOIN orders o ON u.id = o.user_id
            WHERE u.id = ANY(%s)
            ORDER BY u.id, o.created_at DESC
        """, (user_ids,))
# 結果をグループ化
results = {}
for row in cursor.fetchall():
user_id = row[0]
if user_id not in results:
results[user_id] = {
'user': {
'id': row[0],
'name': row[1], 
'email': row[2]
},
'orders': []
}
if row[3]:  # order_id が存在する場合
results[user_id]['orders'].append({
'id': row[3],
'amount': row[4],
'created_at': row[5]
})
return list(results.values())
finally:
pool.putconn(conn)
def lambda_handler(event, context):
# 実行時間: 15秒 → 3秒（80%短縮）
return fetch_users_with_orders(event['user_ids'])

# ❌ 改善前：アイテムを1つずつ書き込み
import boto3
def lambda_handler(event, context):
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('my-table')
# ループ内でput_itemを繰り返し呼び出し（非効率）
for item in event['items']:
table.put_item(Item=item)
return {'message': 'success'}
# ✅ 改善後：クライアント再利用 + バッチ書き込み
import boto3
import json
from typing import List, Dict
# グローバルスコープでクライアントを初期化し、実行環境全体で再利用
dynamodb_client = boto3.resource('dynamodb')
def batch_write_to_dynamodb(table_name: str, items: List[Dict]):
    """
    DynamoDBのbatch_writerを使用して効率的に書き込む。
    25件ずつのバッチ処理はSDKが自動でハンドリングしてくれる。
    """
table = dynamodb_client.Table(table_name)
with table.batch_writer() as batch:
for item in items:
batch.put_item(Item=item)
def lambda_handler(event, context):
items_to_write = event.get('items', [])
batch_write_to_dynamodb('my-table', items_to_write)
return {
'statusCode': 200,
'body': json.dumps({'message': f'{len(items_to_write)} items processed.'})
}

import boto3
import os
import time
import numpy as np
from typing import Dict, Any
# 改善①：モデルは /tmp ディレクトリに遅延ロード＆キャッシュ
#          Lambda LayerやEFSにモデルを配置し、初回呼び出し時に/tmpにコピーする
s3_client = boto3.client('s3')
model = None
MODEL_PATH = '/tmp/model.joblib'
MODEL_BUCKET = os.environ.get('MODEL_BUCKET')
MODEL_KEY = os.environ.get('MODEL_KEY')
def load_model():
    """/tmpにモデルが存在しない場合のみS3からダウンロード"""
global model
if model is None:
if not os.path.exists(MODEL_PATH):
print("Model not found in /tmp, downloading from S3...")
s3_client.download_file(MODEL_BUCKET, MODEL_KEY, MODEL_PATH)
# ここで実際のモデルロード処理を行う（例：joblib.load）
# model = joblib.load(MODEL_PATH)
# ダミーのモデルとしておく
model = lambda x: np.mean(x) * 10
print("Model loaded.")
return model
def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """最適化されたAI推論Lambda関数"""
import json
start_time = time.time()
# 改善②：大きなペイロードはS3の署名付きURL経由で受け渡し
# API GatewayにはS3のオブジェクトキーのみを渡す
if 's3_presigned_url' in event:
# 署名付きURLから直接データをダウンロード（Lambdaのタイムアウトに注意）
# import requests
# response = requests.get(event['s3_presigned_url'])
# input_data = np.frombuffer(response.content)
input_data = np.array([1,2,3,4,5]) # ダミーデータ
else:
# 小さなペイロードは直接渡す
input_data = np.array(event['data'])
# モデルのロード（初回のみ時間がかかる）
ml_model = load_model()
# 改善③：推論実行（CPU集約的）
# Graviton2(arm64)アーキテクチャを選択することで、x86に比べてコストパフォーマンスが向上
prediction = ml_model(input_data)
end_time = time.time()
return {
"statusCode": 200,
"body": json.dumps({
"prediction": prediction,
"execution_time_ms": (end_time - start_time) * 1000
})
}

# template.yaml
Resources:
  MyPythonFunction:
    Type: AWS::Serverless::Function
    Properties:
      # ... (関数プロパティ) ...
      Runtime: python3.12
      SnapStart:
        ApplyOn: PublishedVersions # 'PublishedVersions' または 'None'
      AutoPublishAlias: live

import boto3
from datetime import datetime, time
def configure_provisioned_concurrency():
    """時間帯に応じたProvisioned Concurrency設定"""
lambda_client = boto3.client('lambda')
# 営業時間（9:00-18:00）の設定
business_hours_config = {
'FunctionName': 'my-api-function',
'ProvisionedConcurrencyConfig': {
'ProvisionedConcurrency': 10  # 10並列を常時待機
}
}
# 夜間・休日の設定
off_hours_config = {
'FunctionName': 'my-api-function', 
'ProvisionedConcurrencyConfig': {
'ProvisionedConcurrency': 2   # 2並列のみ待機
}
}
current_time = datetime.now().time()
business_start = time(9, 0)   # 9:00
business_end = time(18, 0)    # 18:00
if business_start <= current_time <= business_end:
config = business_hours_config
else:
config = off_hours_config
# 設定を適用
lambda_client.put_provisioned_concurrency_config(**config)
return {
'applied_config': config,
'cost_optimization': 'Time-based scaling applied'
}

# template.yaml (AWS SAM)
Resources:
  MyApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: my-api-function
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.11
      MemorySize: 1024
      Timeout: 30
      ProvisionedConcurrencyConfig:
        ProvisionedConcurrentExecutions: 10 # 常に10個の実行環境をウォーム状態に保つ
      AutoPublishAlias: live
      DeploymentPreference:
        Type: Canary10Percent5Minutes # カナリアリリースで安全にデプロイ

# Lambda Extension for custom metrics
import json
import requests
import os
from typing import Dict, Any
class PerformanceExtension:
    """カスタムパフォーマンスメトリクス収集"""
def __init__(self):
self.metrics_endpoint = os.environ.get('METRICS_ENDPOINT')
self.function_name = os.environ.get('AWS_LAMBDA_FUNCTION_NAME')
def collect_metrics(self, execution_data: Dict[str, Any]):
        """実行メトリクスの収集"""
metrics = {
'function_name': self.function_name,
'timestamp': execution_data['timestamp'],
'duration': execution_data['duration'],
'memory_used': execution_data['memory_used'],
'cold_start': execution_data.get('cold_start', False),
'error_count': execution_data.get('error_count', 0)
}
# 非同期でメトリクス送信
try:
requests.post(
self.metrics_endpoint,
json=metrics,
timeout=1  # タイムアウト短縮
)
except requests.RequestException:
# メトリクス送信失敗は無視（本処理に影響させない）
pass
def optimize_based_on_metrics(self) -> Dict[str, str]:
        """メトリクスに基づく最適化提案"""
# 過去のメトリクスを分析
historical_data = self.fetch_historical_metrics()
recommendations = []
# コールドスタート頻度が高い場合
if historical_data['cold_start_rate'] > 0.3:
recommendations.append(
"Provisioned Concurrency の設定を検討してください"
)
# メモリ使用率が低い場合
if historical_data['avg_memory_utilization'] < 0.3:
recommendations.append(
f"メモリ設定を {historical_data['current_memory'] // 2}MB に削減可能"
)
# 実行時間が長い場合
if historical_data['avg_duration'] > 5000:  # 5秒以上
recommendations.append(
"並列処理またはアルゴリズムの最適化を検討してください"
)
return {
'recommendations': recommendations,
'potential_cost_savings': self.calculate_cost_savings(historical_data)
}

import functools
import time
import random
from typing import Callable, Any
def exponential_backoff_retry(
max_retries: int = 3,
base_delay: float = 0.5, # 秒
max_delay: float = 10.0, # 秒
exceptions: tuple = (Exception,)
):
    """
    指数バックオフとジッターによるリトライを実装したデコレータ。
    Lambdaの実行時間制限を考慮し、最大遅延を設定することが重要。
    """
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_retries:
print(f"Max retries ({max_retries}) exceeded for {func.__name__}.")
raise
# 指数バックオフ + ジッター
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(0, delay * 0.2) # ジッターを少し加える
total_delay = delay + jitter
# Lambdaの残り実行時間を確認
context = args[1] if len(args) > 1 and hasattr(args[1], 'get_remaining_time_in_millis') else None
if context and context.get_remaining_time_in_millis() < total_delay * 1000 + 500:
print(f"Not enough time for another retry. Aborting.")
raise last_exception
print(f"Retry {attempt + 1}/{max_retries} for {func.__name__} after {total_delay:.2f}s delay. Error: {e}")
time.sleep(total_delay)
raise last_exception
return wrapper
return decorator
# --- 使用例 ---
import requests
class ApiCallError(Exception):
pass
@exponential_backoff_retry(max_retries=4, exceptions=(ApiCallError,))
def call_flaky_api(event: dict, context: object) -> dict:
    """不安定な外部APIを呼び出すサンプル関数"""
try:
response = requests.get("https://api.example.com/data", timeout=5)
if response.status_code >= 500:
raise ApiCallError(f"Server error: {response.status_code}")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
raise ApiCallError(f"Request failed: {e}") from e
def lambda_handler(event, context):
    """リトライ戦略を組み込んだLambda関数"""
import json
try:
api_data = call_flaky_api(event, context)
return {
'statusCode': 200,
'body': json.dumps(api_data)
}
except ApiCallError as e:
print(f"API call failed after retries: {e}")
return {
'statusCode': 502,
'body': json.dumps({'error': 'External API is unavailable.'})
}
except Exception as e:
print(f"An unexpected error occurred: {e}")
return {
'statusCode': 500,
'body': json.dumps({'error': 'An internal error occurred.'})
}

# template.yaml
Resources:
  MyArmFunction:
    Type: AWS::Serverless::Function
    Properties:
      # ... (関数プロパティ) ...
      Architectures:
        - arm64 # 'x86_64' から 'arm64' に変更

-- Lambda関数のパフォーマンス分析クエリ
fields @timestamp, @duration, @memorySize, @maxMemoryUsed, @initDuration
| filter @type = "REPORT"
| stats 
    avg(@duration) as avg_duration,
    max(@duration) as max_duration,
    avg(@maxMemoryUsed) as avg_memory_used,
    avg(@initDuration) as avg_cold_start,
    count() as total_invocations,
    sum(@initDuration > 0) as cold_starts
by bin(5m)
| sort @timestamp desc

import boto3
def setup_performance_alarms():
    """パフォーマンス監視アラームの設定"""
cloudwatch = boto3.client('cloudwatch')
# コールドスタート率アラーム
cloudwatch.put_metric_alarm(
AlarmName='Lambda-HighColdStartRate',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=2,
MetricName='ColdStartRate',
Namespace='AWS/Lambda',
Period=300,
Statistic='Average',
Threshold=30.0,  # 30%以上でアラート
ActionsEnabled=True,
AlarmActions=[
'arn:aws:sns:region:account:lambda-performance-alerts'
],
AlarmDescription='Lambda cold start rate is too high'
)
# 実行時間アラーム
cloudwatch.put_metric_alarm(
AlarmName='Lambda-HighDuration',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=3,
MetricName='Duration',
Namespace='AWS/Lambda',
Period=300,
Statistic='Average',
Threshold=5000.0,  # 5秒以上でアラート
ActionsEnabled=True,
AlarmActions=[
'arn:aws:sns:region:account:lambda-performance-alerts'
],
AlarmDescription='Lambda execution time is too high'
)

import functools
import time
import psutil
import boto3
import json
from typing import Callable, Any
class LambdaPerformanceMonitor:
    """Lambda パフォーマンス監視クラス"""
def __init__(self):
self.cloudwatch = boto3.client('cloudwatch')
def put_custom_metrics(self, function_name: str, metrics: dict):
        """カスタムメトリクスをCloudWatchに送信"""
try:
self.cloudwatch.put_metric_data(
Namespace='Lambda/Performance',
MetricData=[
{
'MetricName': 'ExecutionTime',
'Value': metrics['execution_time'],
'Unit': 'Milliseconds',
'Dimensions': [{'Name': 'FunctionName', 'Value': function_name}]
},
{
'MetricName': 'MemoryUtilization',
'Value': metrics['memory_utilization'],
'Unit': 'Percent',
'Dimensions': [{'Name': 'FunctionName', 'Value': function_name}]
}
]
)
except Exception as e:
print(f"Failed to put custom metrics: {e}")
def performance_decorator(func: Callable) -> Callable:
    """パフォーマンス測定とカスタムメトリクス送信を行うデコレータ"""
@functools.wraps(func)
def wrapper(event: dict, context: Any) -> Any:
start_time = time.time()
# psutilを使う場合はLambda Layerにパッケージを含める必要があります
process = psutil.Process()
start_memory = process.memory_info().rss
try:
result = func(event, context)
return result
finally:
# パフォーマンス測定
execution_time = (time.time() - start_time) * 1000  # ms
end_memory = process.memory_info().rss
memory_used = (end_memory - start_memory) / (1024 * 1024)  # MB
# contextオブジェクトからメモリ上限を取得
memory_limit = context.memory_limit_in_mb
memory_utilization = (memory_used / memory_limit) * 100 if memory_limit > 0 else 0
# メトリクス送信
monitor = LambdaPerformanceMonitor()
monitor.put_custom_metrics(context.function_name, {
'execution_time': execution_time,
'memory_utilization': memory_utilization
})
# 構造化ログとしてパフォーマンス情報を出力
print(json.dumps({
'performance_metrics': {
'execution_time_ms': round(execution_time, 2),
'memory_used_mb': round(memory_used, 2),
'memory_utilization_percent': round(memory_utilization, 2)
}
}))
return wrapper
@performance_decorator
def lambda_handler(event, context):
    """パフォーマンス監視デコレータを適用したLambda関数"""
# ここに本来のビジネスロジックを記述
time.sleep(1) # ダミーの処理
return {'statusCode': 200, 'body': '"Success"'}

import random
from typing import Dict, Any
class LambdaPerformanceABTest:
    """Lambda関数のA/Bテスト実装"""
def __init__(self):
self.function_name = function_name
self.test_configs = {
'A': {  # 現在の設定
'memory_size': 512,
'timeout': 30,
'environment': {'OPTIMIZATION_LEVEL': 'standard'}
},
'B': {  # 最適化版
'memory_size': 1024,
'timeout': 15,
'environment': {'OPTIMIZATION_LEVEL': 'aggressive'}
}
}
def route_request(self, event: Dict[str, Any]) -> str:
        """リクエストをA/Bグループに振り分け"""
# ユーザーIDベースの一貫した振り分け
user_id = event.get('user_id', 'anonymous')
hash_value = hash(user_id) % 100
# 50:50 の振り分け
return 'A' if hash_value < 50 else 'B'
def execute_with_config(self, event: Dict[str, Any], config_name: str):
        """指定された設定で関数を実行"""
config = self.test_configs[config_name]
# 環境変数を設定
os.environ.update(config['environment'])
# パフォーマンス測定開始
start_time = time.time()
try:
# ビジネスロジック実行
result = self.execute_business_logic(event)
# メトリクス記録
execution_time = time.time() - start_time
self.record_ab_test_metrics(config_name, execution_time, True)
return result
except Exception as e:
execution_time = time.time() - start_time
self.record_ab_test_metrics(config_name, execution_time, False)
raise
def record_ab_test_metrics(self, config: str, duration: float, success: bool):
        """A/Bテストメトリクスの記録"""
cloudwatch = boto3.client('cloudwatch')
cloudwatch.put_metric_data(
Namespace='Lambda/ABTest',
MetricData=[
{
'MetricName': 'Duration',
'Dimensions': [
{'Name': 'Config', 'Value': config},
{'Name': 'FunctionName', 'Value': self.function_name}
],
'Value': duration * 1000,  # ミリ秒
'Unit': 'Milliseconds'
},
{
'MetricName': 'Success',
'Dimensions': [
{'Name': 'Config', 'Value': config},
{'Name': 'FunctionName', 'Value': self.function_name}
],
'Value': 1 if success else 0,
'Unit': 'Count'
}
]
)

{
  "proposal_summary": {
    "project_name": "AWS Lambda パフォーマンス最適化コンサルティング",
    "client_name": "株式会社Example",
    "objective": "月額クラウドコストの30%削減と、主要APIの応答時間50%短縮"
  },
  "current_state_analysis": {
    "monthly_lambda_invocations": "1,000万回",
    "current_monthly_cost": "$2,500",
    "avg_api_response_time": "2.8秒",
    "p95_response_time": "5.2秒",
    "error_rate": "0.5%"
  },
  "proposed_improvements": {
    "cost_reduction_target": "30%削減（月額約$750の節約）",
    "performance_improvement_target": "平均応答時間1.4秒以下",
    "availability_improvement": "エラー率0.1%以下への低減",
    "estimated_duration": "3週間"
  },
  "investment_and_roi": {
    "consulting_fee": "$8,000（一括）",
    "projected_annual_savings": "$9,000",
    "first_year_roi": "112.5%",
    "additional_value": [
      "ユーザー体験の向上による離脱率低下",
      "システムの信頼性向上による機会損失の削減",
      "将来的なスケーラビリティの確保"
    ]
  }
}

### プロジェクト：主要サービスのAPIパフォーマンス最適化
- **期間**: 2025年1月～2025年3月（3ヶ月）
- **役割**: リードエンジニア
- **課題**: ユーザー数の増加に伴うAPI応答時間の悪化と、Lambdaコストの急増。
- **実施内容**:
  - CloudWatchとX-Rayによるボトルネック分析を実施。
  - パッケージサイズ最適化（250MB → 15MB）、依存関係の見直し。
  - I/Oバウンド処理に非同期`asyncio`を導入し、外部API呼び出しを並列化。
  - `AWS Lambda Power Tuning`を活用し、メモリ設定を科学的に最適化（1024MB → 512MB）。
  - Provisioned Concurrencyを導入し、主要エンドポイントのコールドスタートを撲滅。
- **達成した成果**:
  - **コスト削減**: Lambda関連コストを月額$2,500から$1,500へ、**40%削減**を達成。
  - **パフォーマンス向上**: APIのP95応答時間を5.2秒から0.8秒へ、**84%短縮**。
  - **信頼性向上**: エラー率を0.5%から0.05%へ低減し、サービスの安定性を大幅に向上。
  - **ビジネスインパクト**: ユーザー満足度の向上と、年間約$12,000のコスト削減に直接貢献した。