Simulazione utente

La simulazione utente utilizza un LLM-backed attore per interpretare il ruolo di un utente finale che interagisce con il tuo agente. Tu definisci il profilo e l'obiettivo dell'attore e l'attore conduce una conversazione a più turni con il tuo agente fino al raggiungimento dell'obiettivo o al raggiungimento del limite di turni.

Questa funzionalità è utile quando desideri:

La simulazione utente funziona sia con i programmi di esecuzione di set di dati su richiesta che in batch.

Come funziona

Il runner elabora ogni scenario simulato attraverso un ciclo di conversazione:

Profilo dell'attore

Ogni scenario simulato richiede un ActorProfile nome che definisca chi è l'attore e cosa vuole ottenere:

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

Configurazione della simulazione

SimulationConfigControlla il comportamento dell'attore ed è impostato nella configurazione di valutazione del corridore:

from bedrock_agentcore.evaluation import SimulationConfig

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

Schema del set di dati

Uno scenario simulato utilizza actor_profile e input invece di: 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"
      ]
    }
  ]
}

FileDatasetProviderrileva automaticamente il tipo di scenario dalla struttura JSON: gli scenari con un actor_profile campo (e nessun turns campo) vengono caricati come. SimulatedScenario

Utilizzo con il batch dataset runner

L'esempio seguente esegue una valutazione simulata dello scenario utilizzando il batch dataset runner. Attiva BatchEvaluationRunConfig e includi le SimulatedScenario istanze nel set simulation_config di dati:

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

Utilizzo con il dataset runner su richiesta

Il programma di esecuzione del set di dati su richiesta segue lo stesso schema. Imposta simulation_config EvaluationRunConfig e includi le SimulatedScenario istanze nel set di dati:

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')})")

Condizioni di interruzione

Una conversazione simulata termina quando viene soddisfatta una delle seguenti condizioni:

Suggerimenti per scenari simulati efficaci