Simulation utilisateur

La simulation utilisateur fait appel à un LLM-backed acteur pour jouer le rôle d'un utilisateur final interagissant avec votre agent. Vous définissez le profil et l'objectif de l'acteur, et celui-ci mène une conversation à plusieurs tours avec votre agent jusqu'à ce que l'objectif soit atteint ou que la limite de tours soit atteinte.

Cela est utile lorsque vous souhaitez :

La simulation utilisateur fonctionne à la fois avec les exécuteurs de jeux de données à la demande et par lots.

Comment ça marche

Le coureur traite chaque scénario simulé par le biais d'une boucle de conversation :

Profil de l'acteur

Chaque scénario simulé nécessite un ActorProfile qui définit qui est l'acteur et ce qu'il souhaite accomplir :

{
  "actor_profile": {
    "context": "A customer who purchased a laptop last week and it arrived with a cracked screen",
    "goal": "Get a replacement laptop shipped within 2 business days",
    "traits": {
      "expertise": "non-technical",
      "tone": "frustrated but polite",
      "patience": "low"
    }
  }
}

Configuration de simulation

Le SimulationConfig contrôle le comportement de l'acteur et est défini sur la configuration d'évaluation du coureur :

from bedrock_agentcore.evaluation import SimulationConfig

simulation_config = SimulationConfig(
    model_id="<model-id>",
)

schéma du jeu de données

Un scénario simulé utilise actor_profile et input au lieu de turns :

{
  "scenarios": [
    {
      "scenario_id": "geography-student",
      "scenario_description": "A curious student asks geography questions",
      "actor_profile": {
        "traits": {"expertise": "novice", "tone": "curious"},
        "context": "A student studying world geography who wants to learn about capitals",
        "goal": "Find out the capital cities of at least two different countries"
      },
      "input": "Hi! I'm studying geography. Can you help me learn about world capitals?",
      "max_turns": 5,
      "assertions": [
        "Agent provides accurate capital city information",
        "Agent is helpful and encouraging to the student"
      ]
    }
  ]
}

FileDatasetProviderdétecte automatiquement le type de scénario à partir de la structure JSON : les scénarios avec un actor_profile champ (et aucun turns champ) sont chargés en tant queSimulatedScenario.

Utilisation avec le gestionnaire de jeux de données par lots

L'exemple suivant exécute une évaluation de scénario simulée à l'aide du gestionnaire de données par lots. Définissez simulation_config BatchEvaluationRunConfig et incluez des SimulatedScenario instances dans le jeu de données :

import boto3
import json
from bedrock_agentcore.evaluation import (
    BatchEvaluationRunner,
    BatchEvaluationRunConfig,
    BatchEvaluatorConfig,
    CloudWatchDataSourceConfig,
    SimulationConfig,
    AgentInvokerInput,
    AgentInvokerOutput,
    Dataset,
    SimulatedScenario,
    ActorProfile,
)

AGENT_ARN = "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/MyAgent-abc123"  # Replace with your agent runtime ARN
REGION = "us-west-2"  # Replace with your region
RUNTIME_ID = AGENT_ARN.split("/")[-1]
AGENT_NAME = RUNTIME_ID.rsplit("-", 1)[0]
ENDPOINT_NAME = "DEFAULT"
LOG_GROUP = f"/aws/bedrock-agentcore/runtimes/{RUNTIME_ID}-{ENDPOINT_NAME}"
SERVICE_NAME = f"{AGENT_NAME}.{ENDPOINT_NAME}"
ACTOR_MODEL_ID = "global.anthropic.claude-haiku-4-5-20251001-v1:0"  # Replace with your preferred model

# Define the dataset with simulated scenarios
dataset = Dataset(
    scenarios=[
        SimulatedScenario(
            scenario_id="support-frustrated-customer",
            scenario_description="A frustrated customer with a defective product",
            actor_profile=ActorProfile(
                traits={"expertise": "non-technical", "tone": "frustrated but polite"},
                context="Purchased a laptop last week that arrived with a cracked screen",
                goal="Get a replacement laptop shipped within 2 business days",
            ),
            input="I received my laptop and the screen is cracked. I need help.",
            max_turns=8,
            assertions=[
                "Agent acknowledges the issue and apologizes",
                "Agent offers a replacement or refund",
                "Agent provides a timeline for resolution",
            ],
        ),
        SimulatedScenario(
            scenario_id="support-billing-question",
            scenario_description="A customer with a billing discrepancy",
            actor_profile=ActorProfile(
                traits={"expertise": "moderate", "tone": "calm"},
                context="Noticed a double charge on the last credit card statement",
                goal="Get the duplicate charge reversed and confirmation of the refund",
            ),
            input="I see two charges for the same order on my statement. Can you look into this?",
            max_turns=6,
            assertions=[
                "Agent investigates the billing issue",
                "Agent confirms whether a duplicate charge exists",
            ],
        ),
    ]
)

# Configure the evaluation
config = BatchEvaluationRunConfig(
    batch_evaluation_name="simulated-support-eval",
    evaluator_config=BatchEvaluatorConfig(
        evaluator_ids=[
            "Builtin.GoalSuccessRate",
            "Builtin.Helpfulness",
        ],
    ),
    data_source=CloudWatchDataSourceConfig(
        service_names=[SERVICE_NAME],
        log_group_names=[LOG_GROUP],
        ingestion_delay_seconds=180,
    ),
    simulation_config=SimulationConfig(
        model_id=ACTOR_MODEL_ID,
    ),
    polling_timeout_seconds=1800,
    polling_interval_seconds=30,
)

# Define the agent invoker
agentcore_client = boto3.client("bedrock-agentcore", region_name=REGION)

def agent_invoker(inp: AgentInvokerInput) -> AgentInvokerOutput:
    payload = inp.payload
    if isinstance(payload, str):
        raw_bytes = json.dumps({"prompt": payload}).encode()
    elif isinstance(payload, dict):
        raw_bytes = json.dumps(payload).encode()
    else:
        raw_bytes = json.dumps({"prompt": str(payload)}).encode()

    print(f"[{inp.session_id}] > sending payload: {raw_bytes.decode()}")
    response = agentcore_client.invoke_agent_runtime(
        agentRuntimeArn=AGENT_ARN,
        runtimeSessionId=inp.session_id,
        payload=raw_bytes,
    )
    response_body = response["response"].read()
    print(f"[{inp.session_id}] < received response: {response_body.decode()}")
    return AgentInvokerOutput(agent_output=json.loads(response_body))

# Run the evaluation
runner = BatchEvaluationRunner(region=REGION)
result = runner.run_dataset_evaluation(
    config=config,
    dataset=dataset,
    agent_invoker=agent_invoker,
)

# Display results
print(f"Status: {result.status}")
if result.evaluation_results:
    er = result.evaluation_results
    print(f"Sessions completed: {er.number_of_sessions_completed}")
    print(f"Sessions failed:    {er.number_of_sessions_failed}")
    for summary in er.evaluator_summaries or []:
        avg = summary.statistics.average_score if summary.statistics else None
        print(f"  {summary.evaluator_id}: avg={avg}")

Utilisation avec le générateur de jeux de données à la demande

Le lanceur de jeux de données à la demande suit le même schéma. Définissez simulation_config EvaluationRunConfig et incluez des SimulatedScenario instances dans le jeu de données :

from bedrock_agentcore.evaluation import (
    OnDemandEvaluationDatasetRunner,
    EvaluationRunConfig,
    EvaluatorConfig,
    CloudWatchAgentSpanCollector,
    SimulationConfig,
    FileDatasetProvider,
)

AGENT_ARN = "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/MyAgent-abc123"  # Replace with your agent runtime ARN
REGION = "us-west-2"  # Replace with your region
RUNTIME_ID = AGENT_ARN.split("/")[-1]
ENDPOINT_NAME = "DEFAULT"
LOG_GROUP = f"/aws/bedrock-agentcore/runtimes/{RUNTIME_ID}-{ENDPOINT_NAME}"
ACTOR_MODEL_ID = "global.anthropic.claude-haiku-4-5-20251001-v1:0"  # Replace with your preferred model

# Load dataset (auto-detects simulated scenarios from actor_profile field)
dataset = FileDatasetProvider("simulated_dataset.json").get_dataset()

# Create span collector
span_collector = CloudWatchAgentSpanCollector(
    log_group_name=LOG_GROUP,
    region=REGION,
)

# Configure with simulation support
config = EvaluationRunConfig(
    evaluator_config=EvaluatorConfig(
        evaluator_ids=["Builtin.GoalSuccessRate", "Builtin.Helpfulness"],
    ),
    evaluation_delay_seconds=180,
    max_concurrent_scenarios=5,
    simulation_config=SimulationConfig(
        model_id=ACTOR_MODEL_ID,
    ),
)

# Run
runner = OnDemandEvaluationDatasetRunner(region=REGION)
result = runner.run(
    agent_invoker=agent_invoker,
    dataset=dataset,
    span_collector=span_collector,
    config=config,
)

for scenario in result.scenario_results:
    print(f"Scenario: {scenario.scenario_id} ({scenario.status})")
    for evaluator in scenario.evaluator_results:
        for r in evaluator.results:
            print(f"  {evaluator.evaluator_id}: {r.get('value')} ({r.get('label')})")

Conditions d'arrêt

Une conversation simulée prend fin lorsque l'une des conditions suivantes est remplie :

Conseils pour des scénarios simulés efficaces