在无服务器流水线中异步调用 Amazon Bedrock AgentCore 代理的模式

在无服务器流水线中，对 Amazon Bedrock AgentCore 代理的异步调用模式，可以在你的 AI 代理处理请求期间消除空闲计算成本。一个常见的例子是文档验证：在房地产融资后台，代理可以读取房产记录或贷款合同，推理信息是否完整一致，并返回下游步骤据此操作的判定结果。Amazon Bedrock AgentCore 提供了一个平台，可以使用任何框架或模型，大规模地构建、连接和优化代理。

这些代理引入了一个传统流水线步骤所不具备的特性：它们在回答之前会思考一段时间。思考时间取决于提示词、模型和文档，但很少是即时的，而这种延迟会改变你调用它的方式。最常见的初始实现是使用计算服务（例如 AWS Lambda 函数）来调用代理并等待响应。当该函数等待时，它什么也不做，但它仍在运行，并且你需要为它的每一秒付费。

有必要弄清楚成本究竟落在哪里，因为调用的双方计费方式不同。Amazon Bedrock AgentCore runtime（Amazon Bedrock AgentCore 的一项功能）采用基于消耗的模型，在代理空闲时不收取 CPU 费用。例如，当它等待大型语言模型生成响应，或等待工具或模型上下文协议（MCP）调用返回时，你只需为这段时间的内存付费，而无需为 CPU 付费。调用代理的计算服务则没有这种行为。发出同步调用的 Lambda 函数、容器或 Amazon Elastic Compute Cloud（Amazon EC2）实例会处于阻塞状态。它会持有（并为其支付）全部计算资源，直到代理响应。因此，浪费并不在代理端，而是调用方在空闲地占用着一个打开的连接。

这使得调用方的成本与代理的运行时间挂钩。一个阻塞在代理上的函数，其计费时间基本上等于整个处理时间，而一个启动代理后立即返回的函数，则只需为短暂的调度过程付费。解决方案是在等待期间释放调用方的计算资源，并且仅在代理产生结果后才恢复流水线。在本文中，我们展示了实现此目的的三种模式（任务令牌回调、直接服务集成和持久函数），并将它们与阻塞式的反模式进行对比。

示例流水线

为了在同等条件下比较这些模式，我们让每种模式都运行在同一个流水线中，并且只更改调用代理的那一步。该流水线是一个刻意简化的虚构场景（用于房地产融资的文档验证），选择它是为了保持编排的清晰性。它不是本文的重点，它代表任何调用代理（或其他慢速服务）然后根据结果采取行动的工作流，因此请将你自己的用例代入其中。

1. 提取：一个 AWS Lambda 函数对文档执行光学字符识别（OCR）和文本提取。（提取是模拟的，因此该场景无需真实文档即可运行。）
2. 识别：一个 Lambda 函数对文档进行分类并设置路由标志（shouldOrganize、shouldValidate）。
3. 路由：一个 Choice 状态根据这些标志来引导流程。
4. 整理与验证：一个 Parallel 状态在整理文档的同时，在另一个分支中，Amazon Bedrock AgentCore 代理对其进行验证。这个验证分支是唯一在不同模式间变化的部分。
5. 结果：一个 Lambda 函数处理代理的判定并决定下一步操作（批准，或退回修改）。

下图展示了该流水线。它在每种情况下都保持不变，只有验证分支被替换以演示每种调用模式。

图 1：示例流水线。只有高亮显示的验证分支在不同模式间变化

一个 Amazon Bedrock AgentCore 代理服务于所有四种情况。该代理检查每次调用并选择如何响应：如果它收到 AWS Step Functions 任务令牌，它会在完成时唤醒该执行。如果它收到持久函数回调 ID，它会唤醒该持久函数。如果两者都没有收到，它会直接在响应中返回判定结果。这意味着你可以在不更改或重新部署代理的情况下更改编排模式。

代理如何在不阻塞调用方的情况下返回控制权

其机制是代理操作组中的一项控制返回操作。当代理完成推理后，它会调用一个 Lambda，将结果和任务令牌发布回 Step Functions。（后文描述的模式 2 通过让 Step Functions 直接与 AgentCore 集成，完全消除了这个 Lambda。）

有了智能体之后，本文的其余部分将重点介绍调用它的四种方式。

调用智能体：四种方式

我们首先介绍阻塞式反模式以建立基线成本，然后展示三种避免该模式的方案。全文中的代码和基础设施定义均摘自示例，用于说明每种模式。

阻塞式反模式

最直接的实现方式是在同一个 Lambda 函数中调用智能体并等待其返回答案。这种方式可行且实现简单，因此非常常见，但函数会在智能体思考的整个过程中保持运行状态。

// Lambda 函数在此处阻塞，直到智能体响应
const response = await agentcore.send(
  new InvokeAgentRuntimeCommand({
    agentRuntimeArn: AGENT_RUNTIME_ARN,
    payload: new TextEncoder().encode(JSON.stringify(payload)),
    runtimeSessionId: sessionId,
  })
);
// 函数在智能体思考的整个过程中保持运行并持续计费。

函数的计费时长最终约等于智能体的处理时间。接下来的三种模式消除了这种空闲成本，每种模式都做出了不同的权衡。特别是模式 2 使用了 Step Functions 对 AgentCore Harness（InvokeHarness）的优化集成，完全移除了 Lambda。

模式 1：带调度函数（dispatcher function）的任务令牌回调

该模式在路径中保留了一个 Lambda 函数以执行自定义逻辑，但消除了空闲成本。Step Functions 使用 waitForTaskToken 集成调用该函数，该集成会传递一个任务令牌并暂停执行。函数使用该令牌启动智能体，然后在几秒内返回。执行保持暂停状态，不产生计算费用，直到智能体使用该令牌调用 SendTaskSuccess 来恢复执行。

// 启动智能体，传递任务令牌，然后不等待直接返回
const response = await agentcore.send(
  new InvokeAgentRuntimeCommand({
    agentRuntimeArn: AGENT_RUNTIME_ARN,
    payload: new TextEncoder().encode(JSON.stringify({ ...payload, taskToken })),
    runtimeSessionId: sessionId,
  })
);
// 在此处返回并不会完成该步骤。Step Functions 保持暂停状态，直到
// 智能体使用此任务令牌调用 SendTaskSuccess。
return { dispatched: true };

对应的状态从上下文中传递令牌，并设置超时和心跳作为安全网，这样静默的智能体就会干净地使执行失败，而不是使其无限期暂停：

"ValidateDispatch": {
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
  "Parameters": {
    "FunctionName": "${ValidateDispatcherFunctionArn}",
    "Payload": {
      "taskToken.$": "$$.Task.Token",
      "document.$": "$.document",
      "extractedText.$": "$.extract.extractedText",
      "executionId.$": "$$.Execution.Id"
    }
  },
  "TimeoutSeconds": 120,
  "HeartbeatSeconds": 60,
  "Next": "AgentCoreValidation"
}

成本。Lambda 函数会运行，但只运行到启动 agent 并返回为止：无论 agent 之后运行多久，都只需几秒钟。你只为这短暂的调度付费，而不是为等待付费，因为当 agent 工作时函数已经关闭。等待由暂停的 Step Functions 执行来承担，而它不会对空闲计算计费。这是与阻塞版本的关键区别——在阻塞版本中，函数的计费时间与 agent 的处理时间同步。

模式 2：直接服务集成

当你不需要在 agent 调用周围编写自定义代码时，可以移除调度函数，将 Lambda 完全从路径中剔除。Step Functions 可以通过其 AWS SDK 服务集成直接调用 Amazon Bedrock AgentCore，因此 agent 的响应会直接流入下一个状态。Validate 分支随后变成一个单一的 Task 状态：

"ValidateDirect": {
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:bedrockagentcore:invokeAgentRuntime",
  "Parameters": {
    "AgentRuntimeArn": "${AgentRuntimeArn}",
    "RuntimeSessionId.$": "States.Hash($$.Execution.Id, 'SHA-256')",
    "Payload.$": "States.JsonToString($.prep.agentInput)"
  },
  "ResultSelector": { "raw.$": "$.Response" },
  "TimeoutSeconds": 120,
  "Next": "ParseVerdict"
}

成本。路径中没有 Lambda 函数，因此没有空闲的 Lambda 计算需要付费。Step Functions 承担等待，Standard 工作流按状态转换计费，而不是按等待时长计费，因此处理过程中的实际成本来自 agent 本身。

模式 3：Lambda 持久函数

如果你更愿意用代码在一个地方表达编排逻辑，而不是使用状态机，Lambda 持久函数可以给你相同的成本行为。使用 @aws/durable-execution-sdk-js SDK，流水线各阶段变成 context.step 调用，并行工作变成 context.parallel，对 agent 的等待变成 context.waitForCallback。在等待期间，函数挂起，不计费计算资源。agent 通过 SendDurableExecutionCallbackSuccess 恢复该函数。

// 挂起函数，直到 agent 回调
const result = await ctx.waitForCallback(
  "validate-agentcore",
  async (callbackId) =>
    dispatchAgentCore(callbackId, document, extractedText, executionId),
  { timeout: { seconds: 120 } }
);

成本。一个函数承载整个流水线，但在挂起等待 agent 期间不计费计算资源。你只为挂起之间的短暂执行突发付费，这与 task-token 模式的经济性相同，而不是为等待付费。

衡量差异

重点不在于某个具体数字。agent 的运行时间随提示词、模型和文档而变化。关键在于 task-token 模式中两个值之间的关系：Validate 状态处于活动状态的时间，与调度函数实际被计费的时间。我们测试中的一次运行清晰地展示了这种关系：

Step Functions，ValidateDispatch 状态
   返回（TaskSubmitted）：14:08:19   <- 函数返回并关闭
   恢复（TaskSucceeded）：14:08:34   <- agent 唤醒了执行
   状态活动时长 ...... 19.6 秒

Lambda，调度函数（CloudWatch REPORT）
   计费时长 ...... 4.8 秒

结果：状态活动了 19.6 秒，但函数只被计费 4.8 秒。
中间的约 14.8 秒是等待时间，期间没有 Lambda 函数在运行。

你可以在 Step Functions 事件历史中看到同样的现象：使用 task-token 模式时，TaskSubmitted 事件（函数返回）与 TaskSucceeded（agent 恢复流程）之间隔着 agent 的处理时间。同步调用则没有这样的间隔。

关于这一点有个提醒：总流水线耗时并不是一个有意义的对比指标，因为它主要由智能体自身的推理时间决定，而推理时间每次运行都不同，并且在所有场景中基本一致。有意义的差异在于等待期间你为多少计算资源付费，这一点由表格和计费时长读数体现。调度器的计费时间保持平稳，而智能体的运行时间在增长。前面的数字来自单次运行。请将其视为关系的示意，而非基准测试，并自行测量你的工作负载。

选择模式

最佳实践

防范智能体永不响应的情况。在每个waitForTaskToken状态上设置TimeoutSeconds，使执行以States.Timeout失败，而不是无限挂起。如果你的智能体发送心跳，还要设置HeartbeatSeconds以更快检测智能体失联。捕获错误并将其路由到失败或人工审核路径。

在重试时使用稳定的会话ID。将sessionId设置为从执行上下文派生的值（例如Step Functions执行名称），以便重试时恢复同一智能体会话，而不是重新开始。在任务令牌模式中，在调度器中设置它。在直接集成模式中，在Task状态参数中设置它。

启用AWS X-Ray。在Step Functions和Lambda配置中启用Tracing: Active。X-Ray可以精确显示智能体思考花费了多长时间，以及调用方等待花费了多长时间，从而确认你的模式确实在等待期间释放了计算资源。

按速度而非智能体工作负载来配置调度器函数。调度器只序列化请求并调用端点。256 MB内存和30秒超时通常就足够了。繁重的工作发生在智能体侧。

成本

这些模式会产生Lambda计算、Step Functions状态转换、持久化函数执行存储的费用，以及每种方法都共有的Amazon Bedrock AgentCore运行时和Amazon Bedrock模型推理费用。智能体和模型成本在四种情况下相同。架构变化只影响编排开销，以及在阻塞反模式中浪费的空闲Lambda计算。有关当前定价，请参阅各服务的定价页面。

结论

将AI智能体放入流水线是简单直接的部分。经济地调用它才是区分原型与生产级设计的关键。阻塞在智能体上的Lambda函数实现简单，但隐性成本高，大部分计费时间都在等待。在本文中，我们展示了三种避免这种情况的方法：当路径中需要Lambda函数时使用任务令牌回调，最简单的情况下使用直接服务集成，以及当你更倾向于将编排作为代码时使用持久化函数。这三种方式都由同一个Amazon Bedrock AgentCore智能体驱动，它会根据调用方式调整自己的响应。

完整示例（包括完整智能体、状态机定义和持久化函数）可在GitHub上的sample-bedrock-agentcore-async-stepfunctions仓库中获取。如需深入了解，请参阅Amazon Bedrock AgentCore文档中关于异步任务处理的内容。

对于生产部署，请使用Amazon Bedrock Guardrails对智能体输入和输出实施负责任的AI控制。

关于作者