本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
用户模拟
用户模拟使用角色扮 LLM-backed 演最终用户与您的代理进行交互的角色。您可以定义演员的个人资料和目标,然后该演员与您的经纪人进行多回合对话,直到达到目标或达到回合限制。
注意
用户模拟在 SDK 端调用 Amazon Bedrock 模型来生成参与者的响应。这些调用收取标准亚马逊 Bedrock 模型调用费。有关详细信息,请参阅定AgentCore 价页面
这在你想做时很有用:
-
使用真实变体进行测试:演员每次运行都会生成不同的措辞、后续问题和对话路径,揭示手写场景遗漏的边缘案例。
-
评估开放式对话:对于处理自由形式对话(客户支持、辅导、咨询)的客服,模拟场景比固定的回合顺序更能反映真实的用户行为。
-
扩大场景覆盖范围:与其手写数十个多回合剧本,不如用不同的角色和目标定义演员简介,让演员进行对话。
-
具有多样性的回归测试:多次运行相同的演员档案,以检查您的代理是否处理了具有相同意图的不同表达。
工作原理
运行者通过对话循环处理每个模拟场景:
-
开始:跑步者将场景
input字段作为第一回合发送给您的代理。 -
代理响应:您的代理处理输入并返回响应。
-
演员评估: LLM-backed 演员收到代理人的回复,并根据其个人资料和目标决定下一步该怎么做。参与者生成的结构化响应包含:
-
理由:演员做出回应的内部原因(例如,“代理提供了飞行选项,但没有询问我的首选时间。我应该说明我更喜欢早间航班。”)。这对于调试演员为何以某种方式行事很有用。
-
消息:要发送给代理的下一条消息。
-
停止信号:一个布尔值,表示行为者是否认为其目标已实现。
-
-
继续或停止:如果演员发出目标完成信号 (
stop: true) 或回合数已到达max_turns,则对话结束。否则,演员的下一条消息将成为下一回合的输入。 -
评估:对话完成后,运行者使用配置的评估器评估会话,与预定义场景相同。
演员简介
每个模拟场景ActorProfile都需要一个定义演员是谁及其想要实现的目标:
| 字段 | 必填 | 描述 |
|---|---|---|
|
|
是 |
有关演员的背景信息。描述情况以及演员应知道的任何相关细节。 |
|
|
是 |
演员想在对话中实现的目标。当演员确定目标已实现时,它会发出完成信号。 |
|
|
否 |
Key-value 描述演员特征(例如,专业水平、沟通风格、耐心)的配对。默认为空。 |
{ "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" } } }
仿真配置
SimulationConfig控制演员的行为,并在跑步者的评估配置上进行设置:
| 字段 | 默认值 | 说明 |
|---|---|---|
|
|
默认型号 |
用于演员 LLM 的亚马逊 Bedrock 模型 ID。选择可以遵循复杂角色指令的模型。如果省略,则使用默认模型。 |
from bedrock_agentcore.evaluation import SimulationConfig simulation_config = SimulationConfig( model_id="<model-id>", )
数据集架构
模拟场景使用actor_profile和input代替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" ] } ] }
| 字段 | 必填 | 默认值 | Description |
|---|---|---|---|
|
|
是 |
— |
场景的唯一标识符。 |
|
|
否 |
|
描述场景的可选元数据。可用于组织和识别结果中的场景。 |
|
|
是 |
— |
演员的身份和目标。请参阅演员简介。 |
|
|
是 |
— |
发送给您的代理人以开始对话的第一条消息。 |
|
|
否 |
10 |
对话停止前的最大回合数。必须至少为 1。 |
|
|
否 |
— |
关于预期行为的自然语言断言。由会话级别的评估者使用,例如。 |
注意
由于事先不知道会话流程,expected_response因此模拟场景不支持expected_trajectory每回合。使用assertions模拟场景了解真实情况。
FileDatasetProvider从 JSON 结构中自动检测场景类型:带有actor_profile字段(且没有字turns段)的场景加载为SimulatedScenario。
与批处理数据集运行器一起使用
以下示例使用批处理数据集运行器运行模拟场景评估。设置simulation_configBatchEvaluationRunConfig并在数据集中包含SimulatedScenario实例:
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}")
与按需数据集运行器一起使用
按需数据集运行器遵循相同的模式。设置simulation_configEvaluationRunConfig并在数据集中包含SimulatedScenario实例:
注意
On-demand 评估是根据消耗量收取费用的。有关详细信息,请参阅定AgentCore 价页面
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')})")
停止条件
当满足以下任何条件时,模拟对话结束:
-
目标已完成:行动者确定其目标已实现并发出信号
stop: true。这是预期的结果。 -
达到最大回合数:对话达到
max_turns极限。这可以起到安全保障的作用。如果你的场景经常达到回合限制,可以考虑增加max_turns或简化演员的目标。 -
未生成任何消息:该角色不生成下一条消息,但没有明确表示停止信号。这被视为隐含的目标完成。
有效模拟场景的技巧
-
明确目标:像 “聊一聊” 这样的模糊目标会导致互动不集中。诸如 “获得订单 #12345 的退款” 之类的具体目标为演员提供了明确的终点。
-
使用特质来控制难度:有特征的演员比有特质的演员
"expertise": "expert"提出更难的问题"expertise": "novice"。使用特征在不同的用户群中测试您的代理。 -
设定现实的回合限制:大多数客户支持对话会在 5 到 10 个回合内解决。设置
max_turns得太高会浪费计算;将其设置得太低可能会在目标实现之前中断对话。 -
使用断言来了解基本事实:由于对话流程是动态的,因此无法按回合
expected_response进行。写下断言来描述你所期望的结果,无论采取什么具体的路径。 -
选择合适的演员模型:演员模型应有足够的能力在各个回合中保持角色连贯性。较小的模型适用于简单的角色;具有细微目标的复杂角色受益于能力更强的模型。