View a markdown version of this page

Cómo empezar con la evaluación de lotes - Amazon Bedrock AgentCore

Cómo empezar con la evaluación de lotes

En este tutorial, se pasa de un agente desplegado a los resultados de una evaluación por lotes con un agente de atención al cliente de Acme Store. Creará el agente, lo desplegará, generará sesiones de muestra, realizará una evaluación por lotes y leerá los resultados.

Antes de empezar

Asegúrese de que:

  • La AgentCore CLI instalada (agentcore --version)

  • AWS credenciales con permisos para bedrock-agentcore y logs

  • La búsqueda de transacciones está habilitada en CloudWatch

  • Python 3.10+ (para ejemplos de boto3)

Para obtener más información, consulte Requisitos previos.

En los ejemplos de boto3 se utilizan las siguientes constantes. Sustitúyalos por sus propios valores después de implementar el agente:

REGION = "us-west-2" AGENT_ARN = "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/AcmeSupport-abc123" SERVICE_NAME = "AcmeSupport-abc123.DEFAULT" LOG_GROUP = "/aws/bedrock-agentcore/runtimes/AcmeSupport-abc123-DEFAULT"

Paso 1: Crear e implementar el agente de muestra

Cree un AgentCore proyecto y sustituya el código de agente predeterminado por el del agente de atención al cliente de Acme Store. Este agente tiene cinco herramientas para gestionar pedidos, devoluciones, envíos, descuentos y escaladas.

Creación del proyecto

agentcore create --name AcmeSupport --framework Strands --model-provider Bedrock --memory none cd AcmeSupport

Sustituya el código del agente

Abra app/AcmeSupport/main.py y sustituya su contenido por lo siguiente:

"""Acme Store customer support agent.""" from strands import Agent, tool from strands.models.bedrock import BedrockModel from bedrock_agentcore.runtime import BedrockAgentCoreApp app = BedrockAgentCoreApp() MODEL_ID = "global.anthropic.claude-sonnet-4-6" SYSTEM_PROMPT = ( "You are a helpful customer support assistant for Acme Store. " "Help customers with their orders, returns, and shipping questions." ) @tool def lookup_order(order_id: str) -> str: """Look up an order by ID and return its status, item, and delivery details.""" orders = { "ORD-1001": { "status": "delivered", "item": "Blue T-Shirt (L)", "delivered": "2026-03-28", "total": "$29.99", }, "ORD-1002": { "status": "in_transit", "item": "Running Shoes (10)", "shipped": "2026-03-30", "est_delivery": "2026-04-05", "total": "$89.99", }, "ORD-1003": { "status": "delayed", "item": "Wireless Headphones", "shipped": "2026-03-25", "est_delivery": "2026-03-29", "days_late": 5, "total": "$59.99", }, "ORD-1004": { "status": "processing", "item": "Yoga Mat", "ordered": "2026-04-02", "total": "$34.99", }, "ORD-1005": { "status": "delivered", "item": "Coffee Maker", "delivered": "2026-03-20", "total": "$149.99", }, } return str(orders.get(order_id, {"error": f"Order {order_id} not found"})) @tool def initiate_return(order_id: str, reason: str) -> str: """Initiate a return for an order. Sends a return label to the customer.""" return ( f"Return initiated for {order_id}. Reason: {reason}. " "Return label sent to customer email. Please ship within 14 days." ) @tool def check_shipping_status(order_id: str) -> str: """Check detailed shipping status including carrier location and delays.""" statuses = { "ORD-1002": ( "Package is with carrier, currently in Portland OR. " "On schedule for April 5." ), "ORD-1003": ( "Package delayed at distribution center in Memphis TN. " "Original delivery was March 29. Now 5 days late. " "Acme Store policy: orders delayed 3+ days qualify for 15% discount." ), } return statuses.get(order_id, f"No active shipment found for {order_id}.") @tool def apply_discount(order_id: str, discount_percent: int, reason: str) -> str: """Apply a percentage discount to an order and issue a refund.""" return ( f"Applied {discount_percent}% discount to {order_id}. " f"Reason: {reason}. Refund will appear in 3-5 business days." ) @tool def escalate_to_human(reason: str) -> str: """Escalate the conversation to a human support agent.""" return ( f"Escalated to human agent. Reason: {reason}. " "Estimated wait time: 3 minutes." ) agent = Agent( model=BedrockModel(model_id=MODEL_ID), tools=[lookup_order, initiate_return, check_shipping_status, apply_discount, escalate_to_human], 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()

Implemente y verifique

agentcore deploy

Tras la implementación, compruebe que el agente se esté ejecutando:

agentcore invoke --prompt "What's the status of order ORD-1001?"

Debería ver una respuesta con los detalles del pedido. Anote el ARN del tiempo de ejecución, el nombre del servicio y el grupo de registros de origenagentcore status --json; los necesitará para los ejemplos de boto3.

nota

Si ya tienes un agente implementado en AgentCore Runtime con la observabilidad habilitada, omite este paso y usa tu propio agente durante el resto del tutorial.

Paso 2: Genere sesiones de muestra

Invoque al agente con diversas indicaciones para crear sesiones de evaluación. Estas indicaciones cubren diferentes escenarios: búsquedas de pedidos, devoluciones, retrasos en los envíos, solicitudes de descuento e interacciones multiherramienta.

ejemplo
AgentCore CLI
agentcore invoke --runtime AcmeSupport --prompt "What's the status of my order ORD-1001?" agentcore invoke --runtime AcmeSupport --prompt "I need to return order ORD-1001, the shirt doesn't fit." agentcore invoke --runtime AcmeSupport --prompt "What's the shipping status on ORD-1002?" agentcore invoke --runtime AcmeSupport --prompt "My order ORD-1003 is delayed, can you help?" agentcore invoke --runtime AcmeSupport --prompt "I'd like to check on order ORD-1004 please." agentcore invoke --runtime AcmeSupport --prompt "Can you look up order ORD-1005 for me?" agentcore invoke --runtime AcmeSupport --prompt "I want to return the coffee maker from order ORD-1005, it's defective." agentcore invoke --runtime AcmeSupport --prompt "Where is my order ORD-1002? It should have arrived by now." agentcore invoke --runtime AcmeSupport --prompt "ORD-1003 is really late, I want a discount." agentcore invoke --runtime AcmeSupport --prompt "Can you check order ORD-1001 and tell me when it was delivered?"
AWS SDK (boto3)
import boto3 import json import uuid client = boto3.client("bedrock-agentcore", region_name=REGION) prompts = [ "What's the status of my order ORD-1001?", "I need to return order ORD-1001, the shirt doesn't fit.", "What's the shipping status on ORD-1002?", "My order ORD-1003 is delayed, can you help?", "I'd like to check on order ORD-1004 please.", "Can you look up order ORD-1005 for me?", "I want to return the coffee maker from order ORD-1005, it's defective.", "Where is my order ORD-1002? It should have arrived by now.", "ORD-1003 is really late, I want a discount.", "Can you check order ORD-1001 and tell me when it was delivered?", ] for i, prompt in enumerate(prompts): session_id = f"acme-eval-{uuid.uuid4().hex[:12]}" print(f"[{i+1}/10] {prompt[:60]}...") response = client.invoke_agent_runtime( agentRuntimeArn=AGENT_ARN, runtimeSessionId=session_id, payload=json.dumps({"prompt": prompt}).encode(), ) response_body = response["response"].read() print(f" Done (session: {session_id})") print("\nAll sessions created.")

Espere de 2 a 3 minutos después de la última invocación para asimilar la telemetría antes de continuar CloudWatch .

Paso 3: Ejecute la evaluación por lotes

Inicie una evaluación por lotes para puntuar todas las sesiones recientes. El servicio descubre las sesiones a partir de CloudWatch los registros, ejecuta cada evaluador en función de cada sesión y devuelve los resultados agregados.

ejemplo
AgentCore CLI
agentcore run batch-evaluation \ --runtime AcmeSupport \ --evaluator Builtin.GoalSuccessRate Builtin.Helpfulness Builtin.Faithfulness \ --wait

De forma predeterminada, agentcore run batch-evaluation inicia el trabajo y lo devuelve inmediatamente (sin bloquearlo). Pase --wait a bloquear hasta que la tarea alcance un estado terminal. Con--wait, la CLI resuelve el grupo de CloudWatch registros y el nombre del servicio de la configuración del proyecto, inicia el trabajo, bloquea hasta que alcanza un estado terminal y, a continuación, imprime las puntuaciones promedio por evaluador:

Batch evaluation completed: acme-eval-a1b2c3d4

Sessions: 10 completed, 0 failed, 10 total

Evaluator                           Avg Score
─────────────────────────────────────────────
Builtin.GoalSuccessRate             0.7200
Builtin.Helpfulness                 0.8100
Builtin.Faithfulness                0.8500

Results saved to .cli/jobs/batch-eval-results/

--jsonAñádalo para emitir resultados legibles por máquina (incluidos los del evaluadoraverageScore) para la creación de scripts batchEvaluationId y para etiquetar la ejecución de modo que pueda -n <name> comparar los resultados de las distintas ejecuciones. Por ejemplo:

agentcore run batch-evaluation \ --runtime AcmeSupport \ --evaluator Builtin.GoalSuccessRate Builtin.Helpfulness Builtin.Faithfulness \ -n acme_baseline \ --wait
AWS SDK (boto3)
import boto3 import uuid import time import json eval_client = boto3.client("bedrock-agentcore", region_name=REGION) # Start the batch evaluation response = eval_client.start_batch_evaluation( batchEvaluationName=f"acme_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}") # Poll until complete 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))

Paso 4: Lea los detalles de cada sesión

Las puntuaciones agregadas muestran el panorama general. Para ver las puntuaciones por turno y por evaluador de las sesiones individuales, utilice los comandos de visualización de la CLI integrados o lea los eventos de evaluación directamente desde los registros. CloudWatch

ejemplo
AgentCore CLI

La CLI proporciona comandos de primera clase para ver los trabajos de evaluación de lotes completados y sus resultados. Vea un trabajo específico por su identificador de trabajo de evaluación de lotes o enumere los trabajos anteriores:

# View a batch evaluation job and its results agentcore view batch-evaluation acme-eval-a1b2c3d4 # List batch evaluation jobs agentcore batch-evaluations history

Estos comandos se ejecutan de forma interactiva cuando no se muestra ningún indicador. --jsonAñádalo para una salida no interactiva y legible por máquina, por ejemplo. agentcore view batch-evaluation acme-eval-a1b2c3d4 --json

AWS SDK (boto3)
# Get the output location from the batch evaluation result output = result["outputConfig"]["cloudWatchConfig"] log_group = output["logGroupName"] log_stream = output["logStreamName"] # Read the events logs_client = boto3.client("logs", region_name=REGION) response = logs_client.get_log_events( logGroupName=log_group, logStreamName=log_stream, ) for event in response["events"]: event_attrs = json.loads(event["message"]).get("attributes", {}) print(f"Score: {event_attrs.get('gen_ai.evaluation.score.value')}") print(f"Label: {event_attrs.get('gen_ai.evaluation.score.label')}") print(f"Explanation: {event_attrs.get('gen_ai.evaluation.explanation', '')[:200]}") print()

Siguientes pasos