用 LangGraph + Bedrock AgentCore 在 AWS 上搭一套可扩展的无服务器多智能体系统

预计阅读时间：14 分钟

多智能体系统正从"实验性 demo"走向"生产级服务"，但真正让人头疼的不是编排逻辑本身，而是两件事：状态怎么持久化，以及调用链怎么追踪。AWS 最近推出的 Amazon Bedrock AgentCore 正好瞄准了这两个痛点——它把 Memory 和 Observability 做成了托管服务，而 LangGraph 则提供了灵活的有状态图编排能力。两者组合，可以让你在 AWS 上用纯无服务器架构跑一套可水平扩展的多智能体工作流。

下面拆解这套方案的关键设计，并给出可以直接改造的代码和部署示例。

整体架构：谁负责什么

先看各组件的分工边界，避免混在一起写成一团：

层	组件	职责
编排层	LangGraph Agents	定义有状态的多节点图，控制 agent 间的路由、分支、循环
记忆层	Bedrock AgentCore Memory	跨 session 持久化短期/长期记忆，支持语义检索
可观测层	Bedrock AgentCore Observability	自动采集 trace、span、token 用量，对接 CloudWatch
计算层	AWS Lambda + API Gateway	无服务器运行 LangGraph 图，按调用计费
模型层	Amazon Bedrock	提供 Claude、Llama 等模型推理端点

核心思路：LangGraph 只管编排，不管存储和追踪。Memory 和 Observability 都委托给 AgentCore，这样图的定义保持干净，运维成本也降到了托管级别。

LangGraph 多智能体编排：从单图到路由

LangGraph 的关键抽象是 StateGraph——每个节点读入共享 state、输出更新后的 state，边可以是静态的也可以是条件路由。一个典型的多 agent 编排至少包含一个 router 节点和多个专业 agent 节点。

下面是一个可运行的编排示例，模拟"用户提问 → router 分流 → 专业 agent 处理 → 合并回答"的流程：

# langgraph_multi_agent.py
from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel

# ---- 1. 定义共享状态 ----
class AgentState(BaseModel):
    messages: Annotated[list, add_messages]
    next_agent: str = ""

# ---- 2. 定义模型（使用 Bedrock） ----
from langchain_aws import ChatBedrock

model = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229",
    region_name="us-east-1",
)

# ---- 3. 创建专业子 agent ----
research_agent = create_react_agent(
    model,
    tools=[],  # 实际项目中接入检索工具
    name="researcher",
)

writer_agent = create_react_agent(
    model,
    tools=[],  # 可接入文本生成/格式化工具
    name="writer",
)

# ---- 4. Router 节点：决定下一步 ----
def router(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
    """根据最后一条消息的内容决定路由到哪个 agent。"""
    last_msg = state.messages[-1].content.lower() if state.messages else ""

    if "调研" in last_msg or "分析" in last_msg or "research" in last_msg:
        return "researcher"
    if "撰写" in last_msg or "总结" in last_msg or "write" in last_msg:
        return "writer"
    return "__end__"

def route_entry(state: AgentState) -> dict:
    """入口节点，不做处理，只触发路由。"""
    return {}

# ---- 5. 组装 StateGraph ----
graph = StateGraph(AgentState)

graph.add_node("router", route_entry)
graph.add_node("researcher", research_agent)
graph.add_node("writer", writer_agent)

graph.add_edge(START, "router")
graph.add_conditional_edges("router", router)
graph.add_edge("researcher", "router")   # 处理完回到 router，可继续流转
graph.add_edge("writer", "router")

app = graph.compile()

运行一次调用：

result = app.invoke({
    "messages": [{"role": "user", "content": "请调研 AWS Lambda 冷启动的最新优化方案，然后撰写一份技术总结"}],
})
for msg in result["messages"]:
    print(f"[{msg.name or msg.role}] {msg.content[:200]}")

这个图的特点是 router → agent → router 的循环结构，LangGraph 天然支持这种有环图，而简单的链式 pipeline 框架做不到。实际项目中，router 的判断逻辑可以用 LLM 做分类，也可以用规则引擎，取决于延迟要求和分类复杂度。

接入 Bedrock AgentCore Memory：让 agent 记住上下文

多智能体系统如果每次调用都从零开始，用户体验会很差——用户说"帮我继续上次的分析"，agent 应该能检索到历史会话的关键结论。Bedrock AgentCore Memory 提供了短期会话记忆和长期语义记忆两种存储，通过 API 直接读写。

下面展示如何在 LangGraph 节点中集成 Memory：

# memory_integration.py
import boto3
import json
from datetime import datetime

memory_client = boto3.client(
    "bedrock-agentcore",        # AgentCore 的 SDK 服务名
    region_name="us-east-1",
)

MEMORY_SESSION_ID = "user-42-session-7"

# ---- 写入短期记忆（会话内上下文） ----
def save_short_term_memory(session_id: str, role: str, content: str):
    memory_client.put_memory_entry(
        memoryId=MEMORY_SESSION_ID,
        sessionId=session_id,
        entries=[
            {
                "role": role,
                "content": content,
                "timestamp": datetime.utcnow().isoformat(),
            }
        ],
    )

# ---- 查询长期记忆（跨会话语义检索） ----
def query_long_term_memory(query: str, top_k: int = 5):
    response = memory_client.query_memory(
        memoryId=MEMORY_SESSION_ID,
        query=query,
        topK=top_k,
    )
    return response.get("results", [])

# ---- 在 LangGraph 节点中使用 ----
def researcher_with_memory(state: AgentState):
    # 1. 检索长期记忆，看是否有相关历史结论
    history = query_long_term_memory(
        query=state.messages[-1].content,
        top_k=3,
    )
    context_snippets = [h["content"] for h in history]

    # 2. 把历史上下文注入 prompt
    enriched_messages = state.messages + [
        {"role": "system", "content": f"相关历史结论：{json.dumps(context_snippets)}"}
    ]

    # 3. 调用 agent
    result = research_agent.invoke({"messages": enriched_messages})

    # 4. 存入短期记忆，供本次会话后续节点使用
    save_short_term_memory(
        session_id=MEMORY_SESSION_ID,
        role="researcher",
        content=result["messages"][-1].content,
    )

    return {"messages": result["messages"]}

关键设计决策：长期记忆用语义检索，短期记忆用顺序写入。这样 researcher 拿到的是"和当前问题最相关的历史片段"而非整段聊天记录，既减少 token 消耗，又提高召回精度。

注意：bedrock-agentcore 的 SDK API 名称可能随服务版本迭代调整，部署前请对照最新 AWS 文档确认具体参数。上面的调用形式基于当前公开的 AgentCore Memory 接口设计。

Bedrock AgentCore Observability：不用自己搭 trace 管线

多智能体调用链的追踪是生产环境必须解决的问题——一个用户请求可能经过 router → researcher → router → writer 四五个节点，每个节点内部还有工具调用，如果某个环节超时或返回异常，没有 trace 就像在黑盒里找故障。

AgentCore Observability 的做法是：在 LangGraph 节点执行前后自动埋点，生成 OpenTelemetry 兼容的 span 数据，推送到 CloudWatch。你只需要在启动时注册观察器：

# observability_setup.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# AgentCore Observability 会提供一个 OTLP endpoint
# 以下为典型接入模式
provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(
        endpoint="https://observability.bedrock-agentcore.us-east-1.amazonaws.com/v1/traces",
        # 认证由 AWS IAM 自动处理（通过 sigv4 插件）
    )
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# 之后 LangGraph 的每次节点执行都会自动上报 span
# 在 CloudWatch → Traces 面板中可查看完整调用链

实际收益：你能在 CloudWatch 看到每个 agent 节点的执行时长、token 消耗、以及跨节点的因果关系。如果 researcher 节点耗时从 2s 突然涨到 15s，trace 里一眼就能定位是模型推理慢还是工具调用卡了。

无服务器部署：把 LangGraph 图跑在 Lambda 上

编排逻辑写好了，接下来是部署。把 LangGraph 编译后的图打包成 Lambda 函数，通过 API Gateway 暴露 HTTP 接口，就是一套完整的无服务器多智能体服务。

# template.yaml — AWS SAM 部署模板
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Resources:
  MultiAgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./src
      Handler: app.handler
      Runtime: python3.12
      Timeout: 120           # 多 agent 调用链可能较长，给足超时
      MemorySize: 512        # Bedrock SDK + LangGraph 需要一定内存
      Environment:
        Variables:
          AWS_REGION: us-east-1
          BEDROCK_MODEL_ID: anthropic.claude-3-sonnet-20240229
          MEMORY_SESSION_PREFIX: "agent-mem-"
      Events:
        InvokeAgent:
          Type: Api
          Properties:
            Path: /invoke
            Method: post

  # IAM 权限：Lambda 需要访问 Bedrock 和 AgentCore
  MultiAgentFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: BedrockAccess
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - bedrock:InvokeModel
                Resource: "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229"
        - PolicyName: AgentCoreAccess
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - bedrock-agentcore:PutMemoryEntry
                  - bedrock-agentcore:QueryMemory
                  - bedrock-agentcore:PutTrace
                Resource: "*"

Lambda 的入口函数：

# src/app.py
import json
from langgraph_multi_agent import app as langgraph_app

def handler(event, context):
    body = json.loads(event.get("body", "{}"))
    user_message = body.get("message", "")

    result = langgraph_app.invoke({
        "messages": [{"role": "user", "content": user_message}],
    })

    # 只返回最终回答，不暴露内部 state
    final_answer = result["messages"][-1].content

    return {
        "statusCode": 200,
        "body": json.dumps({"answer": final_answer}),
    }

部署命令：

# 构建并部署
sam build
sam deploy --guided

# 部署完成后测试调用
curl -X POST https://<api-id>.execute-api.us-east-1.amazonaws.com/Prod/invoke \
  -H "Content-Type: application/json" \
  -d '{"message": "调研 Bedrock AgentCore Memory 的核心能力，然后写一份使用指南"}'

几个容易踩的坑和应对思路

Lambda 冷启动 + Bedrock SDK 初始化：LangGraph 和 Bedrock SDK 的 import 链比较长，首次冷启动可能 3-5s。应对方式：用 Provisioned Concurrency 保持少量预热实例，或者在图编译阶段就完成所有初始化，避免在 handler 里重复编译。

Memory 的 session ID 设计：不要用全局固定 ID。推荐格式 {user_id}-{conversation_id}，这样不同用户的记忆互不干扰，同一用户的不同话题也能隔离。长期记忆的语义检索会跨 session，但短期记忆只限当前 session。

Router 的分类准确率：纯关键词匹配在复杂场景下不够用。可以换成 LLM-based router——让模型输出结构化的 {"next": "researcher"} 判断，代价是多一次模型调用和约 200ms 延迟。在路由节点数超过 5 个时，这个代价通常值得。

Observability 的 span 粒度：默认埋点是节点级别。如果你需要追踪工具调用内部的细节（比如某个 API 请求的耗时），需要在工具函数里手动加 with tracer.start_as_current_span("tool_name") 的细粒度埋点。

上线前的检查清单

[ ] LangGraph 图是否编译通过、无环死循环风险（设置 recursion_limit）
[ ] Bedrock 模型 IAM 权限是否限定到具体 foundation model ARN
[ ] AgentCore Memory 的 session ID 是否按用户隔离
[ ] Lambda 超时是否覆盖最长可能的 agent 链路（建议 ≥ 90s）
[ ] Observability trace 是否能在 CloudWatch 正常查看完整 span 链
[ ] API Gateway 是否配置了请求体大小限制（Bedrock 输入有 token 上限）
[ ] 是否设置了 recursion_limit 防止 router 无限循环

这套组合的核心价值是：编排用 LangGraph 保持灵活性，存储和追踪用 AgentCore 保持运维简洁，计算用 Lambda 保持成本可控。三者各管一块，边界清晰，比把所有逻辑塞进一个巨型 Lambda 要容易维护得多。如果你的多智能体场景需要频繁调整路由策略或增减 agent 节点，这个架构的改动成本很低——改图定义、重新部署，不需要重新搭建存储和监控管线。