名称: aws-agentcore-langgraph
描述: 在 AWS Bedrock AgentCore 上部署生产级 LangGraph 智能体。适用于:(1) 采用协调器与专家智能体模式的多智能体系统,(2) 构建具有持久化跨会话记忆的状态化智能体,(3) 通过 AgentCore 网关(MCP、Lambda、API)连接外部工具,(4) 管理分布式智能体间的共享上下文,或 (5) 通过 CLI 部署具备生产级可观测性与扩展性的复杂智能体生态。
基于 AWS Bedrock AgentCore 与 LangGraph 编排的多智能体系统。源码:https://github.com/aws/bedrock-agentcore-starter-toolkit
pip install bedrock-agentcore bedrock-agentcore-starter-toolkit langgraph
uv tool install bedrock-agentcore-starter-toolkit # 安装 agentcore CLI
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) # 路由至工具或 END
builder.add_edge(START, "agent")
graph = builder.compile()
app = BedrockAgentCoreApp() # 封装为运行在 8080 端口的 HTTP 服务(处理 /invocations, /ping)
@app.entrypoint
def invoke(payload, context):
result = graph.invoke({"messages": [("user", payload.get("prompt", ""))]})
return {"result": result["messages"][-1].content}
app.run()
| 命令 | 用途 |
|---|---|
agentcore configure -e agent.py --region us-east-1 |
初始配置 |
agentcore configure -e agent.py --region us-east-1 --name my_agent --non-interactive |
脚本化配置 |
agentcore launch --deployment-type container |
部署(容器模式) |
agentcore launch --disable-memory |
部署时不启用记忆子系统 |
agentcore dev |
启动支持热重载的本地开发服务器 |
agentcore invoke '{"prompt": "Hello"}' |
测试调用 |
agentcore destroy |
清理部署资源 |
session_id 共享上下文from bedrock_agentcore.memory import MemoryClient
memory = MemoryClient()
memory.create_event(session_id, actor_id, event_type, payload) # 存储
events = memory.list_events(session_id) # 检索(返回列表)
python -m bedrock_agentcore.gateway.deploy --stack-name my-agents --region us-east-1
from bedrock_agentcore.gateway import GatewayToolClient
gateway = GatewayToolClient()
result = gateway.call("tool_name", param1=value1, param2=value2)
BEDROCK_AGENTCORE_GATEWAY_URL 环境变量需要多个智能体协同? → 采用协调器+专家智能体模式
需要持久化跨会话记忆? → 使用 AgentCore 记忆系统(而非 LangGraph 检查点)
需要连接外部 API/Lambda? → 使用 AgentCore 网关
单个智能体,逻辑简单? → 使用上方“快速开始”示例
复杂多步骤逻辑? → 使用 StateGraph + tools_condition + ToolNode
/invocations、/ping 端点)tools_condition 用于智能体→工具路由,ToolNode 用于执行my_agent(而非 my-agent)| 问题 | 解决方案 |
|---|---|
on-demand throughput isn't supported |
使用 us.anthropic.claude-* 推理配置 |
Model use case details not submitted |
在 Bedrock 控制台填写 Anthropic 表单 |
Invalid agent name |
使用下划线而非连字符 |
| 写入后记忆为空 | 等待约 10 秒(最终一致性) |
| 容器未读取 .env 文件 | 在 Dockerfile 中设置 ENV,而非 .env 文件 |
| 部署后记忆不工作 | 检查日志中的 "Memory enabled/disabled" 信息 |
list_events 返回空列表 |
检查 actor_id/session_id 是否匹配;event['payload'] 为列表 |
| 网关报错 "Unknown tool" | Lambda 必须从 bedrockAgentCoreToolName 中去除 ___ 前缀 |
| 平台不匹配警告 | 正常现象 - CodeBuild 会处理 ARM64 跨平台构建 |