AgentCore Evaluations のバッチ評価（Preview）を試してみた

"""Corporate IT helpdesk agent."""
from strands import Agent, tool
from strands.models.bedrock import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
SYSTEM_PROMPT = (
    "あなたは社内ITヘルプデスクのアシスタントです。"
    "社員からのIT関連の問い合わせ、パスワードリセット、"
    "システム稼働状況の確認、チケット管理をサポートしてください。"
)

@tool
def lookup_ticket(ticket_id: str) -> str:
    """Look up a support ticket by ID and return its status and details."""
    tickets = {
        "TKT-2001": {
            "status": "open",
            "category": "network",
            "summary": "VPN connection drops during video calls",
            "priority": "high",
            "created": "2026-05-20",
            "assignee": "network-team",
        },
        "TKT-2002": {
            "status": "in_progress",
            "category": "hardware",
            "summary": "Laptop battery drains in 2 hours",
            "priority": "medium",
            "created": "2026-05-18",
            "assignee": "hardware-team",
            "note": "Replacement battery ordered, arriving May 25",
        },
        "TKT-2003": {
            "status": "resolved",
            "category": "software",
            "summary": "Slack notifications not working on macOS",
            "priority": "low",
            "created": "2026-05-15",
            "resolved": "2026-05-16",
            "resolution": "Updated Slack to v4.41, re-enabled notification permissions",
        },
        "TKT-2004": {
            "status": "open",
            "category": "access",
            "summary": "Cannot access AWS console after team transfer",
            "priority": "high",
            "created": "2026-05-22",
            "assignee": "iam-team",
        },
        "TKT-2005": {
            "status": "pending",
            "category": "software",
            "summary": "Need license for JetBrains IntelliJ IDEA",
            "priority": "low",
            "created": "2026-05-21",
            "note": "Waiting for manager approval",
        },
    }
    return str(tickets.get(ticket_id, {"error": f"Ticket {ticket_id} not found"}))

@tool
def reset_password(employee_id: str, system: str) -> str:
    """Reset password for an employee on a specified system."""
    supported = {"active_directory", "aws_console", "jira", "confluence", "slack"}
    if system.lower() not in supported:
        return f"System '{system}' is not supported. Supported: {', '.join(sorted(supported))}"
    return (
        f"Password reset completed for {employee_id} on {system}. "
        "Temporary password sent to registered email. "
        "Please change it within 24 hours."
    )

@tool
def check_system_status(system_name: str) -> str:
    """Check the current operational status of a company system."""
    statuses = {
        "vpn": {
            "status": "degraded",
            "message": "Intermittent drops reported in Tokyo office. Engineering investigating.",
            "since": "2026-05-24T08:30:00+09:00",
        },
        "email": {
            "status": "operational",
            "message": "All email services running normally.",
        },
        "aws_console": {
            "status": "operational",
            "message": "All AWS services accessible.",
        },
        "jira": {
            "status": "maintenance",
            "message": "Scheduled maintenance until 22:00 JST. Read-only access available.",
        },
    }
    info = statuses.get(system_name.lower())
    if not info:
        return f"System '{system_name}' not found. Available: {', '.join(sorted(statuses.keys()))}"
    return str(info)

@tool
def create_ticket(category: str, summary: str, priority: str) -> str:
    """Create a new support ticket for the employee."""
    return (
        "Ticket TKT-3001 created successfully. "
        f"Category: {category}, Priority: {priority}. "
        f"Summary: {summary}. "
        "You will receive email updates on progress."
    )

@tool
def escalate_to_engineer(ticket_id: str, reason: str) -> str:
    """Escalate a ticket to a senior engineer for immediate attention."""
    return (
        f"Ticket {ticket_id} escalated to senior engineering team. "
        f"Reason: {reason}. "
        "Expected response within 30 minutes during business hours."
    )

agent = Agent(
    model=BedrockModel(model_id=MODEL_ID),
    tools=[lookup_ticket, reset_password, check_system_status,
           create_ticket, escalate_to_engineer],
    system_prompt=SYSTEM_PROMPT,
)

@app.entrypoint
def invoke(payload, context):
    result = agent(payload.get("prompt", "Hello"))
    return {"response": str(result)}

if __name__ == "__main__":
    app.run()

VPN の状態を確認しました。以下の結果です：

現在のシステム状態： 低下中（Degraded）

- 詳細: 東京オフィスで断続的な接続ドロップが報告されています
- 対応状況: エンジニアリングチームが現在調査中です
- 発生時刻: 2026年5月24日 08:30 から

VPN の不安定さはシステム側の問題が原因のようです。
エンジニアが対応中なので、状況の改善を待つようお願いします。

確認結果をお知らせします：

メールシステムの状況： 正常に稼働しています
- 現在、すべてのメールサービスは正常に動作しており、システム障害は確認されません。

サポートチケットを作成しました：
- チケットID: TKT-3001
- カテゴリ: メール
- 優先度: 高
- 内容: メールが送信できない問題

システムは正常ですが、個別の環境での問題の可能性があります。

import boto3
import uuid
import time
import json

REGION = "us-east-1"
SERVICE_NAME = "HelpDesk-xyz789.DEFAULT"
LOG_GROUP = "/aws/bedrock-agentcore/runtimes/HelpDesk-xyz789-DEFAULT"

eval_client = boto3.client("bedrock-agentcore", region_name=REGION)

response = eval_client.start_batch_evaluation(
    batchEvaluationName=f"helpdesk_baseline_{uuid.uuid4().hex[:8]}",
    evaluators=[
        {"evaluatorId": "Builtin.GoalSuccessRate"},
        {"evaluatorId": "Builtin.Helpfulness"},
        {"evaluatorId": "Builtin.Faithfulness"},
    ],
    dataSourceConfig={
        "cloudWatchLogs": {
            "serviceNames": [SERVICE_NAME],
            "logGroupNames": [LOG_GROUP],
        }
    },
    clientToken=str(uuid.uuid4()),
)

batch_eval_id = response["batchEvaluationId"]
print(f"Started: {batch_eval_id}")

while True:
    result = eval_client.get_batch_evaluation(batchEvaluationId=batch_eval_id)
    status = result["status"]
    print(f"Status: {status}")

    if status in ("COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED", "STOPPED"):
        break
    time.sleep(30)

print(json.dumps(result, indent=4, default=str))

Score: 1.0
Label: Completely Yes
Explanation: The assistant's response provides information about ticket TKT-2001
based on the tool output. Let me verify each piece of information:

1. Status: Tool shows 'open', Assistant says 'オープン（対応中）' - Matches ✓
2. Category: Tool shows 'network', Assistant says 'ネットワーク' - Matches ✓
3. Summary: Tool shows 'VPN connection drops during video calls',
   Assistant says 'ビデオ通話中のVPN接続切断' - Matches ✓
4. Priority: Tool shows 'high', Assistant says '高' - Matches ✓
5. Created date: Tool shows '2026-05-20', Assistant says '2026年5月20日' - Matches ✓
6. Assignee: Tool shows 'network-team', Assistant says 'ネットワークチーム' - Matches ✓

All information in the assistant's response accurately reflects the data from
the tool output.

Score: 1.0
Label: Above And Beyond
Explanation: The user had two clear goals: (1) determine if there's a system outage
preventing email sending, and (2) create a support ticket. The assistant successfully
accomplished both goals. The response clearly communicates that the email system is
operational (no outage), and confirms ticket TKT-3001 was created with appropriate
details (high priority, email category).

Beyond fulfilling the explicit requests, the assistant goes further by:
- Providing clear, well-structured information with visual formatting
- Explaining that while the system is operational, individual environment issues may exist
- Offering three concrete troubleshooting steps the user can try immediately

import boto3

eval_client = boto3.client("bedrock-agentcore", region_name="us-east-1")

baseline = eval_client.get_batch_evaluation(batchEvaluationId=baseline_id)
treatment = eval_client.get_batch_evaluation(batchEvaluationId=treatment_id)

baseline_summaries = {
    s["evaluatorId"]: s["statistics"]["averageScore"]
    for s in baseline["evaluationResults"]["evaluatorSummaries"]
}
treatment_summaries = {
    s["evaluatorId"]: s["statistics"]["averageScore"]
    for s in treatment["evaluationResults"]["evaluatorSummaries"]
}

print(f"{'Evaluator':<35} {'Baseline':>10} {'Treatment':>10} {'Delta':>10}")
print("=" * 67)
for eid in baseline_summaries:
    b = baseline_summaries[eid]
    t = treatment_summaries.get(eid, 0)
    delta = t - b
    print(f"{eid:<35} {b:>10.4f} {t:>10.4f} {delta:>+10.4f}")