온디맨드 평가 시작하기

다음 단계에 따라 첫 번째 온디맨드 평가를 설정하고 실행합니다.

사전 조건

AgentCore Evaluations OnDemand Evaluation 기능을 사용하려면 다음이 필요합니다.

지원되는 프레임워크

AgentCore Evaluations는 현재 다음과 같은 에이전트 프레임워크 및 계측 라이브러리를 지원합니다.

1단계: 에이전트 생성 및 배포

AgentCore 런타임 시작 안내서에 따라 에이전트를 생성하고 배포합니다. AgentCore 평가 샘플에서 추가 예제를 찾을 수 있습니다.

2단계: 에이전트 호출

다음 명령을 사용하여 에이전트를 호출하고 CloudWatch의 GenAI 관찰성 대시보드에서 트레이스, 세션 및 지표를 봅니다.

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에서 span-logs 다운로드

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"])

위에서를 사용하고 세션 기간을 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 파라미터를 사용하여 대상을 지정할 수 있습니다.

세션 수준 평가자

서비스는 평가당 하나의 세션만 지원하므로 평가 대상을 명시적으로 설정할 필요가 없습니다.

트레이스 수준 평가자

트레이스 수준 평가자(예: 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"]}
)

도구 호출 수준 평가자

스팬 수준 평가자(예: )의 경우 evaluationTarget 파라미터에서 Builtin.ToolSelectionAccuracy 스팬 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 호출은 평가자 결과 목록이 포함된 응답을 반환합니다. 단일 세션에는 여러 트레이스와 도구 호출이 포함될 수 있으므로 이러한 요소는 별도의 엔터티로 평가됩니다. 따라서 단일 API 호출은 여러 평가 결과를 반환할 수 있습니다.

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

결과 제한

API 호출당 반환되는 평가 수는 결과 10개로 제한됩니다. 예를 들어 트레이스 수준 평가자를 사용하여 15개의 트레이스가 포함된 세션을 평가하는 경우 응답에는 최대 10개의 결과가 포함됩니다. API는 일반적으로 평가 품질과 관련된 가장 많은 컨텍스트를 포함하므로 기본적으로 마지막 10개의 평가를 반환합니다.

부분 실패

API 호출은 n개의 평가가 실패하는 동안 이를 처리할 수 있습니다. 다음과 같은 다양한 이유로 장애가 발생할 수 있습니다.

부분 실패의 경우 응답에는 성공한 평가와 실패한 평가가 모두 포함됩니다. 실패한 결과에는 문제를 진단하는 데 도움이 되는 오류 코드와 오류 메시지가 포함됩니다.

스팬 컨텍스트

각 평가자 결과에는 평가된 개체를 식별하는 spanContext 필드가 있습니다.

성공한 결과 항목의 예

이는 하나의 항목일 뿐입니다. 세션에 트레이스가 여러 개 있는 경우 트레이스마다 하나씩 이러한 항목이 여러 개 표시됩니다. 도구 수준 평가자와 마찬가지로 도구 호출이 여러 개 있고 도구 평가자(예: Builtin.ToolSelectionAccuracy)가 제공되는 경우 도구 범위당 하나의 결과가 있습니다.

{
  "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...."
}