用 Amazon Bedrock 同时优化提示词并完成多模型迁移

python -m venv .venv
source .venv/bin/activate
pip install --upgrade boto3

export AWS_REGION=us-east-1
export BEDROCK_MODEL_IDS='model-id-a,model-id-b'
python compare_prompts.py

import json
import os
import time
from statistics import median

import boto3

REGION = os.getenv("AWS_REGION", "us-east-1")
MODEL_IDS = [
    value.strip()
    for value in os.environ["BEDROCK_MODEL_IDS"].split(",")
    if value.strip()
]

if not 1 <= len(MODEL_IDS) <= 5:
    raise SystemExit("BEDROCK_MODEL_IDS must contain between 1 and 5 model IDs")

SYSTEM_PROMPT = """You classify support requests.
Return JSON only, using this schema:
{"category": "billing|technical|account", "urgent": true|false}
Do not add fields or explanatory text.
"""

SAMPLES = [
    "I was charged twice for the same subscription.",
    "Production API calls started returning 503 errors.",
    "I cannot reset my password and need access today.",
]

client = boto3.client("bedrock-runtime", region_name=REGION)
results = []

for model_id in MODEL_IDS:
    latencies = []
    input_tokens = 0
    output_tokens = 0

    for sample in SAMPLES:
        started = time.perf_counter()
        response = client.converse(
            modelId=model_id,
            system=[{"text": SYSTEM_PROMPT}],
            messages=[
                {
                    "role": "user",
                    "content": [{"text": sample}],
                }
            ],
            inferenceConfig={"temperature": 0, "maxTokens": 100},
        )
        elapsed_ms = (time.perf_counter() - started) * 1000
        usage = response.get("usage", {})
        text = response["output"]["message"]["content"][0]["text"]

        try:
            parsed = json.loads(text)
            format_valid = set(parsed) == {"category", "urgent"}
        except (json.JSONDecodeError, TypeError):
            format_valid = False

        latencies.append(elapsed_ms)
        input_tokens += usage.get("inputTokens", 0)
        output_tokens += usage.get("outputTokens", 0)
        results.append(
            {
                "model_id": model_id,
                "input": sample,
                "output": text,
                "format_valid": format_valid,
                "latency_ms": round(elapsed_ms, 1),
            }
        )

    print(
        json.dumps(
            {
                "model_id": model_id,
                "median_latency_ms": round(median(latencies), 1),
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
            },
            ensure_ascii=False,
        )
    )

with open("bedrock-comparison.json", "w", encoding="utf-8") as output_file:
    json.dump(results, output_file, ensure_ascii=False, indent=2)