AWS图代理

功能说明: 本技能涵盖 完整工作流程和配置指南、化工作流场景 等核心能力。

AWS图代理（AWS Graph Agent）

基于 AWS Bedrock AgentCore 与 LangGraph 编排的多代理系统。通过 StateGraph 状态图定义代理工作流，AgentCore Runtime 封装为 HTTP 服务，Memory 管理持久记忆，Gateway 集成外部工具.

主要能力

1. StateGraph 状态图编排

使用 LangGraph StateGraph 定义多代理工作流，支持 tools_condition 自动路由（代理→工具或 END）、ToolNode 预置工具执行器、条件边实现复杂多步逻辑（planner→executor→reviewer 循环）

处理: 解析StateGraph 状态图编排的输入参数,完成核心逻辑,生成结构化输出. 输出: 返回StateGraph 状态图编排的响应数据,含状态码、结果数据和运行日志.

2. AgentCore Runtime HTTP 封装

将代理封装为 8080 端口 HTTP 服务，处理 /invocations（调用）与 /ping（健康检查）端点，支持容器模式部署

处理: 解析AgentCore Runtime HTTP 封装的输入参数,完成核心逻辑,生成结构化输出. 输出: 返回AgentCore Runtime HTTP 封装的响应数据,含状态码、结果数据和运行日志.

• 调用时传入input_params参数,支持创建/查询/导出操作

3. AgentCore Memory 持久记忆

管理跨会话/跨代理的 STM（短期记忆，会话内逐轮）与 LTM（长期记忆，跨会话/跨代理），配套一致性处理模式（写入后约 10s 最终一致，含等待+验证+重试逻辑）

处理: 解析AgentCore Memory 持久记忆的输入参数,完成核心逻辑,生成结构化输出. 输出: 返回AgentCore Memory 持久记忆的响应数据,含状态码、结果数据和运行日志.

• 调用时传入input_params参数,支持创建/查询/导出操作

4. AgentCore Gateway 工具集成

将 API/Lambda 转化为带认证的 Agent 工具接口，支持 Fallback 模拟（本地开发）、Local 工具协议、Production Gateway（生产）三种传输模式

处理: 解析AgentCore Gateway 工具集成的输入参数,完成核心逻辑,生成结构化输出. 输出: 返回AgentCore Gateway 工具集成的响应数据,含状态码、结果数据和运行日志.

• 调用时传入input_params参数,支持创建/查询/导出操作

5. agentcore CLI 全生命周期管理

configure（交互式/脚本化配置）→launch（容器部署）→dev（热重载本地开发）→invoke（测试调用）→destroy（清理资源避免持续计费）

处理: 解析agentcore CLI 全生命周期管理的输入参数,完成核心逻辑,生成结构化输出. 输出: 返回agentcore CLI 全生命周期管理的响应数据,含状态码、结果数据和运行日志. 能力覆盖范围：支持的场景关键词如下：Bedrock、引导式部署、多代理模式库、AWS、多代理部署编排工、全生命周期管理五、大核心能力、适用于多代理协调、的复杂业务系统、跨会话持久记忆代、集成到代理工具链、生产级、代理部署、适用关键词、多代理编排等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持.

使用向导

1. 确认运行环境满足依赖说明中的要求
2. 在AI Agent对话中调用本技能,提供必要的输入参数
3. 检查输出结果,根据需要进行后续处理

详细的输入输出格式请参考下方章节说明。

应用场景

何时使用：

• 需要多代理协调的复杂业务系统（客服、电商、医疗、金融）
• 需要跨会话持久记忆的对话代理（用户偏好、历史决策）
• 需要将外部 API/Lambda 集成到代理工具链的企业应用
• 希望在生产环境大规模部署 AI 代理的团队
• 使用 LangGraph 进行复杂多步逻辑编排的开发者

输入：Python 代理代码（基于 langgraph + bedrock-agentcore）+ AWS 凭据 + Bedrock 模型使用审批 输出：部署成功的 HTTP 代理服务（8080 端口）+ CLI 管理能力 + 持久记忆与网关工具集成

不适用场景：

• 未完成 Bedrock 模型使用审批的账户
• 不需要多代理协调的简单单一代理场景（可直接用基础 LLM API）
• 非 AWS 平台部署需求

参数说明

pip install bedrock-agentcore bedrock-agentcore-starter-toolkit langgraph
uv tool install bedrock-agentcore-starter-toolkit  # 安装 agentcore CLI

Step 2：预检清单（部署前必读）

Step 3：编写代理代码（StateGraph + Runtime）

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from typing import Annotated
from typing_extensions import TypedDict
# ...
class State(TypedDict):
    messages: Annotated[list, add_messages]
# ...
builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge(START, "agent")
graph = builder.compile()
# ...
app = BedrockAgentCoreApp()
@app.entrypoint
def invoke(payload, context):
    result = graph.invoke({"messages": [("user", payload.get("prompt", ""))]})
    return {"result": result["messages"][-1].content}
app.run()
```bash
# 在此执行相关操作
echo "操作完成"
```bash
# 交互式配置
agentcore configure -e agent.py --region us-east-1
# 脚本化配置（CI/CD）
py --region us-east-1 --name my_agent --non-interactive
# 容器模式部署（生产）
agentcore launch --deployment-type container
# 无记忆部署（无状态代理）
agentcore launch --disable-memory
```bash
# 在此执行相关操作
echo "操作完成"
```bash
agentcore dev                              # 热重载本地开发
agentcore invoke '{"prompt": "Hello"}'    # 测试调用
```bash
# 在此执行相关操作
echo "操作完成"
```text
多代理协调? → 编排器+专家模式（编排器根据意图路由到专家，共享 session_id）
跨会话持久记忆? → AgentCore Memory（非 LangGraph checkpoints）
外部 API/Lambda? → AgentCore Gateway
单一代理简单? → 快速开始模板
复杂多步逻辑? → StateGraph + tools_condition + ToolNode
```bash
# 在此执行相关操作
echo "操作完成"
```bash
agentcore destroy

使用范例

示例(补充)

输入：部署一个带工具调用的简单代理

输出（代码与部署）：

# 最简模式：用户输入 → 代理节点 → tools_condition → ToolNode → 回到代理
#                                    → END（无需工具）
builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_conditional_edges("agent", tools_condition)  # 自动路由
builder.add_edge(START, "agent")
graph = builder.compile()

部署：agentcore configure -e agent.py --region us-east-1 && agentcore launch

示例二：记忆系统写入与验证（一致性处理·指数退避）

输入：写入长期记忆并确保一致性

输出（写入+指数退避验证逻辑）：

memory import MemoryClient
import time
# ...
memory = MemoryClient()
memory.create_event(session_id, actor_id, event_type, payload)  # 写入
# ...
# 最终一致性验证（指数退避：2s→4s→8s→16s→30s，最多 5 次）
def verify_with_backoff(memory, session_id, actor_id, event_type, payload,
                        base=2, max_wait=30, max_retries=5):
    for attempt in range(max_retries):
        time.sleep(min(base * (2 ** attempt), max_wait))  # 指数退避+上限
        if memory.list_events(session_id):
            return  # 一致性达成
        if attempt < max_retries - 1:
            memory.create_event(session_id, actor_id, event_type, payload)  # 重写
    raise RuntimeError(f"记忆一致性验证失败：{max_retries} 次重试后仍为空")
# ...
verify_with_backoff(memory, session_id, actor_id, event_type, payload)
# 注意：event['payload'] 是列表类型；确认 actor_id 和 session_id 匹配

示例三：多代理协调（编排器+专家模式）端到端

输入：客服系统按意图路由到客服专家/计费专家，共享 session_id 跨专家记忆

输出（编排器+专家+共享记忆，完整部署）：

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
# ...
class State(TypedDict):
    messages: list
    expert: str
# ...
def orchestrator(state):
    intent = classify(state["messages"][-1])  # 意图分类
    return {"expert": {"投诉": "cs_expert", "账单": "billing_expert"}[intent]}
# ...
def cs_expert(state):
    return {"messages": [handle_cs(state)]}      # 客服专家处理
# ...
def billing_expert(state):
    return {"messages": [handle_billing(state)]}  # 计费专家处理
# ...
builder = StateGraph(State)
builder.add_node("orchestrator", orchestrator)
builder.add_node("cs_expert", cs_expert)
builder.add_node("billing_expert", billing_expert)
builder.add_edge(START, "orchestrator")
builder.add_conditional_edges("orchestrator", lambda s: s["expert"])
builder.add_edge("cs_expert", END)
builder.add_edge("billing_expert", END)
graph = builder.compile()

py --region us-east-1 && agentcore launch。两个专家共享同一 session_id`，通过 AgentCore Memory 实现跨专家记忆传递（编排器写入意图，专家读取上下文），避免用户重复陈述.

示例四：Gateway 工具集成端到端

输入：将订单查询/退款 Lambda 集成为代理工具，生产网关模式部署

输出（Gateway 注册+代理调用+三种传输模式）：

gateway import GatewayClient
# ...
# 三种传输模式按环境选择
mode = "production"  # "production"（本地开发）/ "local"（本地协议）/ "production"（生产网关）
gateway = GatewayClient(mode=mode)
tools = gateway.register_tools([
    {"name": "search_orders", "lambda_arn": "arn:aws:lambda:us-east-1:…:OrderSearch"},
    {"name": "issue_refund",  "lambda_arn": "arn:aws:lambda:us-east-1:…:Refund"},
])
# 工具名必须去除 Lambda 的 ___ 前缀，否则返回 "Unknown tool"
# ...
agent = create_agent_with_tools(tools)  # 工具自动注入 StateGraph

调用：agentcore invoke '{"prompt": "查询订单 #1234 并退款"}'。代理自动调用 search_orders→issue_refund，Gateway 处理 Lambda 认证与调用。本地开发用 mode="mock" 返回假数据无需真实 Lambda；生产用 mode="production" 走网关鉴权.

前置条件

成本优化策略：无状态部署用 --disable-memory 节省记忆存储成本；测试后立即 agentcore destroy 避免持续运行成本；生产用容器、开发用 agentcore dev；选择低成本区域（如 us-east-1）；使用 us..-* 推理配置降低吞吐量成本. 密钥安全处理规范（零暴露硬性约束）：代理绝不能读取、cat、打印、回显或日志记录 AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY 明文，绝不在聊天/提交/镜像中暴露凭据——无论出于任何原因包括"验证配置"。硬性规则：(1) 凭据只通过环境变量或 IAM Role 注入，禁止硬编码到源码或 Dockerfile ARG；(2) agentcore configure 所需凭据从 AWS CLI 命名配置文件（--profile）或实例元数据获取，禁止在命令行明文传参；(3) 容器部署用任务角色（Task Role）而非 ENV 明文密钥，ENV 仅用于非敏感配置；(4) 如用户在聊天中粘贴了密钥，立即告知已泄露，要求轮换并从历史清除，绝不回显；(5) .env 加入 .gitignore，禁止提交版本库；(6) agentcore destroy 后确认密钥与资源已清理。验证安装状态用 aws sts get-caller-identity（返回角色 ARN，不暴露密钥）而非打印环境变量. 可用性分类：MD+EXEC（纯 Markdown 指令，需要命令行执行能力进行部署与管理）

• API Key：本skill无需额外API Key配置

常见疑问

Q1：部署后记忆为空怎么办？ A：记忆写入后有约 10 秒最终一致性延迟。等待 10 秒后用 list_events 重新查询。如仍为空，检查日志中是否显示"Memory enabled"，确认部署时未使用 --disable-memory. Q2：容器无法读取 .env 文件怎么办？ A：容器模式下 .env 文件不会被自动读取。在 Dockerfile 中使用 ENV 指令设置环境变量，而非依赖 .env 文件. Q3：收到"on-demand throughput isn't supported"错误？ A：使用 us..-* 推理配置文件替代按需吞吐量。这是区域和模型组合的限制. Q4：代理名称无效？ A：代理名称必须字母开头，仅含字母/数字/下划线，1-48 字符。将连字符改为下划线（如 my-agent → my_agent）. Q5：Gateway 返回"Unknown tool"？ A：Lambda 函数必须从 bedrockAgentCoreToolName 参数中去除 ___ 前缀。检查 Lambda 代码中的工具名处理逻辑.

使用约束

1. 记忆最终一致性延迟约10秒：写入后不能立即可读，需等待+验证+重试机制，不适合强一致性场景
2. 依赖 Bedrock 模型审批：未在 Bedrock Console 填写 表单则无法部署，审批流程不可跳过
3. 代理命名规则严格：仅字母/数字/下划线，1-48 字符，连字符等常见命名方式不被接受
4. 容器模式不支持 .env：必须在 Dockerfile 中用 ENV 设置环境变量，与本地开发习惯不同
5. Gateway 工具名需去前缀：Lambda 的 bedrockAgentCoreToolName 必须去除 ___ 前缀，否则返回"Unknown tool"

安全提示

使用前请确认已阅读依赖说明章节，确保运行环境满足安全要求。

效率指标

优势对比

功能矩阵

• 自动化执行: Bedrock Age
• 文件处理: 支持多种文件格式的读取、解析和写入操作
• API集成: 通过标准化接口调用外部服务并处理响应
• 命令执行: 在安全沙箱中执行系统命令并收集结果
• 信息检索: 快速搜索和过滤目标数据

前置条件

• 已安装所需运行环境(参考依赖说明)
• 已获取必要的API密钥或访问凭证(如适用)
• 输入数据已准备就绪

故障恢复流程

针对AWS图代理使用中可能遇到的常见问题,提供以下排查方案:

AWS图代理通用排查步骤

1. 检查输入参数: 确认所有必填参数已提供且格式正确
2. 查看日志输出: 定位具体错误行和异常类型
3. 验证环境配置: 确认依赖库版本和运行环境满足要求
4. 逐步调试: 缩小问题范围,隔离故障模块

使用限制说明

• 极端边界输入可能影响输出质量,建议对异常输入做预校验
• API凭证需妥善管理,避免硬编码到代码中,推荐使用环境变量注入
• 生成结果受模型能力影响,不同模型输出质量可能有差异
• 文件格式兼容性受底层库限制,部分特殊格式可能不被支持
• 不同操作系统的命令行参数可能存在差异,需做平台适配