オンデマンド評価の開始方法

以下の手順に従って、最初のオンデマンド評価をセットアップして実行します。

前提条件

AgentCore Evaluations OnDemand評価機能を使用するには、以下が必要です。

サポートされるフレームワーク

AgentCore Evaluations は現在、次のエージェントフレームワークと計測ライブラリをサポートしています。

ステップ 1: エージェントを作成してデプロイする

AgentCore ランタイム の開始方法ガイドに従って、エージェントを作成してデプロイします。その他の例については、「 AgentCore 評価サンプル」を参照してください。

ステップ 2: エージェントを呼び出す

次のコマンドを使用してエージェントを呼び出し、CloudWatch の GenAI Observability ダッシュボードでトレース、セッション、メトリクスを表示します。

invoke_agent.py の例

import boto3
import json
import uuid

region = "region-code"
ace_demo_agent_arn = "agent-arn from step-2"

agent_core_client = boto3.client('bedrock-agentcore', region_name=region)

text_to_analyze = "Sample text to test agent for agentcore evaluations demo"

payload = json.dumps({
    "prompt": f"Can you analyze this text and tell me about its statistics: {text_to_analyze}"
})

# random session-id, you can set your own here
session_id = "test-ace-demo-session-18a1dba0-62a0-462g"

response = agent_core_client.invoke_agent_runtime(
    agentRuntimeArn=ace_demo_agent_arn,
    runtimeSessionId=session_id,
    payload=payload,
    qualifier="DEFAULT"
)

response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)
print("SessionId:", session_id)

ステップ 3: エージェントを評価する

エージェントにいくつかの呼び出しを行うと、エージェントを評価する準備が整います。評価には以下が必要です。

AgentCore CLI および AgentCore SDK のコードサンプル

次のコードサンプルは、さまざまな開発アプローチを使用してオンデマンド評価を実行する方法を示しています。開発環境と設定に最適な方法を選択します。

AWS SDK

CloudWatch からスパンログをダウンロードする

Evaluate API を呼び出す前に、CloudWatch からスパンログをダウンロードする必要があります。以下の Python コードを使用してこれを行い、オプションで JSON ファイルに保存できます。これにより、異なる評価者との同じセッションのリクエストを簡単に行うことができます。

import boto3
import time
import json
from datetime import datetime, timedelta

region = "region-code"
agent_id = "agent-id-from-step-2"
session_id = "session-id-from-step-3"

def query_logs(log_group_name, query_string):
    client = boto3.client('logs', region_name=region)
    start_time = datetime.now() - timedelta(minutes=60) # past 1 hour
    end_time = datetime.now()

    query_id = client.start_query(
        logGroupName=log_group_name,
        startTime=int(start_time.timestamp()),
        endTime=int(end_time.timestamp()),
        queryString=query_string
    )['queryId']

    while (result := client.get_query_results(queryId=query_id))['status'] not in ['Complete', 'Failed']:
        time.sleep(1)

    if result['status'] == 'Failed':
        raise Exception("Query failed")
    return result['results']

def query_session_logs(log_group_name, session_id, **kwargs):
    query = f"""fields @timestamp, @message
    | filter ispresent(scope.name) and ispresent(attributes.session.id)
    | filter attributes.session.id = "{session_id}"
    | sort @timestamp asc"""
    return query_logs(log_group_name, query, **kwargs)

def query_agent_runtime_logs(agent_id, endpoint, session_id, **kwargs):
    return query_session_logs(
        f"/aws/bedrock-agentcore/runtimes/{agent_id}-{endpoint}",
        session_id, **kwargs)

def query_aws_spans_logs(session_id, **kwargs):
    return query_session_logs("aws/spans", session_id, **kwargs)

def extract_messages_as_json(query_results):
    return [json.loads(f['value']) for row in query_results
            for f in row if f['field'] == '@message'
            and f['value'].strip().startswith('{')]

def get_session_span_logs():
    agent_runtime_logs = query_agent_runtime_logs(
        agent_id=agent_id, endpoint="DEFAULT", session_id=session_id
    )
    print(f"Downloaded {len(agent_runtime_logs)} runtime-log entries")

    aws_span_logs = query_aws_spans_logs(session_id=session_id)
    print(f"Downloaded {len(aws_span_logs)} aws/span entries")

    session_span_logs = extract_messages_as_json(aws_span_logs) + extract_messages_as_json(agent_runtime_logs)
    print(f"Returning {len(aws_span_logs) + len(agent_runtime_logs)} total records")
    return session_span_logs

# get the spans from cloudwatch
session_span_logs = get_session_span_logs()

# optional (dump in a json file for reuse)
session_span_logs_file_name = "ace-demo-session.json"
with open(session_span_logs_file_name, "w") as f:
    json.dump(session_span_logs, f, indent=2)

評価を呼び出す

入力スパンを取得したら、 Evaluate API を呼び出すことができます。大規模な言語モデルがトレースをスコアリングしているため、レスポンスには時間がかかる場合があります。

# initialise client
ace_dp_client = boto3.client('bedrock-agentcore', region_name = region)

# call evaluate
response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.Helpfulness", # can be a custom evaluator id as well
    evaluationInput = {"sessionSpans": session_span_logs})

print(response["evaluationResults"])

上記の を使用して session-spans を JSON ファイルにダンプする場合、後で次のように評価を実行することもできます。

with open(session_span_logs_file_name, "r") as f:
    session_span_logs = json.load(f)

# initialise client
ace_dp_client = boto3.client('bedrock-agentcore', region_name = region)

# call evaluate
response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.ToolSelectionAccuracy", # can be a custom evaluator id as well
    evaluationInput = {"sessionSpans": session_span_logs})

print(response["evaluationResults"])

評価ターゲットの使用

セッション内の特定のトレースまたはツールを評価するには、リクエストの evaluationTargetパラメータを使用してターゲットを指定できます。

セッションレベルの評価者

サービスは評価ごとに 1 つのセッションのみをサポートしているため、評価ターゲットを明示的に設定する必要はありません。

トレースレベルの評価者

トレースレベルの評価者 ( Builtin.Helpfulnessや Builtin.Correctness など) の場合、 evaluationTargetパラメータでトレース IDs を設定します。

response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.Helpfulness",
    evaluationInput = {"sessionSpans": session_span_logs},
    evaluationTarget = {"traceIds": ["trace-id-1", "trace-id-2"]}
)

ツールコールレベルの評価者

スパンレベルの評価者 ( など) Builtin.ToolSelectionAccuracy の場合、 evaluationTargetパラメータでスパン IDs を設定します。

response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.ToolSelectionAccuracy",
    evaluationInput = {"sessionSpans": session_span_logs},
    evaluationTarget = {"spanIds": ["span-id-1", "span-id-2"]}
)

ステップ 4: 評価結果

各 Evaluate API コールは、評価者の結果のリストを含むレスポンスを返します。1 つのセッションに複数のトレースとツール呼び出しを含めることができるため、これらの要素は個別のエンティティとして評価されます。したがって、単一の API コールが複数の評価結果を返す可能性があります。

{
    "evaluationResults": [ {evaluation-result-1}, {evaluation-result_2},.... ]
}

結果の制限

API コールごとに返される評価の数は、10 の結果に制限されます。たとえば、トレースレベルの評価者を使用して 15 個のトレースを含むセッションを評価する場合、レスポンスには最大 10 個の結果が含まれます。デフォルトでは、API は過去 10 件の評価を返します。通常、評価品質に関連するコンテキストが最も多いためです。

部分的な障害

API コールは n 個の評価を処理できますが、そのうち m 個の評価は失敗します。障害は、次のようなさまざまな理由で発生する可能性があります。

部分的な失敗の場合、レスポンスには成功した評価と失敗した評価の両方が含まれます。失敗した結果には、問題の診断に役立つエラーコードとエラーメッセージが含まれます。

スパンコンテキスト

各評価者の結果には、評価されたエンティティを識別するspanContextフィールドがあります。

成功した結果エントリの例

これは 1 つのエントリにすぎません。セッションに複数のトレースがある場合、トレースごとに 1 つずつ、複数のエントリが表示されます。ツールレベルの評価者と同様に、複数のツール呼び出しがあり、ツール評価者 ( など) Builtin.ToolSelectionAccuracy が指定されている場合、ツールスパンごとに 1 つの結果になります。

{
  "evaluatorArn": "arn:aws:bedrock-agentcore:::evaluator/Builtin.Helpfulness",
  "evaluatorId": "Builtin.Helpfulness",
  "evaluatorName": "Builtin.Helpfulness",
  "explanation": ".... evaluation explanation will be added here ...",
  "context": {
    "spanContext": {
      "sessionId": "test-ace-demo-session-18a1dba0-62a0-462e",
      "traceId": "....trace_id......."
    }
  },
  "value": 0.83,
  "label": "Very Helpful",
  "tokenUsage": {
    "inputTokens": 958,
    "outputTokens": 211,
    "totalTokens": 1169
  }
}

失敗した結果エントリの例

{
    "evaluatorArn": "arn:aws:bedrock-agentcore:::evaluator/Builtin.Helpfulness",
    "evaluatorId": "Builtin.Helpfulness",
    "evaluatorName": "Builtin.Helpfulness",
    "context": {
        "spanContext": {
            "sessionId": "test-ace-demo-session-18a1dba0-62a0-462e",
            "traceId": "....trace_id......."
        }
    },
    "errorMessage": ".... details of the error....",
    "errorCode": ".... name/code of the error...."
}