用并发扫描为 Amazon SageMaker AI 生成式模型端点精准定容

实例数 = ceil(目标请求速率 / 单实例可持续请求速率 / 目标利用率)

ceil(15 / 3 / 0.7) = 8 个实例

aws --version

aws sagemaker create-ai-benchmark-job \
  --generate-cli-skeleton input \
  > benchmark-job.json

aws sagemaker create-ai-benchmark-job \
  --cli-input-json file://benchmark-job.json

aws sagemaker describe-ai-benchmark-job \
  --job-name YOUR_BENCHMARK_JOB_NAME

import json
import math
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import boto3
from botocore.config import Config

ENDPOINT_NAME = os.environ["ENDPOINT_NAME"]
REGION = os.getenv("AWS_REGION", "us-east-1")
CONCURRENCY_LEVELS = [
    int(value) for value in os.getenv("CONCURRENCY", "1,2,4,8,16").split(",")
]
REQUESTS_PER_LEVEL = int(os.getenv("REQUESTS_PER_LEVEL", "40"))
WARMUP_REQUESTS = int(os.getenv("WARMUP_REQUESTS", "5"))

payload = json.loads(
    os.getenv(
        "PAYLOAD_JSON",
        json.dumps({
            "inputs": "Explain why concurrency testing matters for model serving.",
            "parameters": {"max_new_tokens": 128, "do_sample": False},
        }),
    )
)
body = json.dumps(payload).encode("utf-8")

runtime = boto3.client(
    "sagemaker-runtime",
    region_name=REGION,
    config=Config(
        max_pool_connections=max(CONCURRENCY_LEVELS) + 4,
        retries={"max_attempts": 1, "mode": "standard"},
    ),
)


def invoke_once():
    started = time.perf_counter()
    try:
        response = runtime.invoke_endpoint(
            EndpointName=ENDPOINT_NAME,
            ContentType="application/json",
            Accept="application/json",
            Body=body,
        )
        response["Body"].read()
        return {"ok": True, "latency": time.perf_counter() - started}
    except Exception as exc:
        return {
            "ok": False,
            "latency": time.perf_counter() - started,
            "error": type(exc).__name__,
        }


def percentile(values, p):
    if not values:
        return None
    ordered = sorted(values)
    index = max(0, math.ceil((p / 100) * len(ordered)) - 1)
    return ordered[index]


print(f"Warming up endpoint with {WARMUP_REQUESTS} requests...")
for _ in range(WARMUP_REQUESTS):
    invoke_once()

for concurrency in CONCURRENCY_LEVELS:
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=concurrency) as pool:
        futures = [pool.submit(invoke_once) for _ in range(REQUESTS_PER_LEVEL)]
        results = [future.result() for future in as_completed(futures)]
    elapsed = time.perf_counter() - started

    successful = [item["latency"] for item in results if item["ok"]]
    errors = len(results) - len(successful)

    summary = {
        "concurrency": concurrency,
        "requests": len(results),
        "successful_requests": len(successful),
        "elapsed_seconds": round(elapsed, 3),
        "throughput_rps": round(len(successful) / elapsed, 3),
        "mean_latency_seconds": round(sum(successful) / len(successful), 3)
        if successful else None,
        "p50_latency_seconds": round(percentile(successful, 50), 3)
        if successful else None,
        "p95_latency_seconds": round(percentile(successful, 95), 3)
        if successful else None,
        "p99_latency_seconds": round(percentile(successful, 99), 3)
        if successful else None,
        "error_rate": round(errors / len(results), 4),
    }
    print(json.dumps(summary, ensure_ascii=False))

python -m pip install --upgrade boto3

export AWS_REGION=us-east-1
export ENDPOINT_NAME=my-generative-ai-endpoint
export CONCURRENCY=1,2,4,8,16
export REQUESTS_PER_LEVEL=50
export WARMUP_REQUESTS=5

python sweep.py | tee sweep-results.jsonl

export PAYLOAD_JSON='{"prompt":"Write a short capacity planning checklist.","max_tokens":128}'
python sweep.py