aws-strands-agents-agentcore

AWS Strands Agents & AgentCore

概述

AWS Strands Agents SDK：用于构建 AI Agent 的开源 Python 框架，采用模型驱动的编排方式（代码极简，由模型决定如何使用工具）

Amazon Bedrock AgentCore：用于在生产环境中部署、运行和扩展 Agent 的企业平台

两者的关系：Strands SDK 可以独立运行，也可以搭配 AgentCore 平台服务使用。AgentCore 是可选项，但能提供企业级功能（8 小时运行时、流式传输、记忆、身份、可观测性）。

---

快速入门决策树

你要构建什么？

单一用途的 Agent：

• 事件驱动（S3、SQS、定时）→ 部署到 Lambda
• 交互式且需要流式传输 → AgentCore Runtime
• API 端点（无状态）→ Lambda

多 Agent 系统：

• 确定性工作流 → Graph 模式
• 自主协作 → Swarm 模式
• 简单委派 → Agent-as-Tool 模式

工具/集成服务（MCP）：

• 始终部署到 ECS/Fargate 或 AgentCore Runtime
• 绝不使用 Lambda（有状态，需要持久连接）

部署示例请参阅 architecture.md。

---

关键限制

MCP 服务要求

1. 传输方式：必须使用 streamable-http（不能使用 stdio）
2. 端点：必须位于 0.0.0.0:8000/mcp
3. 部署方式：必须部署到 ECS/Fargate 或 AgentCore Runtime（绝不能使用 Lambda）
4. 请求头：必须接受 application/json 和 text/event-stream

原因：MCP 服务有状态，需要持久连接。Lambda 是临时的，不适合此用途。

详情请参阅 limitations.md。

工具数量限制

• 模型难以处理超过 50-100 个工具
• 解决方案：实现语义搜索，以动态加载工具

实现方法请参阅 patterns.md。

令牌管理

• Claude 4.5：200K 上下文（最多使用约 180K）
• 长对话需要对话管理器
• 多 Agent 会使成本增加 5-10 倍

策略请参阅 limitations.md。

---

部署决策矩阵

组件	Lambda	ECS/Fargate	AgentCore Runtime
无状态 Agent	✅ 非常适合	❌ 过度配置	❌ 过度配置
交互式 Agent	❌ 不支持流式传输	⚠️ 可行	✅ 理想选择
MCP 服务	❌ 绝不使用	✅ 标准选择	✅ 支持更多功能
运行时长	< 15 分钟	不限	最长 8 小时
冷启动	有（30-60 秒）	无	无

---

多 Agent 模式选择

模式	复杂度	可预测性	成本	使用场景
单 Agent	低	高	1x	大多数任务
Agent as Tool	低	高	2-3x	简单委派
Graph	高	非常高	3-5x	确定性工作流
Swarm	中	低	5-8x	自主协作

建议：从单 Agent 开始，根据需要逐步扩展。

示例请参阅 architecture.md。

---

何时阅读参考文件

patterns.md

• 基础 Agent 工厂模式（可复用组件）
• MCP 服务注册表模式（工具目录）
• 语义工具搜索（超过 50 个工具时）
• 工具设计最佳实践
• 安全模式
• 测试模式

observability.md

• AWS AgentCore 可观测性平台设置
• 托管于运行时与自托管配置
• 多轮对话的会话跟踪
• OpenTelemetry 设置
• 成本跟踪钩子
• 生产环境可观测性模式

evaluations.md

• AWS AgentCore Evaluations - 使用 LLM-as-a-Judge 进行质量评估
• 13 个内置评估器（Helpfulness、Correctness、GoalSuccessRate 等）
• 使用自定义提示词和模型创建自定义评估器
• 在线（持续）和按需评估模式
• CloudWatch 集成和告警

limitations.md

• MCP 服务部署问题
• 工具选择问题（> 50 个工具）
• Token 溢出
• Lambda 限制
• 多 Agent 成本问题
• 限流错误
• 冷启动延迟

---

#-Driven 理念

核心概念：Strands Agents 将编排交由模型处理，而不是要求编写显式控制流代码。

# Traditional: Manual orchestration (avoid)
while not done:
    if needs_research:
        result = research_tool()
    elif needs_analysis:
        result = analysis_tool()

# Strands: Model decides (prefer)
agent = Agent(
    system_prompt="You are a research analyst. Use tools to answer questions.",
    tools=[research_tool, analysis_tool]
)
result = agent("What are the top tech trends?")
 automatically orchestrates: research_tool → analysis_tool → respond

---

选择

主要提供商：通过 AWS Bedrock 使用 Anthropic Claude

模型 ID 格式：anthropic.claude-{model}-{version}

当前模型（截至 2025 年 1 月）：

• anthropic.claude-sonnet-4-5-20250929-v1:0 - 生产环境
• anthropic.claude-haiku-4-5-20251001-v1:0 - 快速／经济
• anthropic.claude-opus-4-5-20250514-v1:0 - 复杂推理

查看最新模型：

aws bedrock list-foundation-models --by-provider anthropic \
  --query 'modelSummaries[*].[modelId,modelName]' --output table

---

快速示例

基础 Agent

from strands import Agent
from strands.models import BedrockModel
from strands.session import DynamoDBSessionManager
from strands.agent.conversation_manager import SlidingWindowConversationManager

agent = Agent(
    agent_id="my-agent",
    model=BedrockModel(model_id="anthropic.claude-sonnet-4-5-20250929-v1:0"),
    system_prompt="You are helpful.",
    tools=[tool1, tool2],
    session_manager=DynamoDBSessionManager(table_name="sessions"),
    conversation_manager=SlidingWindowConversationManager(max_messages=20)
)

result = agent("Process this request")

基础 Agent 工厂模式请参阅 patterns.md。

MCP 服务（ECS/Fargate）

from mcp.server import FastMCP
import psycopg2.pool

# Persistent connection pool (why Lambda won't work)
db_pool = psycopg2.pool.SimpleConnectionPool(minconn=1, maxconn=10, host="db.internal")

mcp = FastMCP("Database Tools")

@mcp.tool()
def query_database(sql: str) -> dict:
    conn = db_pool.getconn()
    try:
        cursor = conn.cursor()
        cursor.execute(sql)
        return {"status": "success", "rows": cursor.fetchall()}
    finally:
        db_pool.putconn(conn)

# CRITICAL: streamable-http mode
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

部署详情请参阅 architecture.md。

工具错误处理

from strands import tool

@tool
def safe_tool(param: str) -> dict:
    """Always return structured results, never raise exceptions."""
    try:
        result = operation(param)
        return {"status": "success", "content": [{"text": str(result)}]}
    except Exception as e:
        return {"status": "error", "content": [{"text": f"Failed: {str(e)}"}]}

工具设计模式请参阅 patterns.md。

可观测性

AgentCore Runtime（自动）：

# Install with OTEL support
# pip install 'strands-agents[otel]'
# Add 'aws-opentelemetry-distro' to requirements.txt

from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()
agent = Agent(...)  # Automatically instrumented

@app.entrypoint
def handler(payload):
    return agent(payload["prompt"])

自托管：

export AGENT_OBSERVABILITY_ENABLED=true
export OTEL_PYTHON_DISTRO=aws_distro
export OTEL_RESOURCE_ATTRIBUTES="service.name=my-agent"

opentelemetry-instrument python agent.py

通用 OpenTelemetry：

from strands.observability import StrandsTelemetry

# Development
telemetry = StrandsTelemetry().setup_console_exporter()

# Production
telemetry = StrandsTelemetry().setup_otlp_exporter()

详细模式请参阅 observability.md。

---

会话存储选择

Local dev         → FileSystem
Lambda agents     → S3 or DynamoDB
ECS agents        → DynamoDB
Interactive chat  → AgentCore Memory
Knowledge bases   → AgentCore Memory

存储后端比较请参阅 architecture.md。

---

何时使用 AgentCore Platform 与仅使用 SDK

仅使用 Strands SDK

• 简单、无状态的 Agent
• 需要严格控制成本
• 不需要企业级功能
• 希望灵活部署

使用 Strands SDK + AgentCore Platform

• 需要支持 8 小时运行时
• 需要流式响应
• 需要企业级安全／合规
• 需要跨会话智能
• 希望使用托管基础设施

平台服务详情请参阅 architecture.md。

---

常见反模式

1. ❌ 为 Agent 配置过多工具（> 50 个） → 使用语义搜索
2. ❌ 没有对话管理 → 实现 SlidingWindow 或 Summarising
3. ❌ 将 MCP 服务部署到 Lambda → 使用 ECS/Fargate
4. ❌ 没有配置超时 → 在所有位置设置执行限制
5. ❌ 忽略 Token 限制 → 实现对话管理器
6. ❌ 没有成本监控 → 从一开始就实施成本跟踪

详情请参阅 patterns.md 和 limitations.md。

---

生产环境检查清单

部署前：

• 已配置对话管理
• 已启用 AgentCore Observability 或配置 OpenTelemetry
• 已配置 AgentCore Evaluations 以监控质量
• 已实现可观测性钩子
• 已启用成本跟踪
• 所有工具均已配置错误处理
• 已验证安全权限
• MCP 服务已部署到 ECS/Fargate
• 已设置超时限制
• 已配置会话后端（生产环境使用 DynamoDB）
• 已配置 CloudWatch 告警

---

参考文件导航

• architecture.md - 部署模式、多 Agent 编排、会话存储、AgentCore 服务
• patterns.md - 基础组件、工具设计、安全性、测试、性能优化
• limitations.md - 已知限制、变通方案、缓解策略、挑战
• observability.md - AgentCore Observability 平台、ADOT、GenAI 仪表板、OpenTelemetry、钩子、成本跟踪
• evaluations.md - AgentCore Evaluations、内置评估器、自定义评估器、质量监控

---

要点总结

1. MCP 服务必须使用 streamable-http，绝不能使用 Lambda
2. 工具超过 15 个时，使用语义搜索
3. 始终实现对话管理
4. 多 Agent 的成本会增加 5-10 倍（从第一天起就跟踪）
5. 在所有地方设置超时限制
6. 工具中的错误处理是必不可少的
7. 无状态场景使用 Lambda，交互式场景使用 AgentCore
8. 在生产环境中使用 AgentCore Observability 和 Evaluations
9. 从简单开始，逐步增加复杂度
10. 默认保障安全
11. 将配置与代码分离