Bedrock の tool use で AI 駆動 A/B テストの意思決定エンジンを検証する

import json
 
TOOL_DEFINITIONS = [
    {
        "toolSpec": {
            "name": "get_user_profile",
            "description": "Get user behavioral profile and preferences.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "user_id": {"type": "string", "description": "The user ID"}
                    },
                    "required": ["user_id"],
                }
            },
        }
    },
    {
        "toolSpec": {
            "name": "get_similar_users",
            "description": "Find users with similar behavior patterns.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "user_id": {"type": "string"},
                        "limit": {"type": "integer", "default": 10},
                    },
                    "required": ["user_id"],
                }
            },
        }
    },
    {
        "toolSpec": {
            "name": "get_variant_performance",
            "description": "Get real-time variant performance metrics.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "experiment_id": {"type": "string"},
                        "variant_id": {"type": "string"},
                    },
                    "required": ["experiment_id", "variant_id"],
                }
            },
        }
    },
]

USER_PROFILES = {
    "user_001": {  # ロイヤルティメンバー
        "user_id": "user_001",
        "engagement_score": 0.89, "conversion_likelihood": 0.24,
        "interaction_style": "focused", "attention_span": "short",
        "successful_variants": ["A", "simple_design"],
        "confidence_score": 0.87, "device_type": "mobile",
        "visit_frequency": "frequent",
        "similarity_cluster": "premium_mobile_focused",
    },
    "user_002": {  # 新規ユーザー（クーポンサイト経由）
        "user_id": "user_002",
        "engagement_score": 0.15, "conversion_likelihood": 0.05,
        "interaction_style": "explorer", "attention_span": "medium",
        "successful_variants": [],
        "confidence_score": 0.12, "device_type": "mobile",
        "visit_frequency": "first_visit",
        "similarity_cluster": "new_deal_seeker",
    },
    "user_003": {  # 矛盾シグナル用
        "user_id": "user_003",
        "engagement_score": 0.62, "conversion_likelihood": 0.18,
        "interaction_style": "explorer", "attention_span": "long",
        "successful_variants": ["B", "social_proof"],
        "confidence_score": 0.71, "device_type": "desktop",
        "visit_frequency": "regular",
        "similarity_cluster": "social_proof_responsive",
    },
}
 
SIMILAR_USERS = {
    "user_001": {
        "count": 52, "avg_conversion_rate": 0.21,
        "preferred_variants": ["A"],
        "shared_characteristics": ["mobile", "loyalty_member", "focused_buyer"],
    },
    "user_002": {
        "count": 39, "avg_conversion_rate": 0.18,
        "preferred_variants": ["B"],
        "shared_characteristics": ["first_visit", "coupon_site_referrer", "deal_seeking"],
        "note": "Similar new users from deal sites show 2.3x higher conversion with incentive messaging",
    },
    "user_003": {
        "count": 34, "avg_conversion_rate": 0.22,
        "preferred_variants": ["B"],
        "shared_characteristics": ["desktop", "social_proof_responsive", "explorer"],
        "note": "Similar users show 34% higher conversion with social proof emphasis",
    },
}
 
VARIANT_PERFORMANCE = {
    "A": {
        "current_performance": {
            "impressions": 3900, "clicks": 312, "conversions": 125,
            "conversion_rate": 0.032, "confidence": 0.89,
        },
        "has_performance_data": True,
    },
    "B": {
        "current_performance": {
            "impressions": 3850, "clicks": 385, "conversions": 158,
            "conversion_rate": 0.041, "confidence": 0.95,
        },
        "has_performance_data": True,
    },
}
 
# 検証 3 用: Variant A の全体コンバージョン率が高いデータセット
VARIANT_PERFORMANCE_CONFLICTING = {
    "A": {
        "current_performance": {
            "impressions": 5000, "clicks": 400, "conversions": 210,
            "conversion_rate": 0.042, "confidence": 0.92,
        },
        "has_performance_data": True,
    },
    "B": {
        "current_performance": {
            "impressions": 4800, "clicks": 370, "conversions": 182,
            "conversion_rate": 0.038, "confidence": 0.90,
        },
        "has_performance_data": True,
    },
}
 
def execute_tool(tool_name, tool_input, use_conflicting=False):
    if tool_name == "get_user_profile":
        data = USER_PROFILES.get(tool_input["user_id"], {"error": "Not found"})
    elif tool_name == "get_similar_users":
        data = SIMILAR_USERS.get(tool_input["user_id"], {"count": 0})
    elif tool_name == "get_variant_performance":
        perf = VARIANT_PERFORMANCE_CONFLICTING if use_conflicting \
            else VARIANT_PERFORMANCE
        data = perf.get(tool_input.get("variant_id", ""), {"error": "Not found"})
    else:
        data = {"error": f"Unknown tool: {tool_name}"}
    return json.dumps(data)

import json, time, boto3
from tools import TOOL_DEFINITIONS, execute_tool
 
client = boto3.client("bedrock-runtime", region_name="us-east-1")
MODEL_ID = "global.anthropic.claude-sonnet-4-20250514-v1:0"
 
SYSTEM_PROMPT = """You are an expert A/B testing optimization specialist \
with access to tools for gathering user behavior data.
 
CRITICAL INSTRUCTIONS:
1. Call tools to gather information needed for your decision
2. Consider: device type, user behavior, session context, \
variant performance, similar user patterns
3. Make data-driven decisions based on tool results
4. Your final response MUST be ONLY valid JSON with no additional text
 
RESPONSE FORMAT: Return ONLY this JSON object:
{"variant_id": "A or B", "confidence": 0.0-1.0, \
"reasoning": "Detailed explanation"}"""
 
def build_user_prompt(user_id, experiment_id, device_type,
                      is_mobile, current_page, referrer_type,
                      extra_context=""):
    ctx = (f"\n\nADDITIONAL CONTEXT:\n{extra_context}"
           if extra_context else "")
    return f"""Select the optimal variant for this user \
in experiment {experiment_id}.
 
USER CONTEXT:
- User ID: {user_id}
- Device: {device_type} (Mobile: {is_mobile})
- Current Page: {current_page}
- Referrer: {referrer_type}{ctx}
 
AVAILABLE VARIANTS:
- Variant A: "Buy Now" — Clean, direct CTA
- Variant B: "Buy Now – Free Shipping" — CTA with incentive messaging
 
INSTRUCTIONS:
1. Call tools to gather user profile, similar users, \
and variant performance data
2. Analyze all signals together
3. Respond with ONLY the JSON object"""
 
def run_verification(user_id, experiment_id, device_type, is_mobile,
                     current_page, referrer_type,
                     use_conflicting=False, extra_context=""):
    prompt = build_user_prompt(
        user_id, experiment_id, device_type,
        is_mobile, current_page, referrer_type, extra_context)
    messages = [{"role": "user", "content": [{"text": prompt}]}]
    tool_calls_log = []
    turn = 0
    start_time = time.time()
 
    while True:
        turn += 1
        response = client.converse(
            modelId=MODEL_ID, messages=messages,
            system=[{"text": SYSTEM_PROMPT}],
            toolConfig={"tools": TOOL_DEFINITIONS},
        )
        output_message = response["output"]["message"]
        messages.append(output_message)
 
        if response["stopReason"] == "tool_use":
            tool_results = []
            for block in output_message["content"]:
                if "toolUse" in block:
                    tool = block["toolUse"]
                    result = execute_tool(
                        tool["name"], tool["input"],
                        use_conflicting=use_conflicting)
                    tool_calls_log.append({
                        "turn": turn, "tool": tool["name"],
                        "input": tool["input"]})
                    tool_results.append({"toolResult": {
                        "toolUseId": tool["toolUseId"],
                        "content": [{"json": json.loads(result)}],
                    }})
            messages.append({"role": "user", "content": tool_results})
 
        elif response["stopReason"] == "end_turn":
            final_text = "".join(
                b["text"] for b in output_message["content"]
                if "text" in b)
            return {
                "decision": json.loads(final_text.strip()),
                "tool_calls": tool_calls_log,
                "turns": turn,
                "elapsed_seconds": round(
                    time.time() - start_time, 2),
                "usage": response.get("usage", {}),
            }

import json
from verify import run_verification
 
def print_result(label, r):
    d = r["decision"]
    print(f"\n{'='*60}")
    print(f"  {label}")
    print(f"{'='*60}")
    print(f"  Variant:    {d.get('variant_id', '?')}")
    print(f"  Confidence: {d.get('confidence', '?')}")
    print(f"  Reasoning:  {d.get('reasoning', '?')}")
    print(f"\n  Tool calls ({len(r['tool_calls'])}):")
    for tc in r["tool_calls"]:
        print(f"    Turn {tc['turn']}: {tc['tool']}"
              f"({json.dumps(tc['input'])})")
    print(f"\n  Turns: {r['turns']}  |  Time: {r['elapsed_seconds']}s")
 
# 検証 1: ロイヤルティメンバー
r1 = run_verification(
    user_id="user_001", experiment_id="cta_test_2024",
    device_type="iPhone", is_mobile=True,
    current_page="/products/premium-headphones",
    referrer_type="direct",
    extra_context="Premium loyalty member (already has free shipping "
                  "benefit). Fast, goal-oriented browsing pattern. "
                  "Frequent purchaser.",
)
print_result("Verification 1: Loyalty Member", r1)
 
# 検証 2: 新規ユーザー
r2 = run_verification(
    user_id="user_002", experiment_id="cta_test_2024",
    device_type="Android", is_mobile=True,
    current_page="/deals/spring-sale",
    referrer_type="coupon_site (RetailMeNot)",
    extra_context="First-time visitor. No loyalty status. "
                  "Slow, comparison-focused browsing pattern.",
)
print_result("Verification 2: New User", r2)
 
# 検証 3: 矛盾するシグナル
r3 = run_verification(
    user_id="user_003", experiment_id="cta_test_2024",
    device_type="Desktop Chrome", is_mobile=False,
    current_page="/products/wireless-earbuds",
    referrer_type="direct",
    use_conflicting=True,
)
print_result("Verification 3: Conflicting Signals", r3)

python run_all.py

r1 = run_verification(
    user_id="user_001", experiment_id="cta_test_2024",
    device_type="iPhone", is_mobile=True,
    current_page="/products/premium-headphones",
    referrer_type="direct",
    extra_context="Premium loyalty member (already has free shipping benefit). "
                  "Fast, goal-oriented browsing pattern. Frequent purchaser.",
)

r2 = run_verification(
    user_id="user_002", experiment_id="cta_test_2024",
    device_type="Android", is_mobile=True,
    current_page="/deals/spring-sale",
    referrer_type="coupon_site (RetailMeNot)",
    extra_context="First-time visitor. No loyalty status. "
                  "Slow, comparison-focused browsing pattern.",
)

r3 = run_verification(
    user_id="user_003", experiment_id="cta_test_2024",
    device_type="Desktop Chrome", is_mobile=False,
    current_page="/products/wireless-earbuds",
    referrer_type="direct",
    use_conflicting=True,  # VARIANT_PERFORMANCE_CONFLICTING を使用
)