在无服务器流水线中调用 Amazon Bedrock AgentCore 智能体的异步模式

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

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

了解成本实际落在哪一侧会有所帮助，因为调用的双方是分别计费的。Amazon Bedrock AgentCore 的运行时（Amazon Bedrock AgentCore 的一项能力）采用按量计费模式，在智能体空闲时不收取 CPU 费用。例如，在等待大语言模型生成响应，或等待工具调用、Model Context Protocol（MCP）调用返回期间，你只需为那段时间的内存付费，CPU 不收费。调用智能体的计算服务则没有这种机制。发出同步调用的 Lambda 函数、容器或 Amazon Elastic Compute Cloud（Amazon EC2）实例会一直阻塞，直到智能体响应。它占用（并付费）其全部计算资源，直到智能体返回。因此浪费不在智能体侧，而在调用方——在开放连接上空等。

这就导致调用方的成本与智能体的运行时间挂钩。阻塞在智能体上的函数，其计费时长基本上等于智能体的处理时间，而启动智能体后立即返回的函数，只为短暂的调度过程付费。解决方案是在等待期间释放调用方的计算资源，只在智能体有结果时才恢复流水线。在这篇文章中，我们展示三种实现这一点的模式（任务令牌回调、直接服务集成和持久函数），并将它们与阻塞式反模式进行对比。

为了在同等条件下比较各模式，我们让每个模式都经历相同的流水线，只改变调用智能体的那个步骤。流水线故意设计得简单，是一个虚构的场景（为房地产融资文档验证），目的是保持编排逻辑清晰。这不是文章的重点。它代表的是任何调用智能体（或另一个慢速服务）后对结果执行操作的 workflow，所以用你自己的用例来替换它即可。

流水线有五个阶段：

提取（Extract）：AWS Lambda 函数对文档执行光学字符识别（OCR）和文本提取。（提取是模拟的，因此该场景可以在没有真实文档的情况下运行。）

识别（Identify）：Lambda 函数对文档进行分类并设置路由标志（shouldOrganize、shouldValidate）。

路由（Route）：Choice 状态根据这些标志引导流程。

组织和验证（Organize and Validate）：Parallel 状态在组织文档的同时，在另一个分支中由 Amazon Bedrock AgentCore 智能体对其进行验证。这个 Validate 分支是唯一在模式之间会发生变化的部分。

结果（Result）：Lambda 函数处理智能体的裁决，决定下一步操作（批准，或退回修正）。

下图展示了该流水线。它在每种情况下都保持不变。只有高亮的 Validate 分支会被交换，以演示每种调用模式。

图 1：示例流水线。只有高亮的 Validate 分支在不同模式之间发生变化

单个 Amazon Bedrock AgentCore 智能体服务于所有四种情况。智能体检查每次调用并选择如何响应：如果收到 AWS Step Functions 任务令牌，就会在完成后唤醒该执行；如果收到持久函数回调 ID，就唤醒持久函数；如果两者都未收到，就将裁决直接放在响应中返回。这意味着你可以在不更改或重新部署智能体的情况下切换编排模式。

智能体如何在不阻塞调用方的情况下返回控制权

其机制是智能体动作组中的一个返回控制权动作。当智能体完成推理后，它调用一个 Lambda，该 Lambda 将结果和任务令牌一起发回 Step Functions。（模式 2 稍后描述，它通过让 Step Functions 直接与 AgentCore 集成来完全消除这个 Lambda。）

以下代码展示了这个 Lambda 的核心：

# 智能体得出裁决后调用的工具
@tool
def conclude_validation(approved: bool, issues: list, summary: str) -> str:
    verdict = {"approved": approved, "issues": issues,
               "summary": summary, "source": "agentcore"}

# 传入了 Step Functions 任务令牌：恢复该执行
    if task_token:
        sfn.send_task_success(taskToken=task_token, output=json.dumps(verdict))
        return "Step Functions resumed."

# 传入了持久函数回调 ID：恢复持久函数
    if callback_id:
        lambda_client.send_durable_execution_callback_success(
            CallbackId=callback_id, Result=json.dumps(verdict).encode("utf-8"))
        return "Durable function resumed."

# 两者都未传入：这是同步调用，内联返回裁决
    return "Verdict recorded."

入口点根据相同的信号决定是以后台模式还是同步模式运行：

@app.async_task
async def validate_document_async(prompt, document, extracted_text):
    # 后台工作；conclude_validation 完成后触发正确的回调
    agent = build_agent()
    await agent.invoke_async(message(prompt, document, extracted_text))

@app.entrypoint
async def handler(event):
    task_token  = event.get("taskToken")    # 由任务令牌模式传入
    callback_id = event.get("callbackId")   # 由持久函数模式传入

# 异步：启动工作并立即返回"已接受"
    if task_token or callback_id:
        asyncio.create_task(validate_document_async(...))
        return {"status": "accepted"}

# 同步：立即运行并在响应中返回裁决
    agent = build_agent()
    await agent.invoke_async(message(...))
    return verdict

智能体就位后，文章其余部分聚焦于调用它的四种方式。

调用智能体：四种方案

我们从阻塞式反模式开始，建立基准成本，然后展示三种避免该问题的模式。全程的代码和基础设施定义均来自示例，用于说明每种模式。

阻塞式反模式

最直接的实现是在同一个 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：带调度函数的任务令牌回调

这种模式在路径中保留 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 };

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

Cost。Lambda 函数会运行，但只运行到启动智能体并返回为止：只需几秒，无论智能体后续花费多长时间。你只需为那段短暂的调度付费，而无需为等待过程付费，因为函数在智能体工作期间已经关闭。等待由暂停的 Step Functions 执行持有，不会为闲置计算资源计费。这正是阻塞版本的关键区别——在阻塞版本中，函数计费时间会跟踪智能体的处理时间。

模式 2：直接服务集成

当你不需要在智能体调用周围添加自定义代码时，可以移除调度器函数，让 Lambda 完全离开这条路径。Step Functions 可以通过其 AWS SDK 服务集成直接调用 Amazon Bedrock AgentCore，因此智能体的响应直接流入下一个状态。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"
}

Cost。路径中没有 Lambda 函数，因此无需为闲置 Lambda 计算资源付费。Step Functions 持有等待状态，而 Standard 工作流按状态转换计费，而非按等待持续时间计费，因此在处理过程中有意义的成本就是智能体本身的费用。

模式 3：Lambda 持久函数

如果你更希望将编排表达为集中在一处的代码，Lambda 持久函数可以提供相同的成本行为。使用 @aws/durable-execution-sdk-js SDK 时，管道阶段变为 context.step 调用，并行工作变为 context.parallel，等待智能体响应变为 context.waitForCallback。在等待期间，函数被挂起，不计算计算资源费用。智能体使用 SendDurableExecutionCallbackSuccess 恢复它。

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

Cost。一个函数承载整个管道，但在挂起等待智能体期间不计算计算资源费用。你只需为暂停之间的短暂执行时段付费，这与任务令牌模式的经济学相同，而非等待时间本身。

测量差异

关键不在于某个具体数字。智能体的运行时间会因提示词、模型和文档而异。重要的是任务令牌模式中两个值之间的关系：Validate 状态活跃了多长时间，而调度器函数实际被计费了多长时间。我们测试中的一次运行使这种关系清晰可见：

Step Functions, ValidateDispatch state
   Returned (TaskSubmitted): 14:08:19   <- 函数返回并关闭
   Resumed  (TaskSucceeded): 14:08:34   <- 智能体唤醒执行
   State active ........ 19.6s

Lambda, dispatcher function (CloudWatch REPORT)
   Billed duration ..... 4.8s

结果：状态活跃了 19.6s，但函数仅被计费 4.8s。
其间的约 14.8s 是没有 Lambda 函数运行的等待时间。

你可以在 Step Functions 事件历史中看到同样的情况：使用任务令牌模式时，TaskSubmitted 事件（函数已返回）与 TaskSucceeded 事件（智能体恢复了流程）之间被智能体的处理时间隔开。同步调用则没有这种间隔。

下表总结了在每个模式中如何实现 Validate 分支（在上图中高亮显示）：

阅读时需要注意一点：整个管道持续时间不是一个有用的比较指标，因为它主要由智能体自身的推理时间决定，而推理时间每次运行都不同，在所有场景中基本相同。有意义的区别在于等待期间你为多少计算资源付费，这由表格和计费时长读数来体现。调度器的计费时间保持不变，而智能体的运行时间增长。前面的数字来自单次运行。请将其作为关系说明，而非基准测试，并测量你自己的工作负载。

下表总结了各种权衡，以帮助你选择模式：

防护措施

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

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

开启 AWS X-Ray。在 Step Functions 和 Lambda 配置中启用 Tracing: Active。X-Ray 准确显示智能体花了多长时间思考 vs 调用者花了多长时间等待，确认你的模式在间隔期间确实释放了计算资源。

为速度而非智能体的工作负载调整调度器函数大小。调度器只负责序列化请求并调用端点。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 控制。