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