Doramagic 项目包 · 项目说明书

sdk-python 项目

生成时间:2026-05-28 08:56:25 UTC

项目概述

Strands Agents 是一个简单而强大的 SDK,采用模型驱动的方式来构建和运行 AI 智能体(Agent)。从简单的对话助手到复杂的自主工作流,从本地开发到生产部署,Strands Agents 都能满足您的需求。资料来源:README.md

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 核心模块结构

继续阅读本节完整说明和来源证据。

章节 安装

继续阅读本节完整说明和来源证据。

章节 基本使用

继续阅读本节完整说明和来源证据。

核心特性

Strands Agents SDK 提供以下核心功能:

特性类别说明
轻量级与灵活简洁的智能体循环,开箱即用,支持完全自定义
模型无关支持 Amazon Bedrock、Anthropic、Gemini、LiteLLM、Llama、Ollama、OpenAI、Writer 等多种模型提供商
高级能力多智能体系统、自主智能体、流式输出支持
内置 MCP原生支持 Model Context Protocol (MCP) 服务器,访问数千种预构建工具

资料来源:README.md

系统架构

Strands Agents 的架构设计遵循清晰的模块化原则,主要包含以下几个核心组件:

graph TD
    A[用户应用] --> B[Agent 核心]
    B --> C[事件循环 Event Loop]
    B --> D[模型层 Model Layer]
    B --> E[工具层 Tools Layer]
    B --> F[遥测层 Telemetry]
    
    E --> G[内置工具]
    E --> H[MCP 客户端]
    H --> I[MCP 服务器]
    
    D --> J[BedrockModel]
    D --> K[GeminiModel]
    D --> L[OllamaModel]
    D --> M[OpenAIModel]
    
    F --> N[Tracer Provider]
    F --> O[Meter Provider]

核心模块结构

模块路径功能说明
strands.agent智能体核心实现,包含 Agent 类和事件循环
strands.models模型提供商,支持多种 LLM 后端
strands.tools工具系统,包括 MCP 客户端和执行器
strands.telemetry可观测性支持,追踪和指标收集
strands.types类型定义和异常类

资料来源:strands-py/src/strands/__init__.py

快速开始

安装

pip install strands-agents strands-agents-tools

基本使用

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])
agent("What is the square root of 1764")

资料来源:README.md

MCP 集成

Strands Agents 原生支持 Model Context Protocol (MCP),可以无缝集成 MCP 服务器:

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

aws_docs_client = MCPClient(
    lambda: stdio_client(StdioServerParameters(command="uvx", args=["awslabs.aws-documentation-mcp-server@latest"]))
)

with aws_docs_client:
    agent = Agent(tools=aws_docs_client.list_tools_sync())
    response = agent("Tell me about Amazon Bedrock")

资料来源:README.md

MCP 任务机制

MCP 支持任务增强型工具执行,适用于长时间运行的工具。执行流程如下:

sequenceDiagram
    participant Client as MCPClient
    participant Session as MCPSession
    participant Server as MCPServer
    
    Client->>Server: 1. experimental.call_tool_as_task()
    Server-->>Client: 返回 task_id
    Client->>Server: 2. experimental.poll_task(task_id)
    Server-->>Client: 任务状态更新 (progress)
    Server-->>Client: 任务状态更新 (completed/failed)
    Client->>Server: 3. get_task_result(task_id)
    Server-->>Client: 最终结果

任务配置选项:

参数类型默认值说明
ttltimedelta1 分钟任务生存时间
poll_timeouttimedelta5 分钟轮询超时时间

资料来源:strands-py/src/strands/tools/mcp/mcp_tasks.py:18-32

模型提供商

支持的模型

提供商模型类特点
Amazon BedrockBedrockModel支持 Nova、Claude 等模型,含服务层级支持
Google GeminiGeminiModel支持 Gemini 2.5 Flash 等
OllamaOllamaModel本地模型支持
OpenAIOpenAIModel标准 OpenAI API
LiteLLMLiteLLMModel统一接口访问多种模型

Bedrock 示例

from strands import Agent
from strands.models import BedrockModel

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    temperature=0.3,
    streaming=True,
)
agent = Agent(model=bedrock_model)
agent("Tell me about Agentic AI")

Gemini 示例

from strands import Agent
from strands.models.gemini import GeminiModel

gemini_model = GeminiModel(
    client_args={"api_key": "your_gemini_api_key"},
    model_id="gemini-2.5-flash",
    params={"temperature": 0.7}
)
agent = Agent(model=gemini_model)

资料来源:README.md

工具执行器

Strands Agents 提供灵活的工具体系,支持不同的执行策略:

并发执行器

from strands.tools.executors import ConcurrentToolExecutor

executor = ConcurrentToolExecutor(max_workers=4)
agent = Agent(tools=[tool1, tool2], tool_executor=executor)

顺序执行器

from strands.tools.executors import SequentialToolExecutor

executor = SequentialToolExecutor()
agent = Agent(tools=[tool1, tool2], tool_executor=executor)

资料来源:strands-py/src/strands/tools/executors/__init__.py

可观测性与遥测

Strands Agents 内置 OpenTelemetry 支持,提供追踪和指标收集功能:

from strands.telemetry import StrandsTelemetry

# 快速设置
StrandsTelemetry().setup_console_exporter().setup_otlp_exporter()

# 启用指标收集
StrandsTelemetry().setup_meter(enable_console_exporter=True, enable_otlp_exporter=True)

核心指标

指标名称类型说明
strands.event_loop.cycle_countCounter事件循环周期数
strands.event_loop.latencyHistogram事件循环延迟
strands.tool.call_countCounter工具调用次数
strands.tool.durationHistogram工具执行时长

资料来源:strands-py/src/strands/telemetry/config.py:30-60 资料来源:strands-py/src/strands/telemetry/metrics_constants.py

事件循环机制

Agent 的核心执行流程通过事件循环实现:

graph TD
    A[用户输入] --> B[开始事件循环周期]
    B --> C{停止原因判断}
    
    C -->|end_turn| D[返回结果]
    C -->|tool_use| E[执行工具]
    E --> F{工具执行结果}
    F -->|需要更多工具| C
    F -->|完成| D
    C -->|max_tokens| G[抛出 MaxTokensReachedException]

资料来源:strands-py/src/strands/event_loop/event_loop.py

开发者指南

本地开发

cd strands-py
pip install hatch
hatch test        # 运行单元测试
hatch fmt         # 格式化和 lint

项目结构

sdk-python/
├── strands-py/              # 主要 SDK 代码
│   └── src/strands/
│       ├── agent/           # Agent 核心
│       ├── models/          # 模型提供商
│       ├── tools/           # 工具系统
│       ├── telemetry/       # 遥测
│       └── event_loop/      # 事件循环
├── site/                    # 文档站点
└── README.md

资料来源:README.md

版本信息

当前最新版本:v1.41.0

近期主要更新:

版本主要特性
v1.41.0MultiAgentPlugin 支持 Swarm 和 Graph 编排器;Bedrock 缓存 TTL 支持
v1.40.0主动上下文压缩;Token 计数 AccessDenied 错误缓存
v1.39.0OpenAI 提供商支持 AWS Profile;MCPClientInitializationError 根因信息
v1.38.0MCP CallToolResult.isError 标志保留;tiktoken 令牌计数估计
v1.37.0滑动窗口对话管理器回退点;实验性检查点支持
v1.36.0Agent 构造器支持可调用 Hook 回调;Bedrock 服务层级支持

资料来源:README.md

已知问题与限制

社区报告的已知问题:

Issue问题描述状态
#2351Python wheel 发布包缺少 README 和 LICENSE待修复
#2347GeminiModel 在 safety_settings 阻止内容时因 usage_metadata=None 崩溃待修复
#2342MetricsClient 单例非线程安全待修复
#1812MCP 任务进度更新功能缺失功能请求

相关链接

另请参阅

资料来源:README.md

快速开始

本页面介绍 Strands Agents Python SDK 的快速上手流程,帮助开发者从零开始创建和运行 AI Agent。通过本指南,您将了解如何安装 SDK、创建基础 Agent、配置模型提供商、以及集成工具和 MCP 服务器。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 基础安装

继续阅读本节完整说明和来源证据。

章节 虚拟环境

继续阅读本节完整说明和来源证据。

章节 开发安装

继续阅读本节完整说明和来源证据。

环境要求

在开始之前,请确保您的开发环境满足以下要求:

要求最低版本说明
Python3.10+SDK 需要 Python 3.10 或更高版本
pip最新版本用于安装 SDK 包
AWS 凭证-仅在使用 Bedrock 模型时需要

Strands Agents SDK 采用模块化设计,核心包和工具包分开发布,可按需安装:

  • strands-agents:核心 Agent 框架
  • strands-agents-tools:内置工具集

安装

基础安装

使用 pip 安装 Strands Agents SDK 的最简方式:

pip install strands-agents strands-agents-tools

此安装方式包含核心框架和常用内置工具,如计算器、网页搜索、文件操作等。

虚拟环境

建议在虚拟环境中安装以避免依赖冲突:

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Linux/macOS:
source venv/bin/activate
# Windows:
.\venv\Scripts\activate

# 安装 SDK
pip install strands-agents strands-agents-tools

开发安装

如需参与 SDK 开发,可从源码安装:

cd strands-py
pip install -e .

开发依赖可通过 hatch 工具管理,包括测试和代码格式化:

pip install hatch
hatch test        # 运行单元测试
hatch fmt         # 格式化和 lint

创建第一个 Agent

基础示例

以下是创建最简单 Agent 的示例,使用内置的计算器工具:

from strands import Agent
from strands_tools import calculator

# 创建 Agent 并传入工具列表
agent = Agent(tools=[calculator])

# 向 Agent 发送消息
response = agent("What is the square root of 1764")
print(response)

此示例展示了 Strands Agents 的核心设计理念:简单的 Agent 循环,完全可定制化。

使用默认模型

默认情况下,Agent 使用 Amazon Bedrock 的 Claude 4 Sonnet 模型:

注意:使用默认的 Amazon Bedrock 模型提供商需要配置 AWS 凭证,并在 us-west-2 区域启用 Claude 4 Sonnet 模型的访问权限。

配置 AWS 凭证的常用方式:

  1. 环境变量
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-west-2"
  1. AWS 配置文件
aws configure
  1. IAM 角色(在 EC2、Lambda 或 ECS 环境中自动获取)

模型配置

Strands Agents 支持多种模型提供商,可通过 model 参数指定不同的模型。

Amazon Bedrock

Bedrock 是 AWS 提供的托管式 AI 服务,Strands Agents 提供原生支持:

from strands import Agent
from strands.models import BedrockModel

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    temperature=0.3,
    streaming=True,  # 启用流式输出
)

agent = Agent(model=bedrock_model)
response = agent("Tell me about Agentic AI")

支持的 Bedrock 模型包括:

  • Amazon Nova 系列
  • Claude 4 Sonnet
  • Claude 3.5 Sonnet
  • Claude 3 Haiku
  • Llama 系列

Google Gemini

使用 Gemini 模型时需要提供 API 密钥:

from strands import Agent
from strands.models.gemini import GeminiModel

gemini_model = GeminiModel(
    client_args={
        "api_key": "your_gemini_api_key",
    },
    model_id="gemini-2.5-flash",
    params={"temperature": 0.7}
)

agent = Agent(model=gemini_model)
response = agent("Tell me about Agentic AI")

Ollama(本地模型)

Ollama 允许在本地运行开源大语言模型:

from strands import Agent
from strands.models.ollama import OllamaModel

ollama_model = OllamaModel(
    model_id="llama3.2:latest",
    base_url="http://localhost:11434",
    params={"temperature": 0.7}
)

agent = Agent(model=ollama_model)
response = agent("What can you help me with?")

其他支持的提供商

提供商模型类型配置说明
AnthropicClaude 系列需要 API 密钥
OpenAIGPT 系列需要 API 密钥
LiteLLM多提供商统一接口支持 100+ 种模型
LlamaAPILlama 系列需要 API 密钥
LlamaCpp本地量化模型需要本地部署
WriterWriter 自有模型需要 API 密钥

工具集成

内置工具

Strands Agents 提供丰富的内置工具集,通过 strands-agents-tools 包提供:

from strands import Agent
from strands_tools import calculator, current_date, python_executor

agent = Agent(tools=[calculator, current_date, python_executor])

response = agent(
    "What is today's date, and calculate 15% tip on a $47.50 bill"
)

常用内置工具包括:

工具名称功能描述
calculator数学计算
current_date获取当前日期时间
python_executor执行 Python 代码
strands-file-read读取文件内容
strands-file-write写入文件内容
strands-bash执行 Shell 命令

自定义工具

开发者可以创建自定义工具函数:

from strands import Agent
from strands.types import ToolResult

def get_weather(location: str) -> ToolResult:
    """获取指定位置的天气信息"""
    # 实际实现中应调用天气 API
    weather_data = {"location": location, "temperature": "22°C", "condition": "晴"}
    
    return ToolResult(
        status="success",
        content=[{"text": f"{location}今天天气:{weather_data['condition']},气温{weather_data['temperature']}"}]
    )

agent = Agent(tools=[get_weather])
response = agent("What's the weather like in Beijing?")

MCP 服务器集成

Model Context Protocol (MCP) 是一种开放协议,允许 AI 模型与外部数据源和工具连接。Strands Agents 原生支持 MCP 服务器。

基础连接

以下示例连接 AWS 文档 MCP 服务器:

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# 创建 MCP 客户端
aws_docs_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(
            command="uvx",
            args=["awslabs.aws-documentation-mcp-server@latest"]
        )
    )
)

# 使用上下文管理器确保连接正确关闭
with aws_docs_client:
    agent = Agent(tools=aws_docs_client.list_tools_sync())
    response = agent(
        "Tell me about Amazon Bedrock and how to use it with Python"
    )

MCP 任务支持

MCP 支持任务增强的工具执行,适用于长时间运行的工具调用。任务执行流程如下:

graph TD
    A[开始工具调用] --> B{服务器支持任务?}
    B -->|是| C[创建任务 call_tool_as_task]
    B -->|否| D[直接调用工具 call_tool]
    C --> E[轮询任务状态 poll_task]
    E --> F{任务完成?}
    F -->|否| E
    F -->|是| G[获取结果 get_task_result]
    D --> H[返回结果]
    G --> H

配置 MCP 任务参数:

from strands.tools.mcp import MCPClient
from strands.tools.mcp.mcp_tasks import TasksConfig
from datetime import timedelta

# 配置任务执行参数
tasks_config = TasksConfig(
    ttl=timedelta(minutes=2),  # 任务生存时间
    poll_timeout=timedelta(minutes=10)  # 轮询超时
)

mcp_client = MCPClient(
    transport_callable=...,
    tasks_config=tasks_config
)

工具过滤

MCP 客户端支持按名称模式过滤可用工具:

from strands.tools.mcp import MCPClient
import re

# 只允许特定工具
mcp_client = MCPClient(
    transport_callable=...,
    tool_filters=[re.compile(r"^aws_.*")]  # 只允许 aws_ 开头的工具
)

多模态交互

Strands Agents 支持多模态输入输出,包括文本、音频和图像:

from strands import Agent
from strands.types import AgentStreamEvent
from strands.models import BedrockModel

# 配置多模态模型
model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    streaming=True,
)

agent = Agent(model=model)

# 配置音频和文本输入输出
async def stream_response():
    async for event in agent.run_async(
        inputs=[audio_io.input(), text_io.input()],  # 语音或文本输入
        outputs=[audio_io.output(), text_io.output()]  # 语音或文本输出
    ):
        if isinstance(event, AgentStreamEvent):
            print(event.content)

遥测和监控

Strands Agents 内置 OpenTelemetry 集成,支持追踪和指标收集:

from strands.telemetry import StrandsTelemetry

# 初始化遥测配置
telemetry = StrandsTelemetry()

# 配置控制台导出器
telemetry.setup_console_exporter()

# 配置 OTLP 导出器(发送到遥测后端)
telemetry.setup_otlp_exporter()

# 启用指标收集
telemetry.setup_meter(enable_console_exporter=True, enable_otlp_exporter=True)

环境变量配置

遥测功能支持通过环境变量配置:

环境变量说明示例
OTEL_EXPORTER_OTLP_ENDPOINTOTLP 端点 URLhttp://localhost:4317
OTEL_EXPORTER_OTLP_HEADERSOTLP 请求头X-api-key=xxx
OTEL_SERVICE_NAME服务名称my-agent-service

常见问题

AWS 凭证问题

问题:使用 Bedrock 模型时出现 AccessDeniedException

解决方案

  1. 确认 AWS 凭证已正确配置
  2. 检查 IAM 用户/角色是否有 Bedrock 访问权限
  3. 确认目标区域(us-west-2)已启用所需模型

MCP 连接问题

问题:MCP 服务器连接失败

解决方案

  1. 检查 MCP 服务器命令是否正确安装(如 uvx
  2. 确认服务器参数配置正确
  3. 查看服务器端日志排查问题

模型兼容性问题

问题:Gemini 模型出现 TypeError

解决方案

  1. 确保使用最新版本的 SDK
  2. 检查 safety_settings 配置是否正确

下一步

完成快速开始后,建议进一步阅读以下内容:

相关链接

来源:https://github.com/strands-agents/sdk-python / 项目说明书

Agent 系统架构

Strands Agents SDK 的核心是 Agent 系统,这是一个模型驱动的 AI Agent 框架,采用简单而强大的设计理念。本文档详细介绍 Agent 的架构设计、核心组件、执行流程以及扩展机制。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 2.1 Agent 类

继续阅读本节完整说明和来源证据。

章节 2.2 AgentBase 抽象基类

继续阅读本节完整说明和来源证据。

章节 2.3 状态管理 (StateManager)

继续阅读本节完整说明和来源证据。

1. 架构概述

Strands Agents 采用模块化架构设计,将 AI Agent 的核心功能分解为多个协同工作的组件。系统以 Agent 为中心,通过事件循环驱动整个执行流程,并支持多种工具执行策略和扩展机制。

graph TD
    subgraph "Agent 核心层"
        A[Agent]
        B[AgentBase]
        C[EventLoop]
        D[StateManager]
    end
    
    subgraph "工具层"
        E[工具注册表]
        F[工具执行器]
        G[MCP 客户端]
    end
    
    subgraph "模型层"
        H[Model Provider]
        I[Structured Output]
    end
    
    subgraph "扩展层"
        J[Hooks 系统]
        K[Plugins 插件]
        L[多 Agent 协作]
    end
    
    A --> B
    B --> C
    C --> D
    C --> F
    F --> E
    F --> G
    C --> H
    A --> J
    A --> K
    K --> L

资料来源:strands-py/src/strands/agent/agent.py:1-100

2. 核心组件

2.1 Agent 类

Agent 是整个 SDK 的主入口点,负责初始化和管理 AI Agent 的生命周期。它封装了模型配置、工具注册、会话管理和事件处理等核心功能。

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])
response = agent("What is 2 + 2?")

Agent 类的主要职责包括:

  • 管理 Agent 的初始化和生命周期
  • 注册和管理工具集合
  • 处理用户输入并生成响应
  • 管理对话历史和状态

资料来源:strands-py/src/strands/agent/agent.py:1-50

2.2 AgentBase 抽象基类

AgentBase 定义了所有 Agent 实现必须遵循的接口规范,确保不同 Agent 实现之间的一致性。

属性/方法类型说明
modelModel配置的模型实例
toolslist[AgentTool]注册的工具列表
system_prefixstr \None系统提示前缀
callbackslist[Callback]回调函数列表
create_stream_events()方法创建流式事件生成器

资料来源:strands-py/src/strands/agent/base.py:1-80

2.3 状态管理 (StateManager)

状态管理器负责维护 Agent 在执行过程中的运行时状态,包括调用栈、工具执行结果和中间结果等。

classDiagram
    class StateManager {
        +invocation_state: dict
        +tool_results: list
        +trace: Trace
        +span: Any
        +reset()
        +update_state(key, value)
        +get_state(key)
    }

资料来源:strands-py/src/strands/agent/state.py:1-60

3. 事件循环机制

3.1 EventLoop 核心流程

事件循环是 Agent 执行的大脑,它驱动整个 Agent 的决策-执行循环。EventLoop 协调模型调用、工具执行和状态管理。

sequenceDiagram
    participant User as 用户输入
    participant Loop as EventLoop
    participant Model as 模型
    participant Tools as 工具执行器
    participant Result as AgentResult

    User->>Loop: 提交请求
    Loop->>Model: 调用模型
    alt 模型决定使用工具
        Model-->>Loop: tool_use 决策
        Loop->>Tools: 执行工具
        Tools-->>Loop: 工具结果
        Loop->>Model: 提交结果继续推理
    else 模型决定结束
        Model-->>Loop: end_turn 决策
    end
    Loop-->>Result: 生成最终结果

资料来源:strands-py/src/strands/event_loop/event_loop.py:1-100

3.2 停止原因 (Stop Reasons)

事件循环根据模型的决策确定何时停止,每种停止原因都有特定的处理逻辑:

停止原因说明处理方式
end_turn对话轮次结束返回最终响应给用户
tool_use模型决定调用工具执行工具并继续循环
max_tokens达到令牌限制抛出 MaxTokensReachedException
interrupt需要用户输入中断执行等待用户响应

资料来源:strands-py/src/strands/types/event_loop.py:1-50

3.3 工具执行流程

当模型决定使用工具时,EventLoop 调用工具执行子系统:

graph LR
    A[tool_use 决策] --> B{执行策略}
    B -->|并发| C[ConcurrentToolExecutor]
    B -->|顺序| D[SequentialToolExecutor]
    C --> E[执行工具 1<br/>执行工具 2<br/>执行工具 N]
    D --> F[执行工具 1<br/>等待完成<br/>执行工具 2]
    E --> G[收集结果]
    F --> G
    G --> H[返回给模型]

资料来源:strands-py/src/strands/tools/executors/__init__.py:1-30

4. 工具执行器

4.1 执行器类型

Strands Agents 提供了两种主要的工具执行策略:

执行器类名适用场景
顺序执行SequentialToolExecutor工具间有依赖关系,必须按顺序执行
并发执行ConcurrentToolExecutor工具相互独立,可并行执行以提高效率
from strands import Agent
from strands.tools.executors import ConcurrentToolExecutor, SequentialToolExecutor

# 使用并发执行器
agent = Agent(
    tools=[tool1, tool2, tool3],
    tool_executor=ConcurrentToolExecutor()
)

# 使用顺序执行器
agent = Agent(
    tools=[dependent_tool1, dependent_tool2],
    tool_executor=SequentialToolExecutor()
)

资料来源:strands-py/src/strands/tools/executors/__init__.py:1-25

4.2 并发执行器实现

ConcurrentToolExecutor 实现了异步并发工具执行,每个工具在独立的任务中运行:

async def _execute_single_tool(
    agent: Any,
    tool_use: ToolUse,
    tool_results: list[ToolResult],
    cycle_trace: Trace,
    cycle_span: Any,
    invocation_state: dict[str, Any],
    task_id: int,
    task_queue: asyncio.Queue,
    task_event: asyncio.Event,
    stop_event: object,
    structured_output_context: "StructuredOutputContext | None",
) -> None:
    """执行单个工具并将结果放入任务队列。"""

资料来源:strands-py/src/strands/tools/executors/concurrent.py:1-80

5. Hooks 扩展系统

5.1 Hook 事件类型

Hooks 系统提供了在 Agent 执行生命周期中注入自定义逻辑的能力。通过订阅特定事件,开发者可以在关键时刻执行自定义代码。

事件类型触发时机常见用途
on_tool_call工具调用前日志记录、参数验证
on_tool_result工具执行后结果处理、监控
on_message消息生成后内容过滤、转换
on_cycle_start循环周期开始状态初始化
on_cycle_end循环周期结束结果处理

资料来源:strands-py/src/strands/hooks/events.py:1-60

5.2 Hook 回调配置

在 Agent 构造函数中传入回调函数:

from strands import Agent
from strands.hooks import AgentHooks

def my_tool_logger(tool_name: str, arguments: dict):
    print(f"Tool called: {tool_name} with {arguments}")

agent = Agent(
    tools=[...],
    callbacks=[
        ("on_tool_call", my_tool_logger)
    ]
)

资料来源:strands-py/src/strands/agent/agent.py:100-150

6. Agent 结果处理

6.1 AgentResult 结构

AgentResult 封装了 Agent 执行后的完整结果,包括消息内容、停止原因和工具使用记录:

@dataclass
class AgentResult:
    """Agent 执行结果的容器。"""
    
    messages: list[Message]          # 对话消息历史
    stop_reason: StopReason          # 停止原因
    usage: dict                      # 令牌使用统计
    tool_results: list[ToolResult]   # 工具执行结果

资料来源:strands-py/src/strands/agent/agent_result.py:1-50

6.2 结果状态映射

对于多 Agent 协作场景,系统会将远程 Agent 的状态映射到本地结果:

A2A 状态本地 stop_reason说明
completedend_turn任务成功完成
failedend_turn (带错误)任务执行失败
canceledend_turn任务被取消
input_requiredinterrupt需要用户输入
auth_requiredinterrupt需要身份验证

资料来源:strands-py/src/strands/multiagent/a2a/_converters.py:1-60

7. 插件系统

7.1 插件架构

Strands Agents 支持通过插件扩展功能,插件可以在 Agent 初始化时注册,提供额外的工具、工具执行器或其他扩展功能。

graph TD
    P[Plugin] -->|注册| A[Agent]
    P -->|提供| T[Tools]
    P -->|提供| E[Executor]
    P -->|提供| H[Hooks]
    
    subgraph "内置插件"
        M[MultiAgentPlugin]
        S[AgentSkills]
    end

7.2 MultiAgentPlugin

v1.41.0 版本引入了 MultiAgentPlugin,支持 Swarm 和 Graph 编排器,用于构建复杂的多 Agent 系统:

from strands.vended_plugins import MultiAgentPlugin

plugin = MultiAgentPlugin(
    orchestrator="swarm",  # 或 "graph"
    agents=[agent1, agent2]
)

资料来源:strands-py/src/strands/vended_plugins/__init__.py:1-30

8. 配置选项汇总

8.1 Agent 构造函数参数

参数类型默认值说明
modelModel必需模型实例或模型配置
toolslist[AgentTool][]可用工具列表
system_prefixstr \NoneNone系统提示前缀
tool_executorToolExecutorSequentialToolExecutor工具执行器
callbackslist[Callback][]回调函数列表
pluginslist[Plugin][]插件列表
max_tokensint \NoneNone最大输出令牌数

资料来源:strands-py/src/strands/agent/agent.py:50-150

9. 常见故障模式

9.1 工具执行异常

错误类型原因解决方案
ToolProviderException工具提供者初始化失败检查 MCP 服务器配置
MCPClientInitializationErrorMCP 客户端连接失败验证服务器地址和网络连接
MaxTokensReachedException输出超出限制减少对话历史或增加 max_tokens

9.2 线程安全问题

注意:MetricsClient 单例在 v1.41.0 版本前存在线程安全问题。建议在多线程环境中使用时注意同步,或升级到最新版本。

资料来源:https://github.com/strands-agents/sdk-python/issues/2342

10. 执行流程图

flowchart TD
    A[用户输入] --> B{Agent 初始化}
    B -->|首次| C[加载工具]
    B -->|后续| D[检查会话状态]
    C --> E[进入事件循环]
    D --> E
    E --> F[调用模型]
    F --> G{停止原因判断}
    G -->|tool_use| H[执行工具]
    H --> I[收集工具结果]
    I --> E
    G -->|end_turn| J[生成响应]
    G -->|interrupt| K[等待用户输入]
    K --> E
    J --> L[返回 AgentResult]

11. See Also

资料来源:strands-py/src/strands/agent/agent.py:1-100

模型提供商

模型提供商(Model Provider)是 Strands Agents SDK 中的核心组件,负责与各种大语言模型(LLM)服务进行交互。SDK 通过统一的抽象接口,支持 Amazon Bedrock、Anthropic、OpenAI、Google Gemini、Ollama、LiteLLM 等多种模型服务商,同时也支持自定义提供商。这种设计使开发者能够在不修改业务逻辑...

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 核心抽象

继续阅读本节完整说明和来源证据。

章节 消息处理流程

继续阅读本节完整说明和来源证据。

章节 提供商总览

继续阅读本节完整说明和来源证据。

概述

模型提供商(Model Provider)是 Strands Agents SDK 中的核心组件,负责与各种大语言模型(LLM)服务进行交互。SDK 通过统一的抽象接口,支持 Amazon Bedrock、Anthropic、OpenAI、Google Gemini、Ollama、LiteLLM 等多种模型服务商,同时也支持自定义提供商。这种设计使开发者能够在不修改业务逻辑的情况下,轻松切换不同的模型后端。

模型提供商的实现遵循策略模式,每个提供商都继承自基础的 Model 类,实现统一的接口方法。SDK 在 v1.35.0 版本中引入了 Bedrock Service Tier 支持,允许开发者按请求控制延迟与成本之间的权衡。资料来源:v1.35.0 Release

架构设计

核心抽象

Strands SDK 的模型层采用分层架构设计。基础抽象层定义了所有模型提供商必须实现的接口规范,包括消息处理、流式响应生成、工具调用等功能。各提供商在此基础上封装具体的 API 调用逻辑。

graph TD
    A[Agent] --> B[Model 抽象基类]
    B --> C[BedrockModel]
    B --> D[AnthropicModel]
    B --> E[OpenAIModel]
    B --> F[GeminiModel]
    B --> G[OllamaModel]
    B --> H[LiteLLMModel]
    B --> I[自定义 Model]
    
    C --> J[AWS SDK / boto3]
    D --> K[Anthropic API]
    E --> L[OpenAI API]
    F --> M[Google AI API]
    G --> N[Ollama 本地服务]
    H --> O[统一 API 网关]

消息处理流程

模型提供商负责处理对话消息并生成响应。当 Agent 发起请求时,模型提供商将消息格式转换为目标 API 所要求的格式,处理 API 响应,并可选地支持流式输出。

支持的模型提供商

提供商总览

提供商配置复杂度部署方式特色功能
BedrockModel中等云服务AWS 集成、Service Tier、缓存
AnthropicModel云服务工具调用、流式响应
OpenAIModel云服务Azure 支持、AWS Profile
GeminiModel中等云服务多模态、安全设置
OllamaModel本地/远程本地部署、低延迟
LiteLLMModel中等统一网关多提供商聚合
LlamaAPIModel云服务开源模型支持
LlamaCppModel中等本地高性能本地推理

BedrockModel

BedrockModel 是与 Amazon Bedrock 服务集成的实现,支持 AWS 凭证和 profile 配置。该提供商在 v1.35.0 版本中新增了 Service Tier 支持,允许在 Priority、Standard、Flex 三个层级间选择,以平衡延迟与成本。

from strands.models import BedrockModel

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    temperature=0.3,
    streaming=True,
)

BedrockModel 支持 AWS profile 认证,开发者可通过 aws_profile 参数指定要使用的凭证配置。资料来源:strands-py/src/strands/models/bedrock.py

AnthropicModel

AnthropicModel 提供与 Anthropic Claude 模型的直接集成。该提供商封装了 Anthropic Messages API,支持工具调用、图像处理和流式响应。

from strands.models import AnthropicModel

anthropic_model = AnthropicModel(
    model_id="claude-sonnet-4-20250514",
    params={"temperature": 0.7}
)

OpenAIModel

OpenAIModel 支持 OpenAI 的 GPT 系列模型以及兼容的 Azure OpenAI 服务。在 v1.39.0 版本中,OpenAI 提供商新增了 AWS Profile 支持功能。资料来源:v1.39.0 Release

from strands.models import OpenAIModel

openai_model = OpenAIModel(
    model_id="gpt-4o",
    client_args={"api_key": "your-api-key"},
    params={"temperature": 0.5}
)

GeminiModel

GeminiModel 集成 Google Gemini 系列模型,支持多模态输入和高级安全设置。需要注意的是,曾有社区反馈在特定条件下(safety_settings 块内容为 None 且 usage_metadata 缺失时)会触发 TypeError。资料来源:BUG Report #2347

from strands.models import GeminiModel

gemini_model = GeminiModel(
    client_args={"api_key": "your-gemini-api-key"},
    model_id="gemini-2.5-flash",
    params={"temperature": 0.7}
)

OllamaModel

OllamaModel 支持连接本地或远程运行的 Ollama 服务。v1.40.0 版本修复了 ollama 模式下 latencyMs 指标的类型返回值问题。资料来源:v1.40.0 Release

from strands.models import OllamaModel

ollama_model = OllamaModel(
    model_id="llama3.2",
    params={"temperature": 0.8}
)

LiteLLMModel

LiteLLMModel 通过 LiteLLM 统一网关访问多种模型提供商,简化了多提供商切换的复杂度。

from strands.models import LiteLLMModel

litellm_model = LiteLLMModel(
    model_id="anthropic/claude-3-sonnet",
    params={"temperature": 0.5}
)

LlamaAPIModel

LlamaAPIModel 支持通过 LlamaAPI 访问开源模型,提供简洁的远程推理接口。

LlamaCppModel

LlamaCppModel 支持本地高性能推理,适用于对延迟敏感且需要数据隐私的场景。

配置参数

通用参数

所有模型提供商共享一组通用配置参数,这些参数定义了模型的基本行为。

参数类型说明默认值
model_idstr模型标识符必需
paramsdict模型推理参数{}
streamingbool启用流式响应False
client_argsdict底层客户端配置{}

推理参数 (params)

params 字典包含传递给模型推理的参数,具体参数因提供商而异。

参数适用范围说明
temperature通用控制随机性,0-2
max_tokens通用最大输出 token 数
top_p通用核采样概率
system通用系统提示词
stop通用停止序列列表

Service Tier (Bedrock)

BedrockModel 支持特殊的 service_tier 参数,用于控制服务级别:

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    service_tier="priority"  # priority | standard | flex
)
级别延迟成本适用场景
priority最低最高实时交互
standard中等中等常规使用
flex较高最低批量处理

使用模式

默认模型配置

SDK 默认使用 BedrockModel 作为模型提供商,但开发者可以轻松切换到其他提供商:

from strands import Agent
from strands.models import OpenAIModel

# 使用 OpenAI 替代默认的 Bedrock
openai_model = OpenAIModel(
    model_id="gpt-4o",
    client_args={"api_key": "your-api-key"}
)

agent = Agent(model=openai_model)
response = agent("你好,请介绍一下你自己")

模型混用

在复杂应用中,可以根据任务类型选择不同的模型:

from strands.models import AnthropicModel, OpenAIModel

# 复杂推理任务使用 Claude
claude = AnthropicModel(model_id="claude-3-5-sonnet-20241022")

# 快速响应任务使用 GPT
gpt = OpenAIModel(model_id="gpt-4o-mini")

# 根据任务路由到不同模型

流式响应

所有模型提供商支持流式响应,通过 streaming=True 启用:

from strands.models import BedrockModel

streaming_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    streaming=True
)

for event in streaming_model.generate("讲述一个故事"):
    print(event)

Token 计费与限制

上下文窗口限制

在 v1.37.0 和 v1.38.0 版本中,SDK 增加了上下文窗口限制查询表和 count_token 方法支持。开发者可以通过模型接口查询 token 使用量,更好地管理上下文长度。资料来源:v1.38.0 Release

# 查询 token 使用量
token_count = model.count_token("要计算 token 的文本")

上下文压缩

v1.40.0 版本引入了主动上下文压缩功能,Conversation Manager 会在上下文接近限制时自动压缩对话历史。资料来源:v1.40.0 Release

常见问题与解决方案

认证配置

问题:模型调用时出现认证错误

解决方案:确保正确配置凭证

# AWS 凭证 (Bedrock)
import boto3
boto3.setup_default_session(profile_name='your-profile')

# API Key (OpenAI/Gemini)
model = OpenAIModel(
    client_args={"api_key": "sk-..."}
)

模型不可用

问题:模型 ID 不存在或区域不支持

解决方案:检查模型 ID 和区域配置

# 使用正确的模型 ID 和区域
bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",  # 确认区域前缀
)

连接超时

问题:远程模型响应超时

解决方案:配置适当的超时时间

from datetime import timedelta

model = OllamaModel(
    model_id="llama3.2",
    client_args={"timeout": timedelta(seconds=120)}
)

Gemini 安全设置异常

当使用 GeminiModel 并配置安全设置时,如果 safety_settings 块内容为 None 且 usage_metadata 缺失,可能触发 TypeError。这是已知的兼容性问题,建议升级到最新版本。资料来源:BUG Report #2347

自定义模型提供商

开发者可以通过继承基础 Model 类创建自定义模型提供商:

from strands.models import Model
from strands.types import Message

class CustomModel(Model):
    def __init__(self, model_id: str, **kwargs):
        super().__init__(model_id=model_id, **kwargs)
    
    async def _generate_async(self, messages: list[Message], **kwargs):
        # 实现自定义生成逻辑
        pass

基础类定义了所有必需的方法签名,包括消息处理、流式生成、工具调用等。自定义提供商需要正确实现这些接口以确保与 Agent 的兼容性。

相关文档

更新日志

版本更新内容
v1.41.0Starlette 依赖升级至 1.x
v1.40.0主动上下文压缩、AccessDenied 错误缓存优化
v1.39.0OpenAI Provider AWS Profile 支持
v1.38.0Token 计数方法、多模态内容处理改进
v1.37.0上下文窗口限制配置、检查点功能
v1.36.0Hook 回调可调用对象支持
v1.35.0Bedrock Service Tier 支持

来源:https://github.com/strands-agents/sdk-python / 项目说明书

工具系统

Strands Agents 的工具系统是 SDK 的核心组件之一,它为 AI Agent 提供了调用外部工具和执行特定任务的能力。工具系统采用模块化设计,支持多种工具类型、执行策略和协议集成,使开发者能够灵活地扩展 Agent 的功能。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 核心组件

继续阅读本节完整说明和来源证据。

概述

Strands Agents 的工具系统是 SDK 的核心组件之一,它为 AI Agent 提供了调用外部工具和执行特定任务的能力。工具系统采用模块化设计,支持多种工具类型、执行策略和协议集成,使开发者能够灵活地扩展 Agent 的功能。

工具系统的设计遵循以下核心原则:

  • 可组合性:工具可以独立使用,也可以组合成复杂的工具链
  • 灵活性:支持同步和异步执行模式,提供多种执行策略
  • 可扩展性:通过 MCP(Model Context Protocol)协议支持第三方工具服务器
  • 可观测性:内置遥测和指标收集,便于监控和调试

资料来源:strands-py/src/strands/tools/__init__.py

架构概览

graph TD
    subgraph "工具系统架构"
        A["Agent"] --> B["工具注册表<br/>(ToolRegistry)"]
        B --> C["工具执行器<br/>(ToolExecutor)"]
        
        C --> D["顺序执行器<br/>(SequentialToolExecutor)"]
        C --> E["并发执行器<br/>(ConcurrentToolExecutor)"]
        
        F["内置工具<br/>(strands_tools)"] --> B
        G["MCP 工具<br/>(MCPClient)"] --> B
        H["自定义工具<br/>(@tool 装饰器)"] --> B
    end
    
    subgraph "执行流程"
        I["工具调用请求"] --> J["注册表查找"] 
        J --> K["参数验证"]
        K --> L["执行器分发"]
        L --> D
        L --> E
        D --> M["结果处理"]
        E --> M
        M --> N["返回结果"]
    end
    
    style A fill:#e1f5fe
    style B fill:#fff3e0
    style C fill:#e8f5e9
    style M fill:#fce4ec

核心组件

组件文件路径职责
工具注册表strands/tools/registry.py管理工具注册、查找和元数据存储
工具装饰器strands/tools/decorator.py提供 @tool 装饰器用于定义工具
工具基类strands/tools/tools.py定义 Tool 基类和工具结果类型
执行器基类strands/tools/executors/_executor.py定义执行器的抽象接口
顺序执行器strands/tools/executors/sequential.py按顺序执行工具调用
并发执行器strands/tools/executors/concurrent.py并行执行多个工具调用
MCP 客户端strands/tools/mcp/mcp_client.py连接 MCP 服务器并暴露工具

资料来源:strands-py/src/strands/tools/registry.py

资料来源:strands-py/src/strands/tools/__init__.py

会话与上下文管理

本文档详细介绍 Strands Agents SDK 中会话(Session)与上下文管理(Context Management)系统的架构、组件和使用方式。这些系统共同确保 AI Agent 能够有效地处理多轮对话、维护状态并智能地管理上下文窗口。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 SessionManager 基类

继续阅读本节完整说明和来源证据。

章节 FileSessionManager 实现

继续阅读本节完整说明和来源证据。

章节 会话数据结构

继续阅读本节完整说明和来源证据。

概述

在构建 AI Agent 时,会话与上下文管理是核心能力之一。Strands Agents SDK 提供了完整的解决方案来处理:

  • 会话生命周期管理:创建、持久化、恢复和清理会话状态
  • 对话历史管理:存储和检索多轮对话中的消息
  • 上下文窗口优化:通过滑动窗口和摘要等技术管理上下文长度
  • 主动上下文压缩:v1.40.0 引入的新特性,可预测性地压缩对话内容

资料来源:event_loop.py:1-50

架构概览

graph TD
    subgraph "会话管理层"
        SM[SessionManager]
        FSM[FileSessionManager]
    end
    
    subgraph "对话管理层"
        CM[ConversationManager]
        SWCM[SlidingWindowConversationManager]
        SUCM[SummarizingConversationManager]
    end
    
    subgraph "事件循环层"
        EL[EventLoop]
    end
    
    subgraph "Agent"
        A[Agent]
    end
    
    SM --> FSM
    CM --> SWCM
    CM --> SUCM
    EL --> CM
    A --> EL
    A --> SM

会话管理层负责会话的持久化和跨会话状态,而对话管理层则专注于单个对话内的消息管理和上下文优化。

资料来源:session_manager.py:1-80

会话管理(Session Management)

SessionManager 基类

SessionManager 是会话管理的核心抽象类,提供了会话生命周期管理的标准接口:

方法说明
create_session()创建新会话
get_session(session_id)根据 ID 获取会话
update_session(session_id, state)更新会话状态
delete_session(session_id)删除会话
list_sessions()列出所有会话

资料来源:session_manager.py:30-75

FileSessionManager 实现

FileSessionManager 是基于文件系统持久化的会话管理器实现:

from strands.session.file_session_manager import FileSessionManager

# 初始化文件会话管理器
session_manager = FileSessionManager(
    storage_dir="./sessions"  # 会话存储目录
)

# 创建新会话
session = session_manager.create_session()
session_id = session.session_id

# 获取会话
current_session = session_manager.get_session(session_id)

# 更新会话状态
session_manager.update_session(session_id, {"key": "value"})

# 删除会话
session_manager.delete_session(session_id)

特点:

  • 使用文件系统存储会话数据(JSON 格式)
  • 适合单机部署和开发环境
  • 支持会话状态的序列化和反序列化

资料来源:file_session_manager.py:1-100

会话数据结构

会话存储的数据结构包含以下关键字段:

字段类型说明
session_idstr会话唯一标识符
created_atdatetime创建时间
updated_atdatetime最后更新时间
statedict自定义状态数据
messageslist[Message]对话历史消息

资料来源:session_manager.py:50-70

对话管理(Conversation Management)

ConversationManager 基类

ConversationManager 是对话管理的核心抽象,定义了消息处理和上下文管理的标准接口:

classDiagram
    class ConversationManager {
        +messages: list[Message]
        +append_message(message: Message)*
        +get_messages() list[Message]*
        +trim_messages()*
        +clear()*
    }
    
    class SlidingWindowConversationManager {
        +window_size: int
        +trim_point: int
        +append_message(message: Message)
        +get_messages() list[Message]
        +trim_messages()
    }
    
    class SummarizingConversationManager {
        +model: Model
        +summary_threshold: int
        +append_message(message: Message)
        +get_messages() list[Message]
        +generate_summary() str
    }
    
    ConversationManager <|-- SlidingWindowConversationManager
    ConversationManager <|-- SummarizingConversationManager

资料来源:conversation_manager.py:1-60

滑动窗口对话管理器

SlidingWindowConversationManager 使用滑动窗口策略管理对话历史:

from strands.agent.conversation_manager import SlidingWindowConversationManager

# 初始化滑动窗口对话管理器
conversation_manager = SlidingWindowConversationManager(
    window_size=10,      # 保留最近 10 条消息
    trim_point=5,        # 当超窗口时,保留最近 5 条工具相关消息作为锚点
)

# 添加消息
conversation_manager.append_message({"role": "user", "content": "你好"})

# 获取当前对话历史
messages = conversation_manager.get_messages()

配置参数:

参数类型默认值说明
window_sizeint10滑动窗口大小(保留消息数)
trim_pointint3工具密集对话时的备用修剪点

资料来源:sliding_window_conversation_manager.py:1-80

摘要对话管理器

SummarizingConversationManager 使用 AI 模型生成对话摘要来压缩上下文:

from strands.agent.conversation_manager import SummarizingConversationManager
from strands.models import BedrockModel

# 使用摘要策略的对话管理器
conversation_manager = SummarizingConversationManager(
    model=BedrockModel(model_id="anthropic.claude-3-sonnet"),
    summary_threshold=20,  # 超过 20 条消息时生成摘要
    max_summary_length=500  # 摘要最大长度
)

# 添加消息 - 当消息数超过阈值时会自动生成摘要
conversation_manager.append_message({"role": "user", "content": "帮我分析销售数据"})

工作流程:

sequenceDiagram
    participant U as 用户
    participant CM as ConversationManager
    participant M as Model
    participant S as Storage
    
    U->>CM: append_message()
    Note over CM: 检查消息数量
    alt 消息数 > threshold
        CM->>M: 生成摘要请求
        M-->>CM: 摘要文本
        CM->>S: 保存摘要,清理旧消息
    else 消息数 <= threshold
        CM->>S: 直接保存消息
    end

资料来源:summarizing_conversation_manager.py:1-120

主动上下文压缩(Proactive Context Compression)

v1.40.0 引入了主动上下文压缩功能,这是对对话管理的重要增强。

核心概念

主动上下文压缩与传统的事后处理不同,它在对话即将触及上下文窗口限制之前就开始准备压缩操作:

# 主动上下文压缩配置
agent = Agent(
    model=BedrockModel(model_id="anthropic.claude-3-sonnet"),
    conversation_manager=SummarizingConversationManager(
        proactive_compression=True,  # 启用主动压缩
        compression_lookahead=5  # 提前 5 条消息开始准备
    )
)

特性优势

特性说明
预测性压缩在上下文窗口即将满时主动触发压缩
减少延迟避免压缩过程中用户等待
智能触发基于消息数量和令牌数的双重判断
AccessDenied 缓存对 count tokens 的 AccessDenied 错误进行缓存

资料来源:event_loop.py:150-200

错误缓存机制

v1.40.0 还引入了对 count tokens AccessDenied 错误的缓存机制:

# 当 IAM 权限不足时,系统会缓存失败状态
# 避免重复调用 count tokens API
conversation_manager._access_denied_cache = {}

这在 AWS 环境中尤其有用,可以减少不必要的 API 调用。

资料来源:event_loop.py:200-250

上下文窗口限制

上下文窗口查找表

v1.39.0 引入了上下文窗口限制的查找表功能:

from strands.models import BedrockModel

model = BedrockModel(model_id="anthropic.claude-3-sonnet")

# 获取模型的上下文窗口限制
context_limit = model.context_window_limit

# 获取当前消息的 token 数量
token_count = model.count_token(messages)

主要模型上下文窗口参考:

模型上下文窗口最大输出
Claude 3 Sonnet200K8K
Claude 3 Haiku200K4K
Claude 3.5 Sonnet200K8K
Claude 3.5 Haiku200K4K

资料来源:event_loop.py:100-150

超出上下文窗口异常

当输入内容超出模型上下文窗口限制时,系统会抛出 ContextWindowOverflowException

try:
    response = agent("非常长的输入内容...")
except ContextWindowOverflowException as e:
    # 处理上下文溢出
    print(f"上下文窗口溢出: {e.message}")
    # 可以考虑使用摘要管理器压缩历史

资料来源:event_loop.py:250-300

与 Agent 的集成

在 Agent 中配置对话管理器

Agent 在初始化时可以选择使用不同的对话管理器:

from strands import Agent
from strands.agent.conversation_manager import SlidingWindowConversationManager

# 方式 1: 使用滑动窗口
agent = Agent(
    model=BedrockModel(model_id="anthropic.claude-3-sonnet"),
    conversation_manager=SlidingWindowConversationManager(
        window_size=20
    ),
    tools=[calculator, file_reader]
)

# 方式 2: 使用摘要策略
agent = Agent(
    model=BedrockModel(model_id="anthropic.claude-3-sonnet"),
    conversation_manager=SummarizingConversationManager(
        model=BedrockModel(model_id="anthropic.claude-3-haiku"),
        summary_threshold=15
    ),
    tools=[calculator, file_reader]
)

# 方式 3: 使用默认管理器(系统自动选择)
agent = Agent(
    model=BedrockModel(model_id="anthropic.claude-3-sonnet"),
    tools=[calculator, file_reader]
)

Agent 运行时的上下文管理

graph LR
    A[Agent.run] --> B{检查消息数量}
    B -->|超过阈值| C[触发压缩]
    B -->|正常范围| D[继续处理]
    C --> E{压缩策略}
    E -->|滑动窗口| F[修剪旧消息]
    E -->|摘要| G[生成摘要替换历史]
    F --> H[继续处理]
    G --> H
    D --> I[调用模型]
    H --> I
    I --> J[返回结果]

资料来源:event_loop.py:300-400

最佳实践

1. 选择合适的对话管理器

场景推荐管理器原因
短对话默认管理器简单高效
长对话(需要摘要)SummarizingConversationManager保留关键信息
实时性要求高SlidingWindowConversationManager无 AI 调用开销
资源受限环境SlidingWindowConversationManager计算成本低

2. 合理配置窗口大小

# 根据模型上下文窗口和平均消息长度配置
SlidingWindowConversationManager(
    window_size=50,  # 假设每条消息平均约 100 tokens
    trim_point=10,   # 工具密集对话保留更多锚点
)

3. 会话持久化策略

# 生产环境建议使用数据库后端
class DatabaseSessionManager(SessionManager):
    def __init__(self, connection_pool):
        self.pool = connection_pool
    
    def create_session(self):
        # 使用数据库事务创建会话
        pass

常见问题与故障排除

问题 1: 对话历史丢失

症状:重启后对话历史丢失

可能原因

  • 未使用持久化的 SessionManager
  • 使用了内存会话管理器

解决方案:使用 FileSessionManager 或实现自定义的持久化管理器

# 启用会话持久化
agent = Agent(
    session_manager=FileSessionManager(storage_dir="./sessions"),
    # ...
)

问题 2: 上下文窗口频繁溢出

症状:对话进行到一定程度后抛出 ContextWindowOverflowException

可能原因

  • 滑动窗口设置过小
  • 摘要生成策略不当

解决方案

  1. 增大滑动窗口大小
  2. 切换到使用摘要策略的对话管理器
  3. 启用主动上下文压缩

问题 3: 摘要质量不佳

症状:生成的摘要丢失关键信息

可能原因

  • 摘要模型选择不当
  • summary_threshold 设置过低

解决方案

SummarizingConversationManager(
    model=BedrockModel(model_id="anthropic.claude-3-sonnet"),  # 使用更强的模型
    summary_threshold=10,  # 增加触发阈值
)

相关版本更新

版本更新内容
v1.41.0MCP 任务执行 TTL 支持增强
v1.40.0新增主动上下文压缩、AccessDenied 错误缓存
v1.39.0新增上下文窗口限制查找表
v1.38.0新增 count_token 方法
v1.37.0新增 context_window_limit 模型配置
v1.36.0Hook 回调机制支持 callable

See Also

资料来源:event_loop.py:1-50

多 Agent 系统

本文档介绍 Strands Agents SDK 中的多 Agent 系统,包括其架构设计、核心组件、编排器类型以及使用方式。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 核心特性

继续阅读本节完整说明和来源证据。

章节 系统组件

继续阅读本节完整说明和来源证据。

章节 模块导出

继续阅读本节完整说明和来源证据。

概述

多 Agent 系统是 Strands Agents SDK 的高级功能,允许开发者构建由多个专业 Agent 协作组成的复杂 AI 应用。在 v1.41.0 版本中,SDK 引入了 MultiAgentPlugin,支持 SwarmGraph 两种编排器,用于管理多 Agent 之间的交互与协作。

资料来源:strands-py/src/strands/plugins/multiagent_plugin.py:1-50

核心特性

特性描述
Swarm 编排器基于动态代理模式的去中心化编排,适合开放性任务
Graph 编排器基于有向图结构的确定性编排,适合结构化工作流
A2A 协议支持原生支持 Agent-to-Agent 通信协议
插件集成通过 MultiAgentPlugin 与 Agent 系统无缝集成

架构设计

系统组件

多 Agent 系统由以下核心组件构成:

graph TB
    subgraph 多Agent系统核心
        MP[MultiAgentPlugin]
        SW[Swarm Orchestrator]
        GR[Graph Orchestrator]
    end
    
    subgraph A2A通信层
        A2AS[A2A Server]
        A2AE[A2A Executor]
        A2AC[A2A Converters]
    end
    
    subgraph Agent层
        A1[Agent 实例 1]
        A2[Agent 实例 2]
        A3[Agent 实例 N]
    end
    
    MP --> SW
    MP --> GR
    SW --> A2AS
    GR --> A2AS
    A2AS --> A2AE
    A2AE --> A2AC
    A2AC --> A1
    A2AC --> A2
    A2AC --> A3

资料来源:strands-py/src/strands/multiagent/__init__.py:1-30

模块导出

strands.multiagent 模块导出了以下核心类和函数:

from strands.multiagent import (
    GraphOrchestrator,
    SwarmOrchestrator,
    AgentSkills,
    Skill,
)

资料来源:strands-py/src/strands/multiagent/__init__.py:1-30

Swarm 编排器

Swarm 编排器采用动态代理模式,实现了一种去中心化的多 Agent 协作方式。Agent 之间可以相互交接(handoff),实现任务的动态传递。

工作原理

graph LR
    A[用户请求] --> SW[Swarm Orchestrator]
    SW --> AG1[Agent Alpha]
    AG1 -->|交接| AG2[Agent Beta]
    AG2 -->|交接| AG3[Agent Gamma]
    AG3 -->|最终响应| SW
    SW --> R[返回结果]

核心类定义

资料来源:strands-py/src/strands/multiagent/swarm.py:1-100

交接机制(Handoff)

Swarm 的核心机制是 handoff,允许一个 Agent 将控制权转移给另一个 Agent:

参数类型描述
targetstr目标 Agent 的标识符
descriptionstr交接原因的描述
contextdict传递给目标 Agent 的上下文

资料来源:strands-py/src/strands/multiagent/swarm.py:50-80

Graph 编排器

Graph 编排器使用有向无环图(DAG)结构定义 Agent 之间的依赖关系,适合需要明确执行顺序的复杂工作流。

工作原理

graph TD
    START[开始] --> INPUT[Input Agent]
    INPUT -->|并行| PROC1[Process Agent 1]
    INPUT -->|并行| PROC2[Process Agent 2]
    PROC1 --> MERGE[Merge Agent]
    PROC2 --> MERGE
    MERGE --> FINAL[Final Agent]
    FINAL --> END[结束]

节点类型

节点类型描述
input工作流输入节点
agentAgent 执行节点
merge聚合节点,合并多个输入
conditional条件节点,根据条件选择分支

资料来源:strands-py/src/strands/multiagent/graph.py:1-100

A2A 协议支持

A2A(Agent-to-Agent)是一种标准化的 Agent 通信协议,允许不同实现、不同部署环境下的 Agent 之间进行互操作。

架构组件

组件描述源码位置
A2AServerA2A 服务器,处理入站请求server.py
A2AExecutorA2A 执行器,处理任务执行executor.py
Converters类型转换器,转换消息格式_converters.py

资料来源:strands-py/src/strands/multiagent/a2a/__init__.py:1-50

任务状态转换

stateDiagram-v2
    [*] --> working: 任务创建
    working --> completed: 执行成功
    working --> failed: 执行失败
    working --> canceled: 用户取消
    working --> input_required: 需要输入
    working --> auth_required: 需要认证
    completed --> [*]
    failed --> [*]
    canceled --> [*]
    input_required --> working: 用户提供输入
    auth_required --> working: 完成认证

A2A 响应转换

convert_response_to_agent_result 函数负责将 A2A 响应映射为 Agent 结果:

A2A 状态Agent 结果
completedend_turn
failedend_turn(带错误信息)
canceledend_turn(带取消信息)
rejectedend_turn(带拒绝信息)
input_requiredinterrupt
auth_requiredinterrupt

资料来源:strands-py/src/strands/multiagent/a2a/_converters.py:80-120

使用方式

通过 MultiAgentPlugin 集成

使用 MultiAgentPlugin 将多 Agent 系统集成到主 Agent 中:

from strands import Agent
from strands.plugins import MultiAgentPlugin
from strands.multiagent import SwarmOrchestrator, GraphOrchestrator

# 创建编排器
orchestrator = SwarmOrchestrator(
    agents=[agent_alpha, agent_beta, agent_gamma]
)

# 创建插件
plugin = MultiAgentPlugin(orchestrator=orchestrator)

# 集成到主 Agent
main_agent = Agent(
    model=model,
    plugins=[plugin],
    system_prompt="You are a coordinator that manages sub-agents."
)

# 调用
response = main_agent("Process the user request")

资料来源:strands-py/src/strands/plugins/multiagent_plugin.py:1-80

创建 A2A 服务器

from strands.multiagent.a2a import A2AServer

server = A2AServer(
    agent=my_agent,
    name="my-a2a-agent",
    description="An agent accessible via A2A protocol",
    version="1.0.0"
)

# 启动服务器
await server.start(host="0.0.0.0", port=8080)

资料来源:strands-py/src/strands/multiagent/a2a/server.py:1-100

创建 A2A 执行器

from strands.multiagent.a2a import A2AExecutor

executor = A2AExecutor(
    agent=my_agent,
    client_kwargs={"timeout": 60}
)

# 通过 A2A 协议调用其他 Agent
result = await executor.execute(
    target="other-agent@a2a://localhost:8080",
    task={"input": "process this data"}
)

资料来源:strands-py/src/strands/multiagent/a2a/executor.py:1-100

配置选项

MultiAgentPlugin 配置

参数类型必需默认值描述
orchestratorOrchestrator-编排器实例(Swarm 或 Graph)
namestr-插件名称
descriptionstr-插件描述

SwarmOrchestrator 配置

参数类型必需默认值描述
agentslist[Agent]-参与协作的 Agent 列表
handoff_instructionsstr-交接指令模板

GraphOrchestrator 配置

参数类型必需默认值描述
nodeslist[Node]-图节点列表
edgeslist[Edge]-图边列表
entry_pointstr-入口节点 ID

与 Skills 系统的集成

多 Agent 系统可以与 Skills 系统协同工作,实现更强大的功能:

from strands import Agent
from strands.plugins import MultiAgentPlugin
from strands.multiagent import SwarmOrchestrator
from strands.vended_plugins.skills import AgentSkills, Skill

# 加载技能
skills = Skill.from_directory("./skills/")

# 创建编排器
orchestrator = SwarmOrchestrator(
    agents=[
        Agent(
            model=model,
            plugins=[AgentSkills(skills=skills)],
            system_prompt="You are a research agent."
        ),
        Agent(
            model=model,
            plugins=[AgentSkills(skills=skills)],
            system_prompt="You are an analysis agent."
        )
    ]
)

plugin = MultiAgentPlugin(orchestrator=orchestrator)

资料来源:strands-py/src/strands/vended_plugins/skills/__init__.py:1-50

常见问题与故障排除

Agent 交接失败

问题: Swarm 编排器中 Agent 交接失败

可能原因:

  1. 目标 Agent 未在 agents 列表中注册
  2. 交接上下文过大,超过了模型上下文限制
  3. 目标 Agent 的 max_tokens 设置过小

解决方案:

# 确保所有 Agent 都已注册
orchestrator = SwarmOrchestrator(
    agents=[agent_alpha, agent_beta, agent_gamma]  # 完整列表
)

# 检查交接上下文大小
orchestrator = SwarmOrchestrator(
    agents=agents,
    handoff_instructions="只传递必要的上下文信息"
)

Graph 执行死循环

问题: Graph 编排器进入死循环

可能原因: 图中存在环形依赖

解决方案: 确保使用有向无环图(DAG),检查 edges 配置中是否存在回路。

A2A 连接超时

问题: A2A 执行器连接超时

解决方案:

executor = A2AExecutor(
    agent=my_agent,
    client_kwargs={
        "timeout": 120,  # 增加超时时间
        "retries": 3     # 增加重试次数
    }
)

版本历史

版本特性
v1.41.0新增 MultiAgentPlugin,支持 Swarm 和 Graph 编排器
早期版本基础 Agent-to-Agent 支持

参见

资料来源:strands-py/src/strands/plugins/multiagent_plugin.py:1-50

MCP 集成

MCP(Model Context Protocol,模型上下文协议)是一种开放标准协议,用于在 AI 代理与外部工具、数据源之间建立标准化通信。Strands Agents SDK 通过内置的 MCPClient 类提供原生 MCP 支持,使开发者能够无缝集成数千种预构建工具。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 整体架构

继续阅读本节完整说明和来源证据。

章节 工具调用流程

继续阅读本节完整说明和来源证据。

章节 MCPClient

继续阅读本节完整说明和来源证据。

概述

MCP(Model Context Protocol,模型上下文协议)是一种开放标准协议,用于在 AI 代理与外部工具、数据源之间建立标准化通信。Strands Agents SDK 通过内置的 MCPClient 类提供原生 MCP 支持,使开发者能够无缝集成数千种预构建工具。

Strands 的 MCP 集成具备以下核心能力:

  • 多传输协议支持:支持 stdio、HTTP/SSE 等多种传输方式
  • 任务增强执行:支持长时间运行工具的任务式执行模型
  • 工具过滤:支持按名称正则或精确匹配过滤可用工具
  • 流式响应:支持 MCP 服务器的流式事件处理
  • 进度报告:支持 MCP 任务的进度更新机制(实验性功能)

资料来源:strands-py/docs/MCP_CLIENT_ARCHITECTURE.md

架构设计

整体架构

MCP 客户端采用双线程架构,主线程负责同步 API,后台线程处理异步 MCP 会话通信。

graph TD
    subgraph "主线程 (同步 API)"
        A[Agent] --> B[MCPClient]
        B --> C[MCPAgentTool 列表]
    end
    
    subgraph "后台线程 (异步会话)"
        D[MCP 背景会话] --> E[ClientSession]
        E --> F[stdio_client / http_client]
        F --> G[MCP Server]
    end
    
    B -.->|通过 Future 通信| D

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:1-50

工具调用流程

sequenceDiagram
    participant Agent
    participant MCPClient
    participant Session as 后台会话
    participant MCPServer as MCP Server
    
    Agent->>MCPClient: 调用工具
    MCPClient->>MCPClient: 检查任务支持
    alt 任务增强执行
        MCPClient->>Session: call_tool_as_task()
        Session->>MCPServer: 创建任务
        MCPServer-->>Session: task_id
        loop 轮询直到完成
            Session->>MCPServer: poll_task()
            MCPServer-->>Session: status update
        end
        Session->>MCPServer: get_task_result()
        MCPServer-->>Session: result
    else 直接调用
        MCPClient->>Session: call_tool()
        Session->>MCPServer: 执行工具
        MCPServer-->>Session: result
    end
    Session-->>MCPClient: MCPCallToolResult
    MCPClient-->>Agent: 返回结果

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:200-280

核心组件

MCPClient

MCPClient 是连接 Strands Agent 与 MCP 服务器的主要入口点,负责管理会话生命周期和工具调用。

#### 构造函数参数

参数类型必需默认值说明
transport_callableCallable[[], AsyncContextManager[ClientSession]]-返回异步会话管理器的可调用对象
startup_timeouttimedelta30秒等待 MCP 服务器启动的超时时间
tool_filters`list[str \re.Pattern]`None工具名称过滤器
prefixstr""工具名称前缀
elicitation_callbackCallableNone用户确认回调
tasks_configTasksConfigNone任务增强执行配置

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:150-220

#### 关键方法

方法返回类型说明
start()None启动 MCP 会话并加载工具
stop()None停止 MCP 会话
tools()list[MCPAgentTool]获取可用工具列表
get_server_instructions()`str \None`获取服务器指令

MCPAgentTool

MCPAgentTool 是对 MCP 工具的封装,使其适配 Strands 的工具接口。

class MCPAgentTool(BaseTool):
    """MCP 工具的 Strands 封装."""
    
    mcp_tool: Tool
    mcp_client: MCPClient

资料来源:strands-py/src/strands/tools/mcp/mcp_agent_tool.py

TasksConfig 配置

任务增强执行配置,用于长时间运行工具的任务式管理。

class TasksConfig(TypedDict, total=False):
    """MCP 任务执行配置.
    
    Attributes:
        ttl: 任务生存时间,默认 1 分钟。
        poll_timeout: 轮询任务完成的超时时间,默认 5 分钟。
    """
    ttl: timedelta
    poll_timeout: timedelta

资料来源:strands-py/src/strands/tools/mcp/mcp_tasks.py

#### 默认配置值

配置项默认值说明
ttl1 分钟任务在服务器端的有效期
poll_timeout5 分钟客户端等待任务完成的最大时间

使用模式

基本 stdio 连接

通过标准输入/输出连接本地 MCP 服务器:

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# 定义 MCP 客户端
mcp_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "./data"]
        )
    )
)

# 创建 Agent 并传入 MCP 客户端
agent = Agent(tools=[mcp_client])
response = agent("列出 ./data 目录下的所有文件")

资料来源:README.md

HTTP/SSE 传输

连接远程 MCP 服务器:

from strands.tools.mcp import MCPClient
from mcp import http_client

mcp_client = MCPClient(
    lambda: http_client("https://mcp-server.example.com/mcp")
)

工具过滤

使用正则表达式或精确字符串过滤可用工具:

from strands.tools.mcp import MCPClient
import re

# 方式 1:精确匹配
client1 = MCPClient(
    transport_callable=transport,
    tool_filters=["read_file", "write_file"]
)

# 方式 2:正则匹配(以 search 开头的所有工具)
client2 = MCPClient(
    transport_callable=transport,
    tool_filters=[re.compile(r"^search_.*")]
)

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:80-120

任务增强执行

为长时间运行的工具启用任务增强模式:

from datetime import timedelta
from strands.tools.mcp import MCPClient, TasksConfig

tasks_config = TasksConfig(
    ttl=timedelta(minutes=5),        # 任务有效期 5 分钟
    poll_timeout=timedelta(minutes=30) # 轮询超时 30 分钟
)

mcp_client = MCPClient(
    transport_callable=transport,
    tasks_config=tasks_config
)

任务增强执行的完整流程:

flowchart LR
    A[call_tool_as_task] --> B[创建任务<br/>返回 task_id]
    B --> C{轮询状态}
    C -->|pending| C
    C -->|completed| D[get_task_result]
    C -->|failed| E[返回错误]
    C -->|cancelled| F[返回取消]
    D --> G[返回工具结果]

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:280-350

工具结果类型

MCP 工具调用返回 MCPCallToolResult 类型:

class MCPCallToolResult(TypedDict):
    """MCP 工具调用结果."""
    
    content: list[MCPContent]  # 内容块列表
    isError: bool              # 是否为错误结果

支持的 MCPContent 类型:

类型说明
MCPTextContent文本内容 {type: "text", text: str}
MCPImageContent图像内容 {type: "image", data: str, mimeType: str}
MCPResourceContent资源引用内容

资料来源:strands-py/src/strands/tools/mcp/mcp_types.py

错误处理

常见错误场景

错误场景错误类型处理方式
MCP 服务器启动超时MCPClientInitializationError增加 startup_timeout
工具调用超时底层超时异常设置合理的 read_timeout_seconds
任务结果获取失败任务错误结果检查 MCPCallToolResult.isError
任务已完成但结果过期错误结果消息重新调用工具

任务错误处理

当任务执行失败时,_create_task_error_result 方法生成标准化的错误响应:

def _create_task_error_result(self, message: str) -> MCPCallToolResult:
    """创建统一的错误结果格式."""
    return MCPCallToolResult(
        isError=True,
        content=[MCPTextContent(type="text", text=message)],
    )

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:180-200

会话状态检查

在执行工具调用前,系统会验证会话状态:

def _is_session_active(self) -> bool:
    """检查会话是否处于活跃状态."""
    if self._background_thread is None or not self._background_thread.is_alive():
        return False
    if self._close_future is not None and self._close_future.done():
        return False
    return True

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:380-395

高级配置

自定义传输工厂

from mcp import ClientSession
from strands.tools.mcp import MCPClient

def create_custom_transport() -> AsyncContextManager[ClientSession]:
    """创建自定义传输的工厂函数."""
    # 自定义连接逻辑
    return custom_client

mcp_client = MCPClient(
    transport_callable=create_custom_transport,
    startup_timeout=timedelta(seconds=60)
)

工具名称前缀

为来自特定服务器的工具有选择性地添加前缀,避免命名冲突:

mcp_client = MCPClient(
    transport_callable=transport,
    prefix="mcp_server_"  # 所有工具名将以 mcp_server_ 开头
)

多 MCP 服务器集成

from strands import Agent
from strands.tools.mcp import MCPClient

# AWS 文档 MCP 服务器
aws_docs_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(command="npx", args=["-y", "@modelcontextprotocol/server-aws-kb-retrieval-server"])
    )
)

# 文件系统 MCP 服务器  
fs_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "./data"])
    )
)

# 组合多个 MCP 客户端的工具
agent = Agent(tools=[aws_docs_client, fs_client])

已知限制与注意事项

实验性功能

  1. MCP 任务进度更新:根据社区反馈(Issue #1812),当前版本的 MCP 任务实现尚未原生支持进度报告功能。FastMCP 已支持此特性,但 Strands SDK 仍在规划中。
  1. 任务 TTL 限制:任务结果在服务器端过期后,即使任务标记为 completed,客户端也可能无法获取结果。详见 MCP 规范 2025-11-25

线程安全

  • MCPClient 本身在单线程环境中使用是安全的
  • MetricsClient 单例存在线程安全问题(Issue #2342),在多线程环境中使用指标功能时需注意

兼容性

  • Python 版本要求:3.10+
  • 旧版 SDK 不支持 MCP Tasks 功能,使用任务配置时需确保 SDK 版本兼容

调试与日志

启用调试日志以排查 MCP 集成问题:

import logging

logging.getLogger("strands.tools.mcp").setLevel(logging.DEBUG)

# 关键日志点包括:
# - MCP 工具调用日志
# - 任务创建和状态轮询日志
# - 会话生命周期日志

资料来源:strands-py/src/strands/tools/mcp/mcp_client.py:50-100

相关文档

另请参阅

资料来源:strands-py/docs/MCP_CLIENT_ARCHITECTURE.md

双向流式交互

双向流式交互(Bidirectional Streaming,简称 bidi)是 Strands Agents SDK 中的一个实验性功能模块,允许 AI 智能体与用户之间进行实时的、双向的数据传输。与传统的请求-响应模式不同,双向流式交互支持持续的数据交换,使得对话更加自然流畅,特别适用于需要实时反馈的应用场景。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 模块结构

继续阅读本节完整说明和来源证据。

章节 核心组件

继续阅读本节完整说明和来源证据。

章节 模型配置表

继续阅读本节完整说明和来源证据。

概述

双向流式交互(Bidirectional Streaming,简称 bidi)是 Strands Agents SDK 中的一个实验性功能模块,允许 AI 智能体与用户之间进行实时的、双向的数据传输。与传统的请求-响应模式不同,双向流式交互支持持续的数据交换,使得对话更加自然流畅,特别适用于需要实时反馈的应用场景。

资料来源:strands-py/src/strands/experimental/bidi/__init__.py

架构设计

模块结构

双向流式交互模块位于 strands.experimental.bidi 命名空间下,采用分层架构设计:

graph TD
    subgraph "表现层 (IO Layer)"
        A[audio.py - 音频输入输出]
        B[text.py - 文本输入输出]
    end
    
    subgraph "模型层 (Models)"
        C[gemini_live.py - Gemini Live]
        D[nova_sonic.py - Nova Sonic]
        E[openai_realtime.py - OpenAI Realtime]
    end
    
    subgraph "智能体层 (Agent Layer)"
        F[agent.py - BidiAgent]
        G[loop.py - 事件循环]
    end
    
    A --> C
    A --> D
    A --> E
    B --> C
    B --> D
    B --> E
    C --> F
    D --> F
    E --> F
    F --> G

核心组件

组件文件路径功能描述
BidiAgentagent/agent.py双向流式交互的核心智能体实现
BidiAgentLoopagent/loop.py管理事件循环和处理流程
GeminiLiveModelmodels/gemini_live.pyGoogle Gemini Live API 适配器
NovaSonicModelmodels/nova_sonic.pyAmazon Nova Sonic 适配器
OpenAIRealtimeModelmodels/openai_realtime.pyOpenAI Realtime API 适配器
AudioIOio/audio.py音频输入输出处理
TextIOio/text.py文本输入输出处理

资料来源:strands-py/src/strands/experimental/bidi/agent/agent.py

支持的模型提供商

双向流式交互模块已实现对以下实时 API 的支持:

模型配置表

模型提供商实现类API 类型支持功能
GoogleGeminiLiveModelGemini Live API文本/音频流
AmazonNovaSonicModelNova Sonic API文本/音频流
OpenAIOpenAIRealtimeModelRealtime API文本/音频流

资料来源:strands-py/src/strands/experimental/bidi/models/gemini_live.py 资料来源:strands-py/src/strands/experimental/bidi/models/nova_sonic.py 资料来源:strands-py/src/strands/experimental/bidi/models/openai_realtime.py

输入输出模式

文本模式

文本模式是双向流式交互的基础模式,通过 TextIO 类实现简单的文本输入输出流。

from strands.experimental.bidi.io.text import TextIO

# 创建文本输入输出处理器
text_io = TextIO()

# 文本模式适合命令行界面和简单的文本交互

资料来源:strands-py/src/strands/experimental/bidi/io/text.py

音频模式

音频模式支持实时语音交互,通过 AudioIO 类处理音频输入输出流。

from strands.experimental.bidi.io.audio import AudioIO

# 创建音频输入输出处理器
audio_io = AudioIO(
    input_sample_rate=16000,
    output_sample_rate=24000,
    input_channels=1,
    output_channels=1
)

音频配置参数表:

参数类型默认值说明
input_sample_rateint16000输入音频采样率 (Hz)
output_sample_rateint24000输出音频采样率 (Hz)
input_channelsint1输入音频声道数
output_channelsint1输出音频声道数

资料来源:strands-py/src/strands/experimental/bidi/io/audio.py

BidiAgent 事件循环

工作流程

BidiAgentLoop 是双向流式交互的核心事件处理引擎,负责协调用户输入、模型调用和输出处理的整个流程。

sequenceDiagram
    participant User as 用户
    participant IO as 输入/输出层
    participant Loop as BidiAgentLoop
    participant Model as 模型适配器
    participant Agent as BidiAgent
    
    User->>IO: 发送输入(文本/音频)
    IO->>Loop: 转发输入事件
    Loop->>Model: 流式请求
    Model-->>Loop: 流式响应
    Loop->>IO: 格式化输出
    IO-->>User: 展示结果
    Loop->>Agent: 更新状态
    Agent->>Agent: 记录对话历史

资料来源:strands-py/src/strands/experimental/bidi/agent/loop.py

事件类型

事件循环支持多种类型的事件处理:

事件类型描述触发场景
文本输入用户发送文本消息CLI 交互、文本聊天
音频输入用户发送语音数据语音对话
模型响应模型返回的文本/音频流AI 回复生成中
状态更新连接状态变化会话开始/结束/错误

使用示例

基本用法

from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models.gemini_live import GeminiLiveModel
from strands.experimental.bidi.io.text import TextIO

# 创建模型实例
model = GeminiLiveModel(model_id="gemini-2.0-flash-exp")

# 创建 IO 处理器
io_handler = TextIO()

# 创建双向智能体
agent = BidiAgent(
    model=model,
    io=io_handler
)

# 启动交互
agent.run()

音频交互模式

from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models.openai_realtime import OpenAIRealtimeModel
from strands.experimental.bidi.io.audio import AudioIO

# 创建音频 IO 处理器
audio_io = AudioIO(
    input_sample_rate=16000,
    output_sample_rate=24000
)

# 创建 OpenAI Realtime 模型
model = OpenAIRealtimeModel(
    model_id="gpt-4o-realtime-preview"
)

# 创建并运行音频智能体
agent = BidiAgent(
    model=model,
    io=audio_io
)
agent.run()

资料来源:strands-py/src/strands/experimental/bidi/agent/agent.py

与标准 Agent 的对比

特性标准 AgentBidiAgent
交互模式请求-响应双向流式
响应延迟完整响应后返回实时流式返回
实时反馈不支持支持中间进度更新
输入类型文本为主文本/音频
MCP 工具支持完整支持有限支持(实验性)
上下文管理自动管理实时流式管理

已知限制与注意事项

实验性状态

双向流式交互模块标记为实验性功能,这意味着:

  1. API 可能变更:接口和行为可能在后续版本中发生变化
  2. 功能不完整:部分功能仍在开发中
  3. 稳定性风险:不建议在生产环境中使用

资料来源:strands-py/src/strands/experimental/bidi/__init__.py

社区相关讨论

根据社区反馈,用户对双向流式交互有以下期待:

  • MCP 任务进度更新 (Issue #1812):用户希望 MCP 工具能够支持长期运行任务的进度反馈机制,这与双向流式交互的实时反馈能力高度相关
  • AG-UI 协议支持 (Issue #140):社区提议通过 AG-UI 协议将 Strands 事件映射到前端应用,这与实时双向交互的需求一致

线程安全注意事项

根据 SDK 的其他模块经验(如 Issue #2342 报告的 MetricsClient 线程安全问题),在多线程环境中使用 BidiAgent 时需要注意线程安全。建议:

  • 每个 BidiAgent 实例在单一线程中使用
  • 避免并发调用同一实例的 run() 方法
  • 妥善管理 IO 处理器的生命周期

未来发展方向

基于社区讨论和技术路线图,双向流式交互可能朝以下方向发展:

  1. MCP 工具集成:将 MCP 工具调用能力扩展到双向流式场景
  2. 进度报告机制:实现 MCP 任务进度更新功能
  3. AG-UI 协议支持:提供标准化的前端连接协议
  4. 多模态支持:扩展对视频等其他模态的支持

相关文档

另请参阅

资料来源:strands-py/src/strands/experimental/bidi/__init__.py

遥测与可观测性

Strands Agents SDK 提供了一套完整的遥测(Telemetry)与可观测性(Observability)系统,用于监控、追踪和记录 Agent 运行时的各种指标和行为。这套系统帮助开发者理解 Agent 的执行流程、诊断问题并优化性能。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 整体架构

继续阅读本节完整说明和来源证据。

章节 追踪器架构

继续阅读本节完整说明和来源证据。

章节 功能说明

继续阅读本节完整说明和来源证据。

概述

Strands 的遥测系统包含三大核心组件:

组件用途主要功能
追踪器(Tracer)分布式追踪记录请求在 Agent 内部的完整执行路径
指标(Metrics)性能指标收集并报告各种运行时指标数据
日志配置日志记录配置和管理应用程序日志

这套系统设计为可扩展的架构,支持与多种后端服务集成,包括 OpenTelemetry 标准兼容的追踪系统。

架构设计

整体架构

graph TD
    A[Agent 应用] --> B[遥测模块]
    B --> C[追踪器 Tracer]
    B --> D[指标收集器 Metrics]
    B --> E[日志系统]
    
    C --> F[OpenTelemetry 兼容后端]
    D --> G[指标存储后端]
    E --> H[日志处理器]
    
    F --> I[追踪数据导出器]
    G --> J[指标数据导出器]
    H --> K[日志输出]
    
    style C fill:#e1f5fe
    style D fill:#e1f5fe
    style E fill:#e1f5fe

追踪器架构

graph LR
    A[用户请求] --> B[Agent]
    B --> C[创建 Span]
    C --> D[模型调用]
    D --> E[工具调用]
    E --> F[工具执行]
    F --> G[结果处理]
    G --> H[返回响应]
    
    D --> D1[模型调用 Span]
    E --> E1[工具调用 Span]
    F --> F1[工具执行 Span]
    
    style D1 fill:#fff3e0
    style E1 fill:#fff3e0
    style F1 fill:#fff3e0

追踪系统(Tracer)

功能说明

追踪系统负责记录 Agent 执行过程中的完整调用链。每个操作都被封装为一个 Span(跨度),包含以下信息:

  • 操作名称和类型
  • 开始和结束时间
  • 父 Span 引用(用于构建调用树)
  • 相关属性和事件

配置参数

参数类型默认值说明
service_namestr"strands-agent"服务名称标识
exporterExporterNone数据导出器
endpointstrNone导出端点地址
insecureboolTrue是否使用非安全连接

使用方式

追踪器支持通过环境变量或代码配置:

# 环境变量配置
import os
os.environ["OTEL_SERVICE_NAME"] = "my-agent"
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:4317"

# 代码中启用追踪
from strands.telemetry import tracer

# 创建带追踪的 Agent
tracer.configure_tracer()

Span 创建流程

sequenceDiagram
    participant Agent
    participant Tracer
    participant Span
    participant Exporter
    
    Agent->>Tracer: 开始操作
    Tracer->>Span: 创建 Span
    Span->>Span: 设置父 Span
    Span->>Span: 添加属性
    Agent->>Span: 执行操作
    Agent->>Span: 记录事件
    Agent->>Tracer: 结束操作
    Tracer->>Span: 结束 Span
    Span->>Exporter: 导出追踪数据

指标系统(Metrics)

功能说明

指标系统负责收集和报告 Agent 运行时的各种量化指标,包括:

  • 计数器(Counter):统计事件发生次数
  • 直方图(Histogram):记录数值分布
  • 计时器(Timer):测量操作耗时
  • 仪表(Gauge):记录当前状态值

支持的指标类型

指标类型用途示例
counter递增计数API 调用次数
histogram数值分布响应时间分布
gauge当前值队列长度

已知问题

⚠️ 已知问题:MetricsClient 单例在 v1.41.0 版本中存在线程安全问题(Issue #2342)。在高并发场景下,多个线程同时访问指标收集器可能导致数据竞争或程序崩溃。此问题预计在后续版本中修复。

使用示例

from strands.telemetry.metrics import MetricsClient

# 获取单例实例
client = MetricsClient.get_instance()

# 记录指标
client.increment_counter("agent_calls_total", tags={"model": "claude-3"})
client.record_histogram("agent_duration_seconds", 1.234)

指标命名规范

Strands 使用以下命名约定:

# 格式:<domain>_<metric_name>_<unit>
# 示例:
agent_calls_total           # Agent 调用总次数
agent_duration_seconds      # Agent 执行耗时
tool_calls_total            # 工具调用总次数
model_tokens_total          # 模型 token 使用量

日志系统

日志配置

日志系统提供结构化日志输出,支持配置日志级别和输出格式:

from strands.telemetry.logging_config import configure_logging

# 配置日志
configure_logging(
    level="INFO",
    format="json",  # 或 "text"
    handlers=["console", "file"]
)

日志级别

级别用途记录内容
DEBUG调试信息详细的执行步骤和中间状态
INFO一般信息主要操作和状态变化
WARNING警告信息非致命问题和建议
ERROR错误信息失败操作和异常
CRITICAL严重错误系统不可用状态

Langfuse 集成

Strands 支持与 Langfuse 进行集成,用于追踪和可视化 Agent 行为:

⚠️ 环境变量隔离:v1.34.1 版本修复了 Langfuse 环境变量的隔离问题(PR #2022),确保 Langfuse 配置不会影响其他遥测组件。
import os
os.environ["LANGFUSE_PUBLIC_KEY"] = "your-public-key"
os.environ["LANGFUSE_SECRET_KEY"] = "your-secret-key"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"

环境变量配置

追踪配置

环境变量说明默认值
OTEL_SERVICE_NAME服务名称strands-agent
OTEL_EXPORTER_OTLP_ENDPOINTOTLP 端点-
OTEL_EXPORTER_OTLP_INSECURE非安全模式true
OTEL_TRACES_EXPORTER追踪导出器otlp

指标配置

环境变量说明
STRANDS_METRICS_ENABLED启用指标收集
STRANDS_METRICS_EXPORT_INTERVAL导出间隔(秒)

Langfuse 配置

环境变量说明
LANGFUSE_PUBLIC_KEY公共密钥
LANGFUSE_SECRET_KEY私密密钥
LANGFUSE_HOST服务地址

最佳实践

生产环境配置

from strands.telemetry import configure_telemetry

configure_telemetry(
    service_name="production-agent",
    tracing={
        "enabled": True,
        "endpoint": "https://otel.collector.company.com:4317",
        "sample_rate": 0.1  # 采样率 10%
    },
    metrics={
        "enabled": True,
        "export_interval": 60
    },
    logging={
        "level": "INFO",
        "format": "json"
    }
)

调试配置

configure_telemetry(
    service_name="debug-agent",
    tracing={
        "enabled": True,
        "sample_rate": 1.0  # 100% 采样
    },
    logging={
        "level": "DEBUG"
    }
)

性能考虑

  1. 采样策略:在生产环境中使用采样率避免性能开销
  2. 异步导出:遥测数据应异步导出,避免阻塞主流程
  3. 批量处理:多个指标批量发送以减少网络开销
  4. 资源限制:设置合理的缓冲区大小和导出间隔

常见问题

追踪数据未导出

问题:配置了追踪但未看到数据。

解决方案

  1. 确认端点地址正确且可访问
  2. 检查防火墙是否允许出站连接
  3. 验证 OpenTelemetry Collector 运行正常
  4. 确认环境变量配置正确

指标收集导致性能下降

问题:启用指标后响应时间增加。

解决方案

  1. 减少指标数量,只收集关键指标
  2. 增加导出间隔减少写入频率
  3. 使用异步导出器
  4. 考虑关闭高开销的直方图指标

线程安全问题

问题:多线程环境下出现异常。

解决方案

Error with Openai API: output new_sensitive (1027)

  1. 升级到最新版本(v1.41.1+)
  2. 避免在多线程环境中共享 MetricsClient 实例
  3. 使用线程

Please check that you have set the OPENAI_API_KEY environment variable with a valid API key.

来源:https://github.com/strands-agents/sdk-python / 项目说明书

失败模式与踩坑日记

保留 Doramagic 在发现、验证和编译中沉淀的项目专属风险,不把社区讨论只当作装饰信息。

medium 仓库名和安装名不一致

用户照着仓库名搜索包或照着包名找仓库时容易走错入口。

medium 能力判断依赖假设

假设不成立时,用户拿不到承诺的能力。

medium 维护活跃度未知

新项目、停更项目和活跃项目会被混在一起,推荐信任度下降。

medium 下游验证发现风险项

下游已经要求复核,不能在页面中弱化。

Pitfall Log / 踩坑日志

项目:strands-agents/sdk-python

摘要:发现 8 个潜在踩坑项,其中 0 个为 high/blocking;最高优先级:身份坑 - 仓库名和安装名不一致。

1. 身份坑 · 仓库名和安装名不一致

  • 严重度:medium
  • 证据强度:runtime_trace
  • 发现:仓库名 sdk-python 与安装入口 strands-agents 不完全一致。
  • 对用户的影响:用户照着仓库名搜索包或照着包名找仓库时容易走错入口。
  • 建议检查:在 npm/PyPI/GitHub 上确认包名映射和官方 README 说明。
  • 复现命令:pip install strands-agents
  • 防护动作:页面必须同时展示 repo 名和真实安装入口,避免用户搜索错包。
  • 证据:identity.distribution | github_repo:983715534 | https://github.com/strands-agents/sdk-python | repo=sdk-python; install=strands-agents

2. 能力坑 · 能力判断依赖假设

  • 严重度:medium
  • 证据强度:source_linked
  • 发现:README/documentation is current enough for a first validation pass.
  • 对用户的影响:假设不成立时,用户拿不到承诺的能力。
  • 建议检查:将假设转成下游验证清单。
  • 防护动作:假设必须转成验证项;没有验证结果前不能写成事实。
  • 证据:capability.assumptions | github_repo:983715534 | https://github.com/strands-agents/sdk-python | README/documentation is current enough for a first validation pass.

3. 维护坑 · 维护活跃度未知

  • 严重度:medium
  • 证据强度:source_linked
  • 发现:未记录 last_activity_observed。
  • 对用户的影响:新项目、停更项目和活跃项目会被混在一起,推荐信任度下降。
  • 建议检查:补 GitHub 最近 commit、release、issue/PR 响应信号。
  • 防护动作:维护活跃度未知时,推荐强度不能标为高信任。
  • 证据:evidence.maintainer_signals | github_repo:983715534 | https://github.com/strands-agents/sdk-python | last_activity_observed missing

4. 安全/权限坑 · 下游验证发现风险项

  • 严重度:medium
  • 证据强度:source_linked
  • 发现:no_demo
  • 对用户的影响:下游已经要求复核,不能在页面中弱化。
  • 建议检查:进入安全/权限治理复核队列。
  • 防护动作:下游风险存在时必须保持 review/recommendation 降级。
  • 证据:downstream_validation.risk_items | github_repo:983715534 | https://github.com/strands-agents/sdk-python | no_demo; severity=medium

5. 安全/权限坑 · 存在安全注意事项

  • 严重度:medium
  • 证据强度:source_linked
  • 发现:No sandbox install has been executed yet; downstream must verify before user use.
  • 对用户的影响:用户安装前需要知道权限边界和敏感操作。
  • 建议检查:转成明确权限清单和安全审查提示。
  • 防护动作:安全注意事项必须面向用户前置展示。
  • 证据:risks.safety_notes | github_repo:983715534 | https://github.com/strands-agents/sdk-python | No sandbox install has been executed yet; downstream must verify before user use.

6. 安全/权限坑 · 存在评分风险

  • 严重度:medium
  • 证据强度:source_linked
  • 发现:no_demo
  • 对用户的影响:风险会影响是否适合普通用户安装。
  • 建议检查:把风险写入边界卡,并确认是否需要人工复核。
  • 防护动作:评分风险必须进入边界卡,不能只作为内部分数。
  • 证据:risks.scoring_risks | github_repo:983715534 | https://github.com/strands-agents/sdk-python | no_demo; severity=medium

7. 维护坑 · issue/PR 响应质量未知

  • 严重度:low
  • 证据强度:source_linked
  • 发现:issue_or_pr_quality=unknown。
  • 对用户的影响:用户无法判断遇到问题后是否有人维护。
  • 建议检查:抽样最近 issue/PR,判断是否长期无人处理。
  • 防护动作:issue/PR 响应未知时,必须提示维护风险。
  • 证据:evidence.maintainer_signals | github_repo:983715534 | https://github.com/strands-agents/sdk-python | issue_or_pr_quality=unknown

8. 维护坑 · 发布节奏不明确

  • 严重度:low
  • 证据强度:source_linked
  • 发现:release_recency=unknown。
  • 对用户的影响:安装命令和文档可能落后于代码,用户踩坑概率升高。
  • 建议检查:确认最近 release/tag 和 README 安装命令是否一致。
  • 防护动作:发布节奏未知或过期时,安装说明必须标注可能漂移。
  • 证据:evidence.maintainer_signals | github_repo:983715534 | https://github.com/strands-agents/sdk-python | release_recency=unknown

来源:Doramagic 发现、验证与编译记录