逐步将 LangGraph Agent 部署到 AWS Bedrock AgentCore

许多 LangGraph 智能体在笔记本电脑上运行良好，但在投入实际使用阶段却会遇到问题。AWS Bedrock AgentCore 通过无需重新设计的可扩展无服务器托管方案解决了这一缺陷。以下内容将介绍如何将现有智能体适配到 AgentCore、创建项目、在本地运行、部署到 AWS，以及在其上线后如何调用它。

AgentCore 是什么

AgentCore 是 AWS 提供的一种托管运行时环境，智能体可在其中被部署、扩展和运行，而无需任何人管理服务器。LangGraph、Strands、CrewAI 和 LlamaIndex 等框架均可使用，OpenAI、Google 和 Anthropic 的模型也同样适用，因此智能体代码无需做任何修改。会话隔离、持久内存、可观测性以及身份管理功能均已内置。

所有操作均由 AgentCore CLI 驱动。它是一个 Node.js 程序，负责生成项目框架、在本地运行代理，并通过底层的 CDK 将其部署到 AWS。

前置条件

• Python 3.12 或更高版本
• 用于管理 Python 包的 uv 或 pip
• Node.js 20 或更高版本，因为该 CLI 是通过 npm 分发的
• 通过 npm install -g aws-cdk 全局安装的 AWS CDK
• 已通过 aws configure 在本地配置好凭据的 AWS 账户
• 能够生成编译后 StateGraph 的现有 LangGraph 代码

步骤 1：安装 CLI

只有一个命令行工具可以完成所有操作：创建项目、在本地运行应用以及部署到 AWS。

npm install -g @aws/agentcore

确认安装已成功：

agentcore --help

第2步：添加依赖项

有两个软件包需要加入代理的依赖列表中。bedrock-agentcore是提供BedrockAgentCoreApp封装类的Python SDK，而aws-opentelemetry-distro则是AWS开发的OpenTelemetry Python监控工具版本。

# pyproject.toml
[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "aws-opentelemetry-distro==0.17.0",
 "bedrock-agentcore>=1.6.3",
    "boto3>=1.42.0",
    "langgraph>=1.1.0",
    "langchain-core>=1.2.0",
    # ... your other existing dependencies
]

使用项目中常用的工作流程进行安装：

uv sync
# or: pip install -e .

步骤3：在main.py中编写入口点

AgentCore要求有一个Python文件作为入口点，该文件需包含一个BedrockAgentCoreApp实例以及一个用@app.entrypoint装饰的函数。LangGraph的逻辑几乎无需更改，只需将其封装在一个简单的运行时入口点中即可。

# main.py
from langchain_core.messages import HumanMessage
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from graphs.my_agent_graph import build_my_agent  # your existing graph builder

# 1. Instantiate the AgentCore app and logger
app = BedrockAgentCoreApp()
log = app.logger

# 2. Build your graph at module load time (startup)
#    AgentCore initialises the module once, then handles concurrent invocations.
#    Any failure here will prevent a broken agent from going live.
def create_agent():
    log.info("Initialising agent...")
    graph = build_my_agent()   # returns your compiled LangGraph StateGraph
    log.info("Agent ready")
    return graph
try:
    graph = create_agent()
except Exception as e:
    log.error(f"Critical failure during agent initialisation: {e}")
    raise  # fail fast - don't let a broken agent start serving requests

# 3. Async helper that drives the LangGraph streaming loop
async def run_agent(user_input: str, session_id: str = "default-session") -> str:
    responses = []
    config = {"configurable": {"thread_id": session_id}}
    async for chunk in graph.astream(
        {"messages": [HumanMessage(content=user_input)]},
        config=config,
        stream_mode="values",
    ):
        messages = chunk.get("messages", [])
        if messages:
            last = messages[-1]
            if getattr(last, "type", None) == "ai":
                responses.append(last)
    if not responses:
        return "No response"
    content = getattr(responses[-1], "content", "")
    return content if isinstance(content, str) else str(content)

# 4. The entrypoint - this is what AgentCore calls on every invocation
@app.entrypoint
async def invoke(payload, context):
    try:
        log.info("Invoke received")
        user_input = payload.get("prompt", "")
        session_id = payload.get("session_id", "default-session")
        if not user_input.strip():
            return {"error": "Prompt cannot be empty"}
        response = await run_agent(user_input, session_id)
        return {"response": response}
    except Exception as e:
        log.error(f"Error: {e}")
        return {"error": str(e)}

# 5. Run the app (only executed when running locally using agentcore dev)
if __name__ == "__main__":
    app.run()

该文件中重要的内容

• BedrockAgentCoreApp() 为运行时提供了基础框架：包括 HTTP 服务器、健康检查路由以及结构化日志，这些功能都已预先配置好。
• 图结构是在模块级别构建的，仅在导入时生成一次，而非每次调用时都重新生成。因此任何启动问题都会立即显现，出错的程序将永远不会收到请求。
• @app.entrypoint 用于标记处理传入请求的函数。该函数会接收到解码后的 JSON 数据 payload，以及与 AgentCore 请求相关的 context。它必须返回一个可序列化为 JSON 的普通字典，纯字符串或自定义类均不可用。
• 传入的 session_id 会变成 LangGraph 的 thread_id，这样 MemorySaver 就能为每个会话单独管理内存。

第 4 步：创建项目

接下来是项目结构设置。在干净的根目录下运行初始化向导，让其构建项目框架：

agentcore create

向导会要求输入项目名称、语言、Python版本以及入口文件等详细信息。入口文件和Python版本必须与应用程序相匹配。之后，agentcore文件夹中会生成一个agentcore.json文件，用于告知AgentCore代码的位置及运行方式。

{
  "name": "MyAgentOnAgentcore",
    "runtimes": [
    {
      "name": "my-agent",
      "build": "CodeZip",
      "entrypoint": "main.py",
      "codeLocation": "app/my-agent/",
      "runtimeVersion": "PYTHON_3_12",
      "networkMode": "PUBLIC",
      "protocol": "HTTP",
	  "envVars": []
	}
  ]
}

第5步：将代码放置到AgentCore会查找的位置

将代理程序的源代码移动或复制到agentcore.json中codeLocation指定的目录里。很容易出错，而位置错误会使得部署过程变得极为麻烦。常见的错误是将代码直接放入agentcore文件夹中，这会耗费大量时间。正确的目录结构应如下所示：

MyAgentOnAgentcore/
├──agentcore
	└── agentcore.json
├──app/
	└──my-agent/          			 ← codeLocation
		├── main.py                  ← entrypoint (matches agentcore.json)
		├── pyproject.toml           ← or requirements.txt
		├── src/
		│   └── my_agent_graph.py    ← your LangGraph graph builder
		│   └── tools.py             ← your tools, utilities, etc.
		└── .env					 ← your configuration variables

main.py 必须位于 codeLocation 的最顶层，并且要与 agentcore.json 中指定的入口点完全一致。

环境变量

AgentCore 可以将环境变量传递给运行时容器。在本地环境中，使用 .env 文件是最简便的方式。在生产环境中，这些值可以存储在 agentcore.json 的 envVars 字段中，当然 .env 文件也同样适用。诸如密码和 API 密钥之类的敏感信息最好存储在 AWS Secrets Manager 中并在运行时加载，此时 AgentCore 的执行角色需要具备读取这些信息的权限。需要注意的是，envVars 是一个 JSON 对象数组，每个对象都包含 name 和 value 两个字段。

第 6 步：验证

在运行任何程序之前，先让 CLI 检查配置是否正确：

agentcore validate

诸如缺少入口点文件或配置中的路径错误这类简单问题就会在这里被检测出来。

第7步：在本地运行并测试

使用以下命令启动本地运行环境：

agentcore dev

它会启动一个与生产环境高度相似的本地HTTP服务器。当所有配置正确后，输出会显示应用已准备就绪并正在监听本地端口。

调用本地服务器

可以通过命令行界面进行调用：

agentcore invoke "Summarise last month sales"

或通过 curl 实现：

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Summarise last month sales", "session_id": "test-001"}'

理想的响应格式是包含智能体答案的 JSON：

{"response": "Last month, total sales were $1.2M across 3 regions..."}

第8步：部署到AWS

当本地测试结果良好后，将其部署到AgentCore运行时环境：

agentcore deploy

在后台，CLI会打包代码，创建一个用于直接部署代码的S3存储桶，生成一个IAM执行角色，并使用CDK部署AgentCore运行时环境。首次部署需要几分钟时间，之后的部署通常会更快。

部署完成后，需检查运行时状态。输出结果会显示运行时的ARN以及用于调用已部署代理的HTTP地址。

agentcore status

第9步：调用已部署的代理

使用CLI是测试已部署版本的最简单方法：

agentcore invoke --prompt "Who are the top 5 customers by revenue?" --session-id "session-id-with-length-greater-than-or-equal-33"

通过 HTTP 调用需要使用 AWS SigV4 对请求进行签名：

curl -X POST "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn-of-MyAgentOnAgentcore-xxxx/invocations" \
  --user "$AWS_ACCESS_KEY_ID:$AWS_SECRET_ACCESS_KEY" \
  --aws-sigv4 "aws:amz:us-east-1:bedrock-agentcore" \
  -H "Content-Type: application/json" \
  -H "x-amzn-bedrock-agentcore-runtime-session-id: session-id-with-length-greater-than-or-equal-33" \
  -d '{"prompt": "Who are the top 5 customers by revenue?"}'

使用 agentcore status 中提供的 HTTP URL。

对于来自其他 Python 代码的调用，SDK 通常是最佳选择。将运行时 ARN 存储在环境变量中，传递提示信息，并在每次后续调用时再次发送之前的会话 ID。

# invoke_agent.py
import json
import boto3
import os
from dotenv import load_dotenv

load_dotenv()

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
runtime_arn = os.getenv("AGENTCORE_RUNTIME_ARN")

def invoke(prompt: str, session_id: str) -> str:
    payload = json.dumps({"prompt": prompt, "session_id": session_id})
    response = client.invoke_agent_runtime(
        agentRuntimeArn=runtime_arn,
        payload=payload,
        contentType="application/json"
    )
    body = json.loads(response["response"].read().decode())
    return body["response"]

if __name__ == "__main__":
    session_id = ""
    print("Ready. Type 'exit' to quit.")

    while True:
        prompt = input("Your question: ").strip()
        if prompt == "exit":
            break

        if not prompt:
            continue

        params = {
            "agentRuntimeArn": runtime_arn,
            "payload": json.dumps({"prompt": prompt}),
            "contentType": "application/json"
        }
        if session_id:
            params["runtimeSessionId"] = session_id

        try:
            response = client.invoke_agent_runtime(**params)
            body = json.loads(response["response"].read().decode())
            session_id = response.get("runtimeSessionId", session_id)
            print(f"Agent: {body['response']}\n")
        except Exception as e:
            print(f"Error: {e}")

在 .env 文件中设置 AGENTCORE_RUNTIME_ARN：

AGENTCORE_RUNTIME_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/MyAgentOnAgentcore-xxxx

在多次调用之间保持会话状态

请查看boto3示例中的runtimeSessionId。AgentCore会在每次响应时返回一个该值。在后续请求中将其传回，即可将相关操作路由到同一个会话上下文，再结合LangGraph的MemorySaver功能，智能体便能在多次独立的HTTP调用之间保留完整的对话记忆。

常见陷阱

• 启动失败是刻意设计的。当create_agent()在导入时抛出异常时，AgentCore不会启动运行时环境。这种设计可以防止配置错误的智能体（如缺少凭证或数据库URL有误）在无人察觉的情况下开始运行。
• 需要添加权限。AgentCore 创建的执行角色必须明确获得该代理所使用的所有 AWS 服务的访问权限，包括 Secrets Manager 和 S3。应在首次部署后添加这些权限。
• 版本必须一致。pyproject.toml 中声明的 Python 版本必须与 agentcore.json 中的运行时设置相匹配。
• 需返回普通字典。如果入口点返回字符串或自定义对象，运行时环境将拒绝接受。

实际会发生哪些变化

要将本地 LangGraph 智能体转换为 AgentCore，只需进行少量修改：安装两个额外包，将图结构封装在 BedrockAgentCoreApp 中，搭建项目框架，在本地进行测试，之后再部署并调用它。图结构、相关工具、模型调用方式以及内存配置基本保持不变。AgentCore 负责基础设施层面的运行，而 LangGraph 仍负责智能体的逻辑实现。

相关阅读

• 你的智能体图结构应使用 YAML：在启动时编译包含 17 个智能体的 LangGraph —— 讲解小型运行时如何将一个 YAML 规范编译为 LangGraph StateGraph：包括状态生成、importlib 插件、标签路由器、自适应调用以及会话级节点替换等功能。

LangGraph支持机器人的验证机制与多智能体路由设计 — 为电子商务领域的LangGraph智能体添加确定性的业务逻辑验证机制、自校正的验证循环，以及由三个专业子图构成的监督模块。