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: 最终结果任务配置选项:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
ttl | timedelta | 1 分钟 | 任务生存时间 |
poll_timeout | timedelta | 5 分钟 | 轮询超时时间 |
资料来源:strands-py/src/strands/tools/mcp/mcp_tasks.py:18-32
模型提供商
支持的模型
| 提供商 | 模型类 | 特点 |
|---|---|---|
| Amazon Bedrock | BedrockModel | 支持 Nova、Claude 等模型,含服务层级支持 |
| Google Gemini | GeminiModel | 支持 Gemini 2.5 Flash 等 |
| Ollama | OllamaModel | 本地模型支持 |
| OpenAI | OpenAIModel | 标准 OpenAI API |
| LiteLLM | LiteLLMModel | 统一接口访问多种模型 |
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_count | Counter | 事件循环周期数 |
strands.event_loop.latency | Histogram | 事件循环延迟 |
strands.tool.call_count | Counter | 工具调用次数 |
strands.tool.duration | Histogram | 工具执行时长 |
资料来源: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.0 | MultiAgentPlugin 支持 Swarm 和 Graph 编排器;Bedrock 缓存 TTL 支持 |
| v1.40.0 | 主动上下文压缩;Token 计数 AccessDenied 错误缓存 |
| v1.39.0 | OpenAI 提供商支持 AWS Profile;MCPClientInitializationError 根因信息 |
| v1.38.0 | MCP CallToolResult.isError 标志保留;tiktoken 令牌计数估计 |
| v1.37.0 | 滑动窗口对话管理器回退点;实验性检查点支持 |
| v1.36.0 | Agent 构造器支持可调用 Hook 回调;Bedrock 服务层级支持 |
资料来源:README.md
已知问题与限制
社区报告的已知问题:
| Issue | 问题描述 | 状态 |
|---|---|---|
| #2351 | Python wheel 发布包缺少 README 和 LICENSE | 待修复 |
| #2347 | GeminiModel 在 safety_settings 阻止内容时因 usage_metadata=None 崩溃 | 待修复 |
| #2342 | MetricsClient 单例非线程安全 | 待修复 |
| #1812 | MCP 任务进度更新功能缺失 | 功能请求 |
相关链接
另请参阅
资料来源:README.md
快速开始
本页面介绍 Strands Agents Python SDK 的快速上手流程,帮助开发者从零开始创建和运行 AI Agent。通过本指南,您将了解如何安装 SDK、创建基础 Agent、配置模型提供商、以及集成工具和 MCP 服务器。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
环境要求
在开始之前,请确保您的开发环境满足以下要求:
| 要求 | 最低版本 | 说明 |
|---|---|---|
| Python | 3.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 凭证的常用方式:
- 环境变量:
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-west-2"
- AWS 配置文件:
aws configure
- 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?")
其他支持的提供商
| 提供商 | 模型类型 | 配置说明 |
|---|---|---|
| Anthropic | Claude 系列 | 需要 API 密钥 |
| OpenAI | GPT 系列 | 需要 API 密钥 |
| LiteLLM | 多提供商统一接口 | 支持 100+ 种模型 |
| LlamaAPI | Llama 系列 | 需要 API 密钥 |
| LlamaCpp | 本地量化模型 | 需要本地部署 |
| Writer | Writer 自有模型 | 需要 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_ENDPOINT | OTLP 端点 URL | http://localhost:4317 |
OTEL_EXPORTER_OTLP_HEADERS | OTLP 请求头 | X-api-key=xxx |
OTEL_SERVICE_NAME | 服务名称 | my-agent-service |
常见问题
AWS 凭证问题
问题:使用 Bedrock 模型时出现 AccessDeniedException
解决方案:
- 确认 AWS 凭证已正确配置
- 检查 IAM 用户/角色是否有 Bedrock 访问权限
- 确认目标区域(us-west-2)已启用所需模型
MCP 连接问题
问题:MCP 服务器连接失败
解决方案:
- 检查 MCP 服务器命令是否正确安装(如
uvx) - 确认服务器参数配置正确
- 查看服务器端日志排查问题
模型兼容性问题
问题:Gemini 模型出现 TypeError
解决方案:
- 确保使用最新版本的 SDK
- 检查
safety_settings配置是否正确
下一步
完成快速开始后,建议进一步阅读以下内容:
- Agent 循环机制:深入理解 Agent 的执行流程
- 工具开发指南:创建自定义工具
- MCP 集成:高级 MCP 配置
- 生产部署:生产环境最佳实践
相关链接
来源:https://github.com/strands-agents/sdk-python / 项目说明书
Agent 系统架构
Strands Agents SDK 的核心是 Agent 系统,这是一个模型驱动的 AI Agent 框架,采用简单而强大的设计理念。本文档详细介绍 Agent 的架构设计、核心组件、执行流程以及扩展机制。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
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 实现之间的一致性。
| 属性/方法 | 类型 | 说明 | |
|---|---|---|---|
model | Model | 配置的模型实例 | |
tools | list[AgentTool] | 注册的工具列表 | |
system_prefix | str \ | None | 系统提示前缀 |
callbacks | list[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 | 说明 |
|---|---|---|
completed | end_turn | 任务成功完成 |
failed | end_turn (带错误) | 任务执行失败 |
canceled | end_turn | 任务被取消 |
input_required | interrupt | 需要用户输入 |
auth_required | interrupt | 需要身份验证 |
资料来源: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]
end7.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 构造函数参数
| 参数 | 类型 | 默认值 | 说明 | |
|---|---|---|---|---|
model | Model | 必需 | 模型实例或模型配置 | |
tools | list[AgentTool] | [] | 可用工具列表 | |
system_prefix | str \ | None | None | 系统提示前缀 |
tool_executor | ToolExecutor | SequentialToolExecutor | 工具执行器 | |
callbacks | list[Callback] | [] | 回调函数列表 | |
plugins | list[Plugin] | [] | 插件列表 | |
max_tokens | int \ | None | None | 最大输出令牌数 |
资料来源:strands-py/src/strands/agent/agent.py:50-150
9. 常见故障模式
9.1 工具执行异常
| 错误类型 | 原因 | 解决方案 |
|---|---|---|
ToolProviderException | 工具提供者初始化失败 | 检查 MCP 服务器配置 |
MCPClientInitializationError | MCP 客户端连接失败 | 验证服务器地址和网络连接 |
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
- MCP 工具系统 - MCP 客户端和工具集成
- 工具执行器 - 并发和顺序执行策略
- Hooks 系统 - 事件回调扩展
- 多 Agent 协作 - A2A 协议和多 Agent 系统
- 模型配置 - 模型提供者和配置
模型提供商
模型提供商(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_id | str | 模型标识符 | 必需 |
| params | dict | 模型推理参数 | {} |
| streaming | bool | 启用流式响应 | False |
| client_args | dict | 底层客户端配置 | {} |
推理参数 (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 的兼容性。
相关文档
- Agent 核心概念 - 了解 Agent 如何使用模型提供商
- MCP 工具集成 - 通过 MCP 扩展模型工具能力
- 多 Agent 系统 - 多模型协作架构
- 配置与凭证管理 - 安全配置敏感信息
更新日志
| 版本 | 更新内容 |
|---|---|
| v1.41.0 | Starlette 依赖升级至 1.x |
| v1.40.0 | 主动上下文压缩、AccessDenied 错误缓存优化 |
| v1.39.0 | OpenAI Provider AWS Profile 支持 |
| v1.38.0 | Token 计数方法、多模态内容处理改进 |
| v1.37.0 | 上下文窗口限制配置、检查点功能 |
| v1.36.0 | Hook 回调可调用对象支持 |
| v1.35.0 | Bedrock 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 Agents SDK 中会话(Session)与上下文管理(Context Management)系统的架构、组件和使用方式。这些系统共同确保 AI Agent 能够有效地处理多轮对话、维护状态并智能地管理上下文窗口。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
在构建 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 Management)
SessionManager 基类
SessionManager 是会话管理的核心抽象类,提供了会话生命周期管理的标准接口:
| 方法 | 说明 |
|---|---|
create_session() | 创建新会话 |
get_session(session_id) | 根据 ID 获取会话 |
update_session(session_id, state) | 更新会话状态 |
delete_session(session_id) | 删除会话 |
list_sessions() | 列出所有会话 |
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_id | str | 会话唯一标识符 |
created_at | datetime | 创建时间 |
updated_at | datetime | 最后更新时间 |
state | dict | 自定义状态数据 |
messages | list[Message] | 对话历史消息 |
对话管理(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_size | int | 10 | 滑动窗口大小(保留消息数) |
trim_point | int | 3 | 工具密集对话时的备用修剪点 |
资料来源: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 错误进行缓存 |
错误缓存机制
v1.40.0 还引入了对 count tokens AccessDenied 错误的缓存机制:
# 当 IAM 权限不足时,系统会缓存失败状态
# 避免重复调用 count tokens API
conversation_manager._access_denied_cache = {}
这在 AWS 环境中尤其有用,可以减少不必要的 API 调用。
上下文窗口限制
上下文窗口查找表
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 Sonnet | 200K | 8K |
| Claude 3 Haiku | 200K | 4K |
| Claude 3.5 Sonnet | 200K | 8K |
| Claude 3.5 Haiku | 200K | 4K |
超出上下文窗口异常
当输入内容超出模型上下文窗口限制时,系统会抛出 ContextWindowOverflowException:
try:
response = agent("非常长的输入内容...")
except ContextWindowOverflowException as e:
# 处理上下文溢出
print(f"上下文窗口溢出: {e.message}")
# 可以考虑使用摘要管理器压缩历史
与 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[返回结果]最佳实践
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
可能原因:
- 滑动窗口设置过小
- 摘要生成策略不当
解决方案:
- 增大滑动窗口大小
- 切换到使用摘要策略的对话管理器
- 启用主动上下文压缩
问题 3: 摘要质量不佳
症状:生成的摘要丢失关键信息
可能原因:
- 摘要模型选择不当
- summary_threshold 设置过低
解决方案:
SummarizingConversationManager(
model=BedrockModel(model_id="anthropic.claude-3-sonnet"), # 使用更强的模型
summary_threshold=10, # 增加触发阈值
)
相关版本更新
| 版本 | 更新内容 |
|---|---|
| v1.41.0 | MCP 任务执行 TTL 支持增强 |
| v1.40.0 | 新增主动上下文压缩、AccessDenied 错误缓存 |
| v1.39.0 | 新增上下文窗口限制查找表 |
| v1.38.0 | 新增 count_token 方法 |
| v1.37.0 | 新增 context_window_limit 模型配置 |
| v1.36.0 | Hook 回调机制支持 callable |
See Also
- Agent 核心概念 - Agent 的基本结构和运行机制
- 事件循环机制 - 深入了解 Agent 的事件循环实现
- 工具系统 - 工具的定义、注册和使用
- MCP 集成 - MCP 服务器连接和工具调用
资料来源:event_loop.py:1-50
多 Agent 系统
本文档介绍 Strands Agents SDK 中的多 Agent 系统,包括其架构设计、核心组件、编排器类型以及使用方式。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
多 Agent 系统是 Strands Agents SDK 的高级功能,允许开发者构建由多个专业 Agent 协作组成的复杂 AI 应用。在 v1.41.0 版本中,SDK 引入了 MultiAgentPlugin,支持 Swarm 和 Graph 两种编排器,用于管理多 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:
| 参数 | 类型 | 描述 |
|---|---|---|
target | str | 目标 Agent 的标识符 |
description | str | 交接原因的描述 |
context | dict | 传递给目标 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 | 工作流输入节点 |
agent | Agent 执行节点 |
merge | 聚合节点,合并多个输入 |
conditional | 条件节点,根据条件选择分支 |
资料来源:strands-py/src/strands/multiagent/graph.py:1-100
A2A 协议支持
A2A(Agent-to-Agent)是一种标准化的 Agent 通信协议,允许不同实现、不同部署环境下的 Agent 之间进行互操作。
架构组件
| 组件 | 描述 | 源码位置 |
|---|---|---|
A2AServer | A2A 服务器,处理入站请求 | server.py |
A2AExecutor | A2A 执行器,处理任务执行 | 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 结果 |
|---|---|
completed | end_turn |
failed | end_turn(带错误信息) |
canceled | end_turn(带取消信息) |
rejected | end_turn(带拒绝信息) |
input_required | interrupt |
auth_required | interrupt |
资料来源: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 配置
| 参数 | 类型 | 必需 | 默认值 | 描述 |
|---|---|---|---|---|
orchestrator | Orchestrator | 是 | - | 编排器实例(Swarm 或 Graph) |
name | str | 否 | - | 插件名称 |
description | str | 否 | - | 插件描述 |
SwarmOrchestrator 配置
| 参数 | 类型 | 必需 | 默认值 | 描述 |
|---|---|---|---|---|
agents | list[Agent] | 是 | - | 参与协作的 Agent 列表 |
handoff_instructions | str | 否 | - | 交接指令模板 |
GraphOrchestrator 配置
| 参数 | 类型 | 必需 | 默认值 | 描述 |
|---|---|---|---|---|
nodes | list[Node] | 是 | - | 图节点列表 |
edges | list[Edge] | 是 | - | 图边列表 |
entry_point | str | 是 | - | 入口节点 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 交接失败
可能原因:
- 目标 Agent 未在
agents列表中注册 - 交接上下文过大,超过了模型上下文限制
- 目标 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 支持,使开发者能够无缝集成数千种预构建工具。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
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_callable | Callable[[], AsyncContextManager[ClientSession]] | 是 | - | 返回异步会话管理器的可调用对象 | |
startup_timeout | timedelta | 否 | 30秒 | 等待 MCP 服务器启动的超时时间 | |
tool_filters | `list[str \ | re.Pattern]` | 否 | None | 工具名称过滤器 |
prefix | str | 否 | "" | 工具名称前缀 | |
elicitation_callback | Callable | 否 | None | 用户确认回调 | |
tasks_config | TasksConfig | 否 | None | 任务增强执行配置 |
资料来源: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
#### 默认配置值
| 配置项 | 默认值 | 说明 |
|---|---|---|
ttl | 1 分钟 | 任务在服务器端的有效期 |
poll_timeout | 5 分钟 | 客户端等待任务完成的最大时间 |
使用模式
基本 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])
已知限制与注意事项
实验性功能
- MCP 任务进度更新:根据社区反馈(Issue #1812),当前版本的 MCP 任务实现尚未原生支持进度报告功能。FastMCP 已支持此特性,但 Strands SDK 仍在规划中。
- 任务 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
相关文档
另请参阅
- 工具执行器 - 了解并发和顺序工具执行策略
- 结构化输出工具 - 使用 MCP 工具返回结构化数据
- Agent Skills 集成 - 集成 AgentSkills.io 技能
- 遥测配置 - 配置 MCP 操作的追踪和监控
双向流式交互
双向流式交互(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核心组件
| 组件 | 文件路径 | 功能描述 |
|---|---|---|
BidiAgent | agent/agent.py | 双向流式交互的核心智能体实现 |
BidiAgentLoop | agent/loop.py | 管理事件循环和处理流程 |
GeminiLiveModel | models/gemini_live.py | Google Gemini Live API 适配器 |
NovaSonicModel | models/nova_sonic.py | Amazon Nova Sonic 适配器 |
OpenAIRealtimeModel | models/openai_realtime.py | OpenAI Realtime API 适配器 |
AudioIO | io/audio.py | 音频输入输出处理 |
TextIO | io/text.py | 文本输入输出处理 |
资料来源:strands-py/src/strands/experimental/bidi/agent/agent.py
支持的模型提供商
双向流式交互模块已实现对以下实时 API 的支持:
模型配置表
| 模型提供商 | 实现类 | API 类型 | 支持功能 |
|---|---|---|---|
GeminiLiveModel | Gemini Live API | 文本/音频流 | |
| Amazon | NovaSonicModel | Nova Sonic API | 文本/音频流 |
| OpenAI | OpenAIRealtimeModel | Realtime 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_rate | int | 16000 | 输入音频采样率 (Hz) |
output_sample_rate | int | 24000 | 输出音频采样率 (Hz) |
input_channels | int | 1 | 输入音频声道数 |
output_channels | int | 1 | 输出音频声道数 |
资料来源: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 的对比
| 特性 | 标准 Agent | BidiAgent |
|---|---|---|
| 交互模式 | 请求-响应 | 双向流式 |
| 响应延迟 | 完整响应后返回 | 实时流式返回 |
| 实时反馈 | 不支持 | 支持中间进度更新 |
| 输入类型 | 文本为主 | 文本/音频 |
| MCP 工具支持 | 完整支持 | 有限支持(实验性) |
| 上下文管理 | 自动管理 | 实时流式管理 |
已知限制与注意事项
实验性状态
双向流式交互模块标记为实验性功能,这意味着:
- API 可能变更:接口和行为可能在后续版本中发生变化
- 功能不完整:部分功能仍在开发中
- 稳定性风险:不建议在生产环境中使用
资料来源: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 处理器的生命周期
未来发展方向
基于社区讨论和技术路线图,双向流式交互可能朝以下方向发展:
- MCP 工具集成:将 MCP 工具调用能力扩展到双向流式场景
- 进度报告机制:实现 MCP 任务进度更新功能
- AG-UI 协议支持:提供标准化的前端连接协议
- 多模态支持:扩展对视频等其他模态的支持
相关文档
另请参阅
遥测与可观测性
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_name | str | "strands-agent" | 服务名称标识 |
exporter | Exporter | None | 数据导出器 |
endpoint | str | None | 导出端点地址 |
insecure | bool | True | 是否使用非安全连接 |
使用方式
追踪器支持通过环境变量或代码配置:
# 环境变量配置
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_ENDPOINT | OTLP 端点 | - |
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"
}
)
性能考虑
- 采样策略:在生产环境中使用采样率避免性能开销
- 异步导出:遥测数据应异步导出,避免阻塞主流程
- 批量处理:多个指标批量发送以减少网络开销
- 资源限制:设置合理的缓冲区大小和导出间隔
常见问题
追踪数据未导出
问题:配置了追踪但未看到数据。
解决方案:
- 确认端点地址正确且可访问
- 检查防火墙是否允许出站连接
- 验证 OpenTelemetry Collector 运行正常
- 确认环境变量配置正确
指标收集导致性能下降
问题:启用指标后响应时间增加。
解决方案:
- 减少指标数量,只收集关键指标
- 增加导出间隔减少写入频率
- 使用异步导出器
- 考虑关闭高开销的直方图指标
线程安全问题
问题:多线程环境下出现异常。
解决方案:
Error with Openai API: output new_sensitive (1027)
- 升级到最新版本(v1.41.1+)
- 避免在多线程环境中共享 MetricsClient 实例
- 使用线程
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 在发现、验证和编译中沉淀的项目专属风险,不把社区讨论只当作装饰信息。
用户照着仓库名搜索包或照着包名找仓库时容易走错入口。
假设不成立时,用户拿不到承诺的能力。
新项目、停更项目和活跃项目会被混在一起,推荐信任度下降。
下游已经要求复核,不能在页面中弱化。
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 发现、验证与编译记录