将Agent工作负载迁移到Amazon Bedrock AgentCore实战指南

在笔记本里能跑的 AI 智能体，到了生产环境就不是同一个 AI 智能体了。当真实用户进来后，你需要操心的事情跟 AI 智能体本身的推理毫无关系。要防止一个用户的会话混入另一个用户的会话，要在多轮对话和跨天场景下保持状态。每次 AI 智能体调用工具时的鉴权都要写进你的代码，底层的操作系统也要打补丁。这只是本文所列十个运维负担中的四个。

AI 智能体进入生产环境后，需要加入 Amazon Bedrock Guardrails 来过滤有害内容、验证是否基于源文档进行 grounding，以及拦截提示词注入攻击。无论你在哪个阶段停下来，这些控制措施对任何 AI 智能体都适用。

本文从一个已有的 AI 智能体出发。它是一个 LangGraph 客服 AI 智能体，负责对每条消息进行分类：愤怒的客户走升级通道，其他所有人则由三个工具来回答。它调用的模型已经对接 Amazon Bedrock。容器、Web 服务器以及进程内的会话状态都由你自己管理。推理是迁移过程中唯一不碰的调用，所以"已经在用 Amazon Bedrock"听起来不算是先发优势——如果你的模型调用是直连 OpenAI 或 Anthropic，那只需要改一个构造函数，这在第 0 阶段会讲到。

本文分两个阶段迁移该 AI 智能体。第一阶段将其迁移到 Amazon Bedrock AgentCore Runtime、Gateway 和 Memory，图结构保持不变。第二阶段将其重建为基于 Strands Agents 的模型驱动规划。停在第一阶段会得到一个托管 AI 智能体，具备托管工具和持久状态。第三阶段将循环交给 AgentCore harness，这是 Amazon Bedrock AgentCore 的能力，本文仅作文档记录，不做代码实现。

本文中的 AI 智能体负责回答支持问题。客户问订单在哪里，或者怎么退货，AI 智能体负责查询、能答的答、不能答的升级。它运行在你提供、维护和扩展的计算资源上。

最后这句话就是本文要讨论的核心。上述描述中没有一条是在说 AI 智能体本身在做什么。

从代码角度看，迁移的范围是明确的，只涉及四个构造体。你需要运维的东西则多得多，下一节将它们一一列出。

最后一行回答的是"AgentCore 帮我卸下了什么"。答案是：没有。Runtime 是你的 AI 智能体运行的地方，而不是决定下一步的地方。手写的分支逻辑是在第二阶段才选择放弃的。

迁移的目的是甩掉那些跟 AI 智能体推理本身无关的工作。Amazon Bedrock AgentCore 是一个平台，用于大规模构建、连接和优化 AI 智能体，支持任意框架或模型。它的各项服务可以逐个接入，每次接入都会卸下特定的负担，下图将这十项运维负担对应到各项服务上。Runtime 接管计算资源，所以操作系统补丁、自动扩缩容和会话隔离不再是你的事。默认情况下它运行在 AWS 托管的基础设施上，你也可以将它挂载到你拥有的虚拟私有云（VPC）上。无论哪种方式，你仍然需要设计网络、在入口前方部署边缘防护，以及决定授权策略。Gateway 接管工具鉴权，用它自己的执行角色调用你的函数。检查点存储交给 Memory，它在多轮对话、处理过程和跨天场景下保持会话状态。

AWS Identity and Access Management（IAM）策略、VPC 配置、Web 应用防火墙（WAF）规则以及密钥轮换在每个阶段都属于你负责的范围。依赖更新要到第三阶段才会迁移，之前任何阶段都不会。

还有三项服务接入后不会替换任何已有组件。Identity broker 负责凭证管理，并为 AI 智能体代表用户调用的 API 刷新 OAuth 访问令牌。本教程不涉及此项演练。AI 智能体使用自己的 IAM 凭证通过 Signature Version 4 对 Gateway 调用进行签名，Gateway 再以自己的执行角色调用 AWS Lambda 目标。这一路径中不需要第三方令牌。策略在 Gateway 层决定每个工具调用的权限。观测性服务将 Runtime 的日志、指标和追踪发送到 Amazon CloudWatch，无需你额外配置。

按阶段顺序推进。第一阶段迁移 AI 智能体的运行位置，不改变它的思维方式，这样只有一个变量在变化。第二阶段迁移它的规划方式，而运行时你已经验证过可行。如果你有团队要重写 AI 智能体，可以直接从第二阶段开始，因为 Gateway、目标和 Memory 存储先建好之后可以同时服务于两个阶段。第三阶段只提供文档，不做代码实现。

图 1：哪些运维工作不再需要你负责，以及谁来规划 AI 智能体的下一步。第一阶段迁移前者，第二阶段迁移后者

迁移教程

示例代码仓库按以下阶段组织，这样每个阶段都可以跟上一个阶段对比。下图展示了这种对比的形态：每个阶段从哪里出发、迁移了什么，以及为什么下一个阶段会那样演进。本文给出的迁移量统计，都是基于已提交的示例而非估算值。

图 2：每个阶段从哪里出发、迁移了什么，以及为什么下一个阶段会那样演进

你需要拥有一个 AWS 账户并启用 Amazon Bedrock 模型访问权限，需要 Python 3.12，以及配置好能够创建 AgentCore、Lambda、Amazon S3 和 IAM 资源的 AWS CLI 凭证。还需要为账户启用一次 CloudWatch Transaction Search，否则本教程产生的追踪记录将无法查看。

git clone https://github.com/aws-samples/sample-migrate-agents-to-amazon-bedrock-agentcore.git
cd sample-migrate-agents-to-amazon-bedrock-agentcore
./setup.sh

该脚本会创建一个虚拟环境并安装七个依赖项。如果你已经有一个对接 Amazon Bedrock 的 LangGraph AI 智能体，新装的依赖项是 strands-agents、bedrock-agentcore、mcp 和 langgraph-checkpoint-aws。其中两个是固定版本而非最低版本要求，因为不锁定版本的 langchain-aws 会解析出更高版本，并连带把 boto3 往前拖。

用测试套件确认安装成功，测试不需要凭证：

source .venv/bin/activate
python -m unittest discover -s tests -q

第 0 阶段：已有的 AI 智能体

在动手修改之前先读一遍 AI 智能体，因为当前的行为是后续每个阶段都必须保留的基线。它是一个编译后的 StateGraph：classify_intent 让模型输出一个词，然后一段手写的 route_intent 读取它。愤怒的客户走 escalate 节点，该节点返回一个固定的移交信息，不做任何模型调用。其他所有人都走 assist 节点，该节点用绑定的工具调用模型：

builder.add_edge(START, "classify_intent")
builder.add_conditional_edges(
    "classify_intent",
    route_intent,
    {"escalate": "escalate", "assist": "assist"},
)
builder.add_edge("escalate", END)
builder.add_conditional_edges("assist", tools_condition)
builder.add_edge("tools", "assist")

return builder.compile(checkpointer=checkpointer)

它挂载了三个工具，以 @tool 装饰器函数的形式对接一个 HTTP 后端：lookup_order、process_return 和 search_faq。它们返回 {"error": ...} 载荷而不是抛异常，因为工具节点内的异常会终止整个运行，而错误载荷是模型可以处理的内容。

状态通过 MemorySaver checkpointer 保存，以 invoke 时传入的 thread_id 为键，这是唯一一个存在硬性限制的部分。进程内的字典会随进程一起消亡，两个副本之间无法互相看到对方的对话。这里其他所有东西在生产规模下都没问题。只有这个不是。

模型是 ChatBedrockConverse，所以推理已经走 Amazon Bedrock，第 0 阶段不涉及任何 AgentCore API。如果你从 OpenAI 或 Anthropic 过来，那个构造函数就是你唯一要改的地方，参数是模型 ID 和 AWS Region。关于各 Region 支持的模型，参考 Amazon Bedrock 中的 Supported models by AWS Region。在迁移任何东西之前先记录基线：记录哪些工具被调用了，以及图状态中的最终消息，而不是模型的回复。这样下一阶段就是一个对比而不是猜测。以下命令在本地运行，不会在 AWS 创建任何资源：

python -m examples.run_walkthrough --stage 0

第 1 阶段：同一 AI 智能体，已迁移

第 0 阶段给你留下了一个可工作的 AI 智能体和一份记录的基线。第一阶段是十个运维负担中有五个不再由你负责，而 AI 智能体的行为是唯一不变的东西。三样东西迁移了：Runtime 接管进程，三个工具中的两个放到 Gateway 后面，会话状态存入 Memory。推理没有迁移，因为它从来就不是问题。

迁移后的包不复制第 0 阶段的代码，而是导入它：

from examples.stage0_langgraph.agent import build_graph
from examples.stage0_langgraph.tools import SUPPORT_TOOLS

这两个 import 携带着图的拓扑结构、路由器、状态模式、全部三份 prompt 和全部三个工具本体，所以没有任何部分会漂移。它们也让成本可计数——从磁盘测量而非靠断言：Agent 内部改了 45 行，新增了 22 行 SDK 未附带的支撑代码，还有 85 行原样导入。那 22 行很少的原因在于：原来最贵的部分是一段手写的 LangGraph checkpointer，接在 AgentCore Memory 上，而现在它已经打包发布了，所以剩下的胶水代码只剩下一个工具适配器。

所有十项运维负担仍然由你承担，基线已记录在三次对话轮次中。循环之下的机器优先运行，因为给操作系统打补丁不是你的 Agent 做的事。这也是你改动最少代码却能获得最大回报的地方：操作系统不再需要你来打补丁，Session 隔离变成每个 Session 一个微 VM。把已有的循环包装进 BedrockAgentCoreApp 并给它一个入口点：

from bedrock_agentcore import BedrockAgentCoreApp
from langchain_core.messages import HumanMessage

app = BedrockAgentCoreApp()

@app.entrypoint
def agent_invocation(payload, context):
    state = support_graph().invoke(
        {"messages": [HumanMessage(payload.get("prompt", ""))]},
        config={"configurable": {"thread_id": context.session_id or "local-session"}},
    )
    return {"result": state["messages"][-1].text}

if __name__ == "__main__":
    app.run()

其中的 invoke 调用就是阶段 0 的那个。变化在于 thread_id 从哪来。阶段 0 自己选了一个。这里它作为 context.session_id 到达，是 Runtime 传入的 RequestContext 的一部分。只构建一次图并保持住，因为每次请求重建图会重建模型客户端，还可能拆除工具所需的 MCP Session。

同样的包装器可以接管 CrewAI 或 LlamaIndex 的循环，或者你自己写的循环：接收一个 payload 字典，调用它，返回一个字典。

循环之下的操作系统和计算资源已经不再归你管了。但工具认证没有。工具放在网关背后的原因在于它随动的东西：认证不再是你的代码要操心的事，触达范围也扩大了，因为一个发布出去的工具有可能被下一个 Agent 调用，而一个进程内部的函数永远只是这个 Agent 的。所以 lookup_order 和 process_return 变成了由 AgentCore Gateway（Amazon Bedrock AgentCore 的一项能力）通过一个 Lambda 函数发布的 MCP 工具。你注册它，但不需要改动它内部的任何代码。

search_faq 在每个阶段都保持为本地 Python 函数，这是正常情况而非妥协。别的 Agent 不需要、也没有策略门控的工具，走这一趟没有任何收益。

转换是对 bedrock-agentcore-control 客户端的两调用。创建网关，选择一个 authorizerType。AWS_IAM 用你已有的凭证签名，CUSTOM_JWT 需要一个 bearer token。

client = boto3.client("bedrock-agentcore-control", region_name=region)

gateway = client.create_gateway(
    name="MigratedAgentGateway",
    roleArn=role_arn,  # gateway execution role
    protocolType="MCP",
    authorizerType="AWS_IAM",
)
gateway_id = gateway["gatewayId"]
gateway_url = gateway["gatewayUrl"]

两个返回值都在后面有用：gateway_id 在注册目标时命名网关，gateway_url 是 Agent 连接到的 MCP 端点。

然后注册一个目标，这就是转换本身。让 Gateway 指向你的函数，并用 JSON schema 声明它的工具。

response = client.create_gateway_target(
    gatewayIdentifier=gateway_id,
    name="supportTools",
    targetConfiguration={
        "mcp": {
            "lambda": {
                "lambdaArn": lambda_arn,
                "toolSchema": {"inlinePayload": TOOL_SCHEMA},
            }
        }
    },
    credentialProviderConfigurations=[
        {"credentialProviderType": "GATEWAY_IAM_ROLE"}
    ],
)

使用 GATEWAY_IAM_ROLE 时，Gateway 以自己的身份调用你的函数，并把工具参数作为原始事件传递，所以需要一个 Amazon API Gateway 处理器来适配。

Agent 发现工具，对每个请求做 SigV4 签名：

import boto3
from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp import MCPClient

from examples.tools.gateway_mcp_tools import SigV4HTTPXAuth

auth = SigV4HTTPXAuth(boto3.Session().get_credentials(),
                      "bedrock-agentcore", region)
mcp_client = MCPClient(lambda: streamablehttp_client(gateway_url, auth=auth))
mcp_client.start()  # not a `with` block: held for the process lifetime
tools = mcp_client.list_tools_sync()

发现到的工具到达时带上了目标名称和三个下划线的前缀。lookup_order 变成 supportTools___lookup_order，在列表中搜索原始名称则什么都找不到。

list_tools_sync() 返回的是 Strands 工具对象，而 LangGraph 工具节点需要的是 LangChain BaseTool 对象，两者没有共享接口。那 22 行就在两者之间做转换。同一个模块合并两个来源，按最后一个 ___ 后面的部分匹配，所以 Gateway 工具会覆盖同名本地函数，而 search_faq 保持本地。

随着计算和工具认证的迁移，对话状态成为这个阶段剩余的边界。它仍然随进程消亡，所以两个实例无法读取同一个对话。Memory 消除了这个限制。阶段 0 的 MemorySaver 变成一个由 AgentCore Memory 支持的检查点程序，用 actor_id 和 session 而非你选的 thread_id 作为键。先创建存储，并慎重设置 event_expiry_days，因为检查点会继承它。

检查点程序是第一方的，是一个依赖项而非你拥有的文件。它就在你已安装的 requirements 里，是带版本锁定的：

langgraph-checkpoint-aws==1.2.1

用它来构造它，只需要 memory id，不需要别的。它不接收 actor_id。它在每次调用时从 RunnableConfig 读取 actor_id 和 thread_id，而不是在构造时绑定任何一个，所以两者随每次调用一起传递：

from langgraph_checkpoint_aws import AgentCoreMemorySaver

graph = build_graph(
    llm=llm,
    tools=tools,
    checkpointer=AgentCoreMemorySaver(memory_id, region_name=region),
)

state = graph.invoke(
    {"messages": [HumanMessage(prompt)]},
    config={"configurable": {"thread_id": session_id, "actor_id": actor_id}},
)

测试持久性，不要假设它。第二个进程只共享 memory、actor 和 session id，就能回答一个从未被告知的订单问题。但不要对事件计数做断言。相同代码的两次运行产生了不同的总数。

你获得了跨实例共享的持久对话状态，同步和异步都支持。saver 还实现了 list 和 delete_thread。历史记录和时间旅行通过 LangGraph 已经在用的接口工作。

先在本地证明这个模块，因为那里的失败是可读的，在 Runtime 内部则不然。app.run() 提供的是同一个 POST /invocations 契约，Runtime 调用的就是这个。然后用 codeConfiguration 调用 CreateAgentRuntime，它是你在 Amazon S3 中源代码的 zip 包，依赖项 vendors 在旁边。没有容器，没有 Amazon Elastic Container Registry（Amazon ECR），没有 Docker。如果你在估算这次迁移的工时，这个句子就是估算值：pip 是部署唯一需要的构建工具。

有两个陷阱会耗费真实时间，而且两个错误的提示都没有写出原因。首先，在笔记本上 pip install -t 安装的是笔记本的 wheel，而 Runtime 是 ARM64（Advanced RISC Machine 64-bit）Linux。为目标应用和部署指定的 Python 版本 vendors，这样 wheel 和运行时环境就一致了：

pip install -r requirements.txt -t build/ \
    --platform manylinux2014_aarch64 --python-version 3.12 --only-binary=:all:

这里的 3.12 是部署目标，不是你本地的解释器。

其次，zip 内部的 requirements.txt 是无效的。归档包就是最终环境。漏了一个依赖，失败信息会显示 Runtime initialization time exceeded ... 30s，而不是发生的 ModuleNotFoundError。把依赖 vendors 进去，不要调超时。

接下来的命令会创建真实的 AWS 资源并开始计费。最后的清理（Clean up）部分会删除本演练创建的所有资源。

然后用你记录的基线做验证。阶段 1 需要两个 ARN，一个脚本创建网关目标背后的 Lambda 并打印两者：

./examples/gateway/lambda_target/deploy.sh
python -m examples.run_walkthrough --stage 1 \
    --role-arn <gateway-role-arn> --lambda-arn <tools-function-arn>

阶段 1 打印出与阶段 0 相同的每轮输出，所以检查是一个 diff，而非对话文本的评判。示例对 diff 做了断言，而不是交给你的眼睛：test_stage1_replay.py 要求网关调用以 supportTools___lookup_order 到达，携带 {"order_id": "12345"}，所以重命名的工具或丢弃的参数会导致运行失败，而不是通过目视检查。

当智能体做了你意料之外的事时，你需要看到它实际做了什么。通常这意味着你要自己负责仪表化：一个追踪包、环境变量、一个收集器来运行。Runtime 托管它所运行的智能体，而 AgentCore Observability（Amazon Bedrock AgentCore 的一项功能）则将结果发送到 CloudWatch。需求中没有追踪包，也不需要设置 OTEL_* 变量。日志组在首次调用后自动出现，一旦启用 Transaction Search，span 就会在 CloudWatch 中显示出来。

Figure 3: 客户端直接调用 Runtime，图形被导入而未被重写，三个工具中有两个移到了 Gateway 后面

这就是第一阶段：同一个智能体给出相同的答案，十项负担中的五项现在由 AgentCore 处理。谁来规划下一步没有改变，而这个问题是第二阶段要解决的。

第二阶段：重构循环，因为你选择了这样做

十项运营负担中有五项已经移走，还有五项留在你手里：VPC 配置、WAF、IAM 策略、密钥轮换和依赖更新。第二阶段不会移动任何一项，因为这个阶段改变的是谁来规划下一步。当手写的分支达到上限时——当 route_intent 是你不断编辑的文件，新的意图意味着一个新节点而不是在提示中加一行——就进入这个阶段。

模型驱动的编排取代了分支。第零阶段的 classify_intent 节点和 route_intent 在 Python 中决定下一步。Strands 智能体将这个决策交给模型，所以 add_conditional_edges 没有对应的实现。这既是收获也是损失。分支是确定性的和可审计的，而模型的规划既不是确定性的也不是可审计的。

bedrock-agentcore SDK 为 Strands 智能体附带了一个会话管理器，所以 wiring 就是一个配置对象和一个构造函数参数：

config = AgentCoreMemoryConfig(
    memory_id=memory_id, session_id=session_id, actor_id=actor_id
)
kwargs["session_manager"] = AgentCoreMemorySessionManager(
    config, region_name=region_name
)

return Agent(**kwargs)

会话管理器与模型、系统提示和工具一起通过 Agent(**kwargs) 进入，它承载了第一阶段的三个 id，其中 thread_id 现在改名为 session_id。

第二阶段复用了第一阶段的 gateway、target 和 Memory store。它不复用第一阶段的 runtime，因为其中的程序是不同的循环。Amazon Bedrock 也没有迁移到这里。每个阶段都是同一个模型。以相同的两个 ARN 运行它：

python -m examples.run_walkthrough --stage 2 \
    --role-arn <gateway-role-arn> --lambda-arn <tools-function-arn>

第二阶段将规划交给了模型，失去了可审计的分支。Amazon Bedrock AgentCore 中的 Policy 在每个工具调用前放置了一个确定性决策：Cedar 规则通过 gateway 在数据平面上对每次调用进行评估，应用程序无法绕过这条路径。模型输出的护栏不在那条路径上，等到它运行时，工具调用已经发生了。

Policy 附加在 Gateway 上而不是循环上，所以这在第一阶段未修改的智能体上也能工作。它出现在这里是因为示例在这里运行它。

示例附带了两条规则：只读身份可以调用 lookup_order，特权身份还可以调用 process_return。没有任何 forbid 规则。Cedar 默认拒绝，所以只读调用者的拒绝就是没有匹配到的 permit。来自 IAM，这是需要改掉的习惯。

两个调用者角色用相同的 IAM 策略创建，这就是为什么 enforcement 可以证明是 Cedar 的而不是 IAM 差距。相同的权限、相同的 gateway、两个调用者对两个工具、一个拒绝。

Figure 4: 循环变成了模型的，内存 wiring 收缩成一个 SDK 自带的会话管理器，Cedar 在 Gateway 处决定每个工具调用

循环现在是模型的了，但它仍然作为你的代码发货。第三阶段是移除最后一块的样子。

第三阶段：把循环交出去

有一个负担仍然与拥有代码绑定在一起，这是唯一移动它的阶段。AgentCore harness 为你运行循环，由 Strands Agents 驱动。你将智能体声明为配置（模型、系统提示、工具、内存和限制），然后 AWS 运行它，所以切换模型是配置变更而不是重新部署。如果你想要这个 harness，你运行的是第二阶段的智能体。它托管一个单一的模型驱动循环，而不是一个图形，所以图形形状的智能体要先变成那个循环才能到达它。这是推荐顺序也是唯一顺序的唯一地方。

图将那一列标记为已记录而非已测量。十项负担中有六项移到那里，而第二阶段是五项，依赖更新是唯一移到这里的一项，因为智能体不再是你的代码了。密钥轮换不在其中。Identity 刷新令牌而不是轮换背后的密钥，所以那个负担在第三阶段也仍然是你的。

迁移改变的不仅仅是基础设施。这些是迁移后团队花费最多时间的模式。

1. 假设功能 parity

假设 parity 会引发一场无人能终结的争论。迁移后你的智能体行为不会完全相同，如果没有在迁移前商定标准，每个措辞上的差异都会成为那场争论的导火索。根据结果而非实现来定义验收标准，然后测试它们：90% 的订单状态查询在无需升级的情况下得到解决，5 秒内响应。AgentCore Evaluations（Amazon Bedrock AgentCore 的一项功能）有内置的评估器。

2. 在进程中保存状态

这一项是在生产环境中付出代价的，因为一个走开的用户，因为没有快速测试空闲时间足够长到能捕获它。会话不是调用：同一个会话包含一次又一次的调用，而 Runtime 在 15 分钟不活动后默认终止执行环境，为同一个会话重新配置一个新的。那个空闲窗口是 idleRuntimeSessionTimeout，可以设置为 60 秒到 8 小时，所以调优它只是移动了截止日期而不是移除它。会话存活了，你的内存状态没有，而症状是状态只在安静期后消失。像第一阶段那样保存图形，但将下一个回合需要的每个事实都保存在 Memory 中。

3. 忽略认证架构

认证差距会造成返工，而不是配置问题：需要用户委托访问的工具出现得很晚，会改变调用路径。所以先映射每个流程：你的智能体如何认证到外部 API，人们如何认证到智能体，你如何划分权限范围。AgentCore Identity（Amazon Bedrock AgentCore 的一项功能）回答第一个，在你的——时引用