一个能在 Notebook 中回答问题的智能体，距离生产系统还差一整套运行基础设施：身份认证、工具接入、会话记忆、故障恢复、可观测性，以及可重复的部署流程。
迁移这类工作负载时，一次性重写通常会放大风险。更稳妥的路径分成两个阶段：先保留 LangGraph 的业务编排，把运行、工具和记忆迁移到 Amazon Bedrock AgentCore；待生产边界稳定后，再用 Strands Agents 引入模型驱动规划，逐步减少手写工作流节点。
第一阶段：保留工作流，替换运行基础设施
第一阶段的目标不是改变智能体如何思考，而是改变它在哪里运行、怎样调用工具，以及如何保存跨轮次状态。
可以把现有系统拆成四层：
| 层次 | 迁移前的常见实现 | 第一阶段的目标 |
|---|---|---|
| 编排 | LangGraph 节点、边和条件路由 | 暂时保留 |
| 运行 | Notebook、脚本或自建 API 容器 | AgentCore Runtime |
| 工具 | 直接调用内部 HTTP API | AgentCore Gateway |
| 状态 | 进程内对象、自建数据库或检查点 | AgentCore Memory |
这种拆分保留了已经验证过的客服流程。例如，“识别意图 → 查询订单 → 判断是否升级人工”的图结构可以继续工作，但节点不再直接持有数据库凭据，也不再依赖当前进程中的聊天记录。
Runtime：把一次实验变成可管理的执行单元
进入 Runtime 前，入口函数应当具备清晰的请求和响应契约。不要让部署入口依赖 Notebook 全局变量，也不要把 HTTP 请求对象一路传入业务节点。一个合适的边界至少包含：
session_id：标识一次连续会话。user_id：用于授权和记忆隔离，不能由不可信客户端任意冒充。message：当前用户输入。request_id：用于重试时去重。- 结构化响应：文本之外，还应允许返回工具调用状态、升级人工标记和错误码。
Runtime 托管了执行环境，并不意味着应用天然具备幂等性。订单取消、退款或工单创建等有副作用的操作，仍应使用 request_id 或业务幂等键防止重复执行。
Gateway：把工具调用变成受控接口
在 Notebook 原型中，工具往往只是一个调用内部接口的 Python 函数。生产环境需要进一步回答：谁能调用、参数是否合法、超时是多少、哪些字段可以进入日志、失败后是否允许重试。
迁移到 Gateway 时，应先整理工具契约，而不是把任意内部 API 原样暴露给模型。以订单查询工具为例，只开放客服任务需要的字段：
{
"name": "get_order_status",
"description": "Return the fulfillment status of an order owned by the authenticated customer.",
"input": {
"order_id": "string"
},
"output": {
"status": "processing | shipped | delivered | cancelled",
"estimated_delivery": "string | null"
}
}
工具描述应说明使用条件，但授权必须由服务端执行。模型决定“要调用哪个工具”，不能等同于模型决定“有权读取哪个订单”。Gateway 后端应根据可信身份校验订单归属。
Memory：保存有价值的状态，而不是保存所有内容
客服智能体通常同时需要两种记忆：
- 短期状态：当前问题、最近一次工具结果、尚未确认的操作。
- 长期信息：用户允许保存的偏好、产品信息或历史问题摘要。
不要默认把完整对话永久写入长期记忆。支付信息、身份凭证、访问令牌以及不必要的个人信息应该在写入前过滤，并设置保留期限。还要定义删除流程，否则“记住用户”很容易变成数据治理负担。
用适配器降低迁移风险
下面是一个可直接运行的本地示例。它不调用真实 AWS 服务，而是明确模拟 Runtime、Gateway 和 Memory 的边界。实际迁移时，可以保留 handle_request，将三个本地适配器替换成 AgentCore 对应 SDK 或服务调用；具体 SDK 方法和部署参数应以项目采用的版本为准。
将代码保存为 agent_adapter.py，使用 Python 3.10 或更高版本运行：
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Protocol
class Memory(Protocol):
def load(self, user_id: str, session_id: str) -> list[dict[str, str]]: ...
def append(self, user_id: str, session_id: str, role: str, text: str) -> None: ...
class Gateway(Protocol):
def call(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: ...
class LocalMemory:
def __init__(self) -> None:
self._sessions: dict[tuple[str, str], list[dict[str, str]]] = {}
def load(self, user_id: str, session_id: str) -> list[dict[str, str]]:
return list(self._sessions.get((user_id, session_id), []))
def append(self, user_id: str, session_id: str, role: str, text: str) -> None:
key = (user_id, session_id)
self._sessions.setdefault(key, []).append({"role": role, "text": text})
class LocalGateway:
def call(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
if tool_name != "get_order_status":
raise ValueError(f"Tool is not allowed: {tool_name}")
if arguments.get("order_id") != "A-100":
return {"found": False}
return {
"found": True,
"status": "shipped",
"estimated_delivery": "2026-04-12",
}
@dataclass(frozen=True)
class Request:
user_id: str
session_id: str
request_id: str
message: str
def langgraph_workflow(
request: Request,
history: list[dict[str, str]],
gateway: Gateway,
) -> str:
# 迁移第一阶段可在这里调用现有的 LangGraph graph.invoke(...)。
if "A-100" in request.message:
order = gateway.call("get_order_status", {"order_id": "A-100"})
if order.get("found"):
return (
f"订单 A-100 当前为 {order['status']}，"
f"预计送达日期为 {order['estimated_delivery']}。"
)
return f"我还需要订单号。当前会话已有 {len(history)} 条历史消息。"
def handle_request(payload: dict[str, str], memory: Memory, gateway: Gateway) -> dict[str, str]:
required = {"user_id", "session_id", "request_id", "message"}
missing = required - payload.keys()
if missing:
raise ValueError(f"Missing fields: {sorted(missing)}")
request = Request(**{key: payload[key] for key in required})
history = memory.load(request.user_id, request.session_id)
memory.append(request.user_id, request.session_id, "user", request.message)
answer = langgraph_workflow(request, history, gateway)
memory.append(request.user_id, request.session_id, "assistant", answer)
return {"request_id": request.request_id, "answer": answer}
if __name__ == "__main__":
memory = LocalMemory()
gateway = LocalGateway()
response = handle_request(
{
"user_id": "customer-42",
"session_id": "support-7",
"request_id": "req-001",
"message": "请查询订单 A-100",
},
memory,
gateway,
)
print(response)
运行命令：
python3 agent_adapter.py
真实接入时，建议保持接口不变并分别替换实现：
memory: Memory = AgentCoreMemoryAdapter(...)
gateway: Gateway = AgentCoreGatewayAdapter(...)
result = handle_request(runtime_event, memory, gateway)
这里的 AgentCoreMemoryAdapter 和 AgentCoreGatewayAdapter 是项目侧适配器名称，并非假定存在的官方类。它们的作用是隔离 SDK 版本、认证方式和序列化格式，让 LangGraph 工作流不必同时承担基础设施迁移。
第二阶段：用 Strands Agents 接管规划
第一阶段稳定后，系统仍然由 LangGraph 显式规定节点和路由。第二阶段可以把适合动态决策的部分迁移到 Strands Agents，让模型根据用户目标、工具描述和当前上下文选择下一步动作。
适合模型驱动规划的任务通常具有这些特征：
- 工具组合会随问题变化，难以列出固定路径。
- 大多数工具是只读或低风险操作。
- 调用失败后可以安全重试或请求用户补充信息。
- 已有离线评测集，可以衡量答案和工具选择是否退化。
并非所有 LangGraph 节点都应删除。退款、取消订单、修改地址等高风险动作仍适合确定性流程：先校验资格，再展示影响，获得明确确认后执行。可行的混合模式是让 Strands 负责规划和信息收集，由受约束的服务端工作流执行最终写操作。
迁移时可以按意图逐步切换，而不是一次替换整个图：
def route_agent(intent: str, request: dict) -> dict:
deterministic = {"refund", "cancel_order", "change_address"}
if intent in deterministic:
return existing_langgraph.invoke(request)
return strands_agent.invoke(request)
这段代码是迁移模式示例，invoke 的实际签名需要根据所用 LangGraph、Strands Agents 和 AgentCore SDK 版本调整。灰度阶段应记录路由结果、工具序列、最终答案和人工接管情况，以便对两套实现进行对照评估。
上线前需要验证什么
迁移的完成标准不应只是“AgentCore 上能够返回文本”。至少需要检查以下项目：
- 身份边界：客户端提供的
user_id是否会被可信身份覆盖或验证。 - 工具权限：每个工具是否按最小权限访问后端，是否校验资源归属。
- 幂等性：Runtime 重试后，有副作用的操作是否只执行一次。
- 记忆治理：写入前是否过滤敏感数据，是否有 TTL、删除和租户隔离策略。
- 超时预算：模型、Gateway 和后端 API 是否共享明确的端到端延迟预算。
- 可观测性：是否能够用
request_id串联运行、工具调用和错误，但不记录敏感正文。 - 质量评测：是否覆盖正常查询、工具失败、提示注入、越权请求和人工升级。
- 回退路径：Strands 规划异常时，是否能回到确定性流程或转交人工。
这条两阶段路线的核心价值在于分离风险：先让 AgentCore 承担 Runtime、Gateway 和 Memory 对应的生产职责，同时保留熟悉的 LangGraph 行为；再用 Strands Agents 改造规划层。团队因此可以分别衡量平台迁移和智能体行为变化，而不是在一次发布中同时调试两类问题。