# https://github.com/strands-agents/sdk-python 项目说明书 生成时间：2026-05-28 08:56:25 UTC ## 目录 - [项目概述](#overview) - [快速开始](#quick-start) - [Agent 系统架构](#agent-system) - [模型提供商](#model-providers) - [工具系统](#tools-system) - [会话与上下文管理](#conversation-management) - [多 Agent 系统](#multi-agent) - [MCP 集成](#mcp-integration) - [双向流式交互](#bidirectional-streaming) - [遥测与可观测性](#telemetry) ## 项目概述 ### 相关页面 相关主题：[Agent 系统架构](#agent-system), [模型提供商](#model-providers), [快速开始](#quick-start)
相关源码文件 以下源码文件用于生成本页说明： - [README.md](https://github.com/strands-agents/sdk-python/blob/main/README.md) - [strands-py/src/strands/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/__init__.py) - [strands-py/src/strands/agent/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/__init__.py) - [strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) - [strands-py/src/strands/tools/mcp/mcp_tasks.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_tasks.py) - [strands-py/src/strands/tools/executors/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/__init__.py) - [strands-py/src/strands/event_loop/event_loop.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) - [strands-py/src/strands/telemetry/config.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/config.py) - [strands-py/src/strands/telemetry/metrics_constants.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/metrics_constants.py) - [site/src/config/navigation.yml](https://github.com/strands-agents/sdk-python/blob/main/site/src/config/navigation.yml)
# 项目概述 Strands Agents 是一个简单而强大的 SDK，采用模型驱动的方式来构建和运行 AI 智能体（Agent）。从简单的对话助手到复杂的自主工作流，从本地开发到生产部署，Strands Agents 都能满足您的需求。资料来源：[README.md](https://github.com/strands-agents/sdk-python/blob/main/README.md) ## 核心特性 Strands Agents SDK 提供以下核心功能： | 特性类别 | 说明 | |---------|------| | **轻量级与灵活** | 简洁的智能体循环，开箱即用，支持完全自定义 | | **模型无关** | 支持 Amazon Bedrock、Anthropic、Gemini、LiteLLM、Llama、Ollama、OpenAI、Writer 等多种模型提供商 | | **高级能力** | 多智能体系统、自主智能体、流式输出支持 | | **内置 MCP** | 原生支持 Model Context Protocol (MCP) 服务器，访问数千种预构建工具 | 资料来源：[README.md](https://github.com/strands-agents/sdk-python/blob/main/README.md) ## 系统架构 Strands Agents 的架构设计遵循清晰的模块化原则，主要包含以下几个核心组件： ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/__init__.py) ## 快速开始 ### 安装 ```bash pip install strands-agents strands-agents-tools ``` ### 基本使用 ```python from strands import Agent from strands_tools import calculator agent = Agent(tools=[calculator]) agent("What is the square root of 1764") ``` 资料来源：[README.md](https://github.com/strands-agents/sdk-python/blob/main/README.md) ## MCP 集成 Strands Agents 原生支持 Model Context Protocol (MCP)，可以无缝集成 MCP 服务器： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/README.md) ### MCP 任务机制 MCP 支持任务增强型工具执行，适用于长时间运行的工具。执行流程如下： ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_tasks.py#L18-L32) ## 模型提供商 ### 支持的模型 | 提供商 | 模型类 | 特点 | |-------|-------|------| | Amazon Bedrock | `BedrockModel` | 支持 Nova、Claude 等模型，含服务层级支持 | | Google Gemini | `GeminiModel` | 支持 Gemini 2.5 Flash 等 | | Ollama | `OllamaModel` | 本地模型支持 | | OpenAI | `OpenAIModel` | 标准 OpenAI API | | LiteLLM | `LiteLLMModel` | 统一接口访问多种模型 | ### Bedrock 示例 ```python 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 示例 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/README.md) ## 工具执行器 Strands Agents 提供灵活的工具体系，支持不同的执行策略： ### 并发执行器 ```python from strands.tools.executors import ConcurrentToolExecutor executor = ConcurrentToolExecutor(max_workers=4) agent = Agent(tools=[tool1, tool2], tool_executor=executor) ``` ### 顺序执行器 ```python from strands.tools.executors import SequentialToolExecutor executor = SequentialToolExecutor() agent = Agent(tools=[tool1, tool2], tool_executor=executor) ``` 资料来源：[strands-py/src/strands/tools/executors/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/__init__.py) ## 可观测性与遥测 Strands Agents 内置 OpenTelemetry 支持，提供追踪和指标收集功能： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/config.py#L30-L60) 资料来源：[strands-py/src/strands/telemetry/metrics_constants.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/metrics_constants.py) ## 事件循环机制 Agent 的核心执行流程通过事件循环实现： ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ## 开发者指南 ### 本地开发 ```bash 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](https://github.com/strands-agents/sdk-python/blob/main/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](https://github.com/strands-agents/sdk-python/blob/main/README.md) ## 已知问题与限制 社区报告的已知问题： | Issue | 问题描述 | 状态 | |-------|---------|------| | [#2351](https://github.com/strands-agents/sdk-python/issues/2351) | Python wheel 发布包缺少 README 和 LICENSE | 待修复 | | [#2347](https://github.com/strands-agents/sdk-python/issues/2347) | GeminiModel 在 safety_settings 阻止内容时因 usage_metadata=None 崩溃 | 待修复 | | [#2342](https://github.com/strands-agents/sdk-python/issues/2342) | MetricsClient 单例非线程安全 | 待修复 | | [#1812](https://github.com/strands-agents/sdk-python/issues/1812) | MCP 任务进度更新功能缺失 | 功能请求 | ## 相关链接 - [官方文档](https://strandsagents.com/) - [快速开始指南](https://strandsagents.com/docs/user-guide/quickstart/) - [Agent 循环概念](https://strandsagents.com/docs/user-guide/concepts/agents/agent-loop/) - [API 参考](https://strandsagents.com/docs/api/python/strands.agent.agent/) - [示例代码](https://strandsagents.com/docs/examples/) - [Discord 社区](https://discord.com/invite/strands) - [GitHub 仓库 (sdk-python)](https://github.com/strands-agents/sdk-python) - [GitHub 仓库 (sdk-typescript)](https://github.com/strands-agents/sdk-typescript) ## 另请参阅 - [Agent 核心实现](../agent/core.md) - [模型层架构](../models/overview.md) - [MCP 集成指南](../tools/mcp-integration.md) - [遥测与监控](../telemetry/overview.md) - [贡献指南](https://github.com/strands-agents/sdk-python/blob/main/CONTRIBUTING.md) --- ## 快速开始 ### 相关页面 相关主题：[项目概述](#overview), [Agent 系统架构](#agent-system)
相关源码文件 以下源码文件用于生成本页说明： - [README.md](https://github.com/strands-agents/sdk-python/blob/main/README.md) - [strands-py/src/strands/agent/agent.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/agent.py) - [strands-py/src/strands/models/bedrock.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/bedrock.py) - [strands-py/src/strands/models/gemini.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/gemini.py) - [strands-py/src/strands/models/ollama.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/ollama.py) - [strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) - [strands-py/src/strands/telemetry/config.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/config.py) - [strands-py/pyproject.toml](https://github.com/strands-agents/sdk-python/blob/main/strands-py/pyproject.toml)
# 快速开始 本页面介绍 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 的最简方式： ```bash pip install strands-agents strands-agents-tools ``` 此安装方式包含核心框架和常用内置工具，如计算器、网页搜索、文件操作等。 ### 虚拟环境 建议在虚拟环境中安装以避免依赖冲突： ```bash # 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: .\venv\Scripts\activate # 安装 SDK pip install strands-agents strands-agents-tools ``` ### 开发安装 如需参与 SDK 开发，可从源码安装： ```bash cd strands-py pip install -e . ``` 开发依赖可通过 `hatch` 工具管理，包括测试和代码格式化： ```bash pip install hatch hatch test # 运行单元测试 hatch fmt # 格式化和 lint ``` ## 创建第一个 Agent ### 基础示例 以下是创建最简单 Agent 的示例，使用内置的计算器工具： ```python 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. **环境变量**： ```bash export AWS_ACCESS_KEY_ID="your-access-key" export AWS_SECRET_ACCESS_KEY="your-secret-key" export AWS_DEFAULT_REGION="us-west-2" ``` 2. **AWS 配置文件**： ```bash aws configure ``` 3. **IAM 角色**（在 EC2、Lambda 或 ECS 环境中自动获取） ## 模型配置 Strands Agents 支持多种模型提供商，可通过 `model` 参数指定不同的模型。 ### Amazon Bedrock Bedrock 是 AWS 提供的托管式 AI 服务，Strands Agents 提供原生支持： ```python 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 密钥： ```python 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 允许在本地运行开源大语言模型： ```python 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` 包提供： ```python 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 命令 | ### 自定义工具 开发者可以创建自定义工具函数： ```python 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 服务器： ```python 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 支持任务增强的工具执行，适用于长时间运行的工具调用。任务执行流程如下： ```mermaid 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 任务参数： ```python 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 客户端支持按名称模式过滤可用工具： ```python from strands.tools.mcp import MCPClient import re # 只允许特定工具 mcp_client = MCPClient( transport_callable=..., tool_filters=[re.compile(r"^aws_.*")] # 只允许 aws_ 开头的工具 ) ``` ## 多模态交互 Strands Agents 支持多模态输入输出，包括文本、音频和图像： ```python 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 集成，支持追踪和指标收集： ```python 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` **解决方案**： 1. 确认 AWS 凭证已正确配置 2. 检查 IAM 用户/角色是否有 Bedrock 访问权限 3. 确认目标区域（us-west-2）已启用所需模型 ### MCP 连接问题 **问题**：MCP 服务器连接失败 **解决方案**： 1. 检查 MCP 服务器命令是否正确安装（如 `uvx`） 2. 确认服务器参数配置正确 3. 查看服务器端日志排查问题 ### 模型兼容性问题 **问题**：Gemini 模型出现 `TypeError` **解决方案**： 1. 确保使用最新版本的 SDK 2. 检查 `safety_settings` 配置是否正确 ## 下一步 完成快速开始后，建议进一步阅读以下内容： - [Agent 循环机制](https://strandsagents.com/docs/user-guide/concepts/agents/agent-loop/)：深入理解 Agent 的执行流程 - [工具开发指南](https://strandsagents.com/docs/user-guide/tools/)：创建自定义工具 - [MCP 集成](https://strandsagents.com/docs/user-guide/tools/mcp/)：高级 MCP 配置 - [生产部署](https://strandsagents.com/docs/user-guide/deploy/operating-agents-in-production/)：生产环境最佳实践 ## 相关链接 - [官方文档](https://strandsagents.com/) - [API 参考](https://strandsagents.com/docs/api/python/strands.agent.agent/) - [示例代码库](https://github.com/strands-agents/sdk-python/tree/main/examples) - [GitHub 仓库](https://github.com/strands-agents/sdk-python) - [Discord 社区](https://discord.com/invite/strands) --- ## Agent 系统架构 ### 相关页面 相关主题：[工具系统](#tools-system), [模型提供商](#model-providers), [会话与上下文管理](#conversation-management)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/agent/agent.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/agent.py) - [strands-py/src/strands/agent/base.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/base.py) - [strands-py/src/strands/agent/state.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/state.py) - [strands-py/src/strands/agent/agent_result.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/agent_result.py) - [strands-py/src/strands/hooks/events.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/hooks/events.py) - [strands-py/src/strands/event_loop/event_loop.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) - [strands-py/src/strands/tools/executors/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/__init__.py) - [strands-py/src/strands/tools/executors/concurrent.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/concurrent.py) - [strands-py/src/strands/multiagent/a2a/_converters.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/_converters.py)
# Agent 系统架构 Strands Agents SDK 的核心是 Agent 系统，这是一个模型驱动的 AI Agent 框架，采用简单而强大的设计理念。本文档详细介绍 Agent 的架构设计、核心组件、执行流程以及扩展机制。 ## 1. 架构概述 Strands Agents 采用模块化架构设计，将 AI Agent 的核心功能分解为多个协同工作的组件。系统以 Agent 为中心，通过事件循环驱动整个执行流程，并支持多种工具执行策略和扩展机制。 ```mermaid 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 的生命周期。它封装了模型配置、工具注册、会话管理和事件处理等核心功能。 ```python 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 在执行过程中的运行时状态，包括调用栈、工具执行结果和中间结果等。 ```mermaid 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 协调模型调用、工具执行和状态管理。 ```mermaid 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 调用工具执行子系统： ```mermaid graph LR A[tool_use 决策] --> B{执行策略} B -->|并发| C[ConcurrentToolExecutor] B -->|顺序| D[SequentialToolExecutor] C --> E[执行工具 1
执行工具 2
执行工具 N] D --> F[执行工具 1
等待完成
执行工具 2] E --> G[收集结果] F --> G G --> H[返回给模型] ``` 资料来源：[strands-py/src/strands/tools/executors/__init__.py:1-30]() ## 4. 工具执行器 ### 4.1 执行器类型 Strands Agents 提供了两种主要的工具执行策略： | 执行器 | 类名 | 适用场景 | |-------|------|---------| | 顺序执行 | SequentialToolExecutor | 工具间有依赖关系，必须按顺序执行 | | 并发执行 | ConcurrentToolExecutor | 工具相互独立，可并行执行以提高效率 | ```python 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 实现了异步并发工具执行，每个工具在独立的任务中运行： ```python 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 构造函数中传入回调函数： ```python 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 执行后的完整结果，包括消息内容、停止原因和工具使用记录： ```python @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 初始化时注册，提供额外的工具、工具执行器或其他扩展功能。 ```mermaid 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 系统： ```python 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. 执行流程图 ```mermaid 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-tools.md) - MCP 客户端和工具集成 - [工具执行器](./tool-executors.md) - 并发和顺序执行策略 - [Hooks 系统](./hooks.md) - 事件回调扩展 - [多 Agent 协作](./multiagent.md) - A2A 协议和多 Agent 系统 - [模型配置](./models.md) - 模型提供者和配置 --- ## 模型提供商 ### 相关页面 相关主题：[Agent 系统架构](#agent-system), [遥测与可观测性](#telemetry)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/models/model.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/model.py) - [strands-py/src/strands/models/bedrock.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/bedrock.py) - [strands-py/src/strands/models/anthropic.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/anthropic.py) - [strands-py/src/strands/models/gemini.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/gemini.py) - [strands-py/src/strands/models/openai.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/openai.py) - [strands-py/src/strands/models/ollama.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/ollama.py) - [strands-py/src/strands/models/litellm.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/litellm.py) - [strands-py/src/strands/models/llamaapi.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/llamaapi.py) - [strands-py/src/strands/models/llamacpp.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/llamacpp.py)
# 模型提供商 ## 概述 模型提供商（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](https://github.com/strands-agents/sdk-python/releases/tag/v1.35.0) ## 架构设计 ### 核心抽象 Strands SDK 的模型层采用分层架构设计。基础抽象层定义了所有模型提供商必须实现的接口规范，包括消息处理、流式响应生成、工具调用等功能。各提供商在此基础上封装具体的 API 调用逻辑。 ```mermaid 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 三个层级间选择，以平衡延迟与成本。 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/models/bedrock.py) ### AnthropicModel AnthropicModel 提供与 Anthropic Claude 模型的直接集成。该提供商封装了 Anthropic Messages API，支持工具调用、图像处理和流式响应。 ```python 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](https://github.com/strands-agents/sdk-python/releases/tag/v1.39.0) ```python 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](https://github.com/strands-agents/sdk-python/issues/2347) ```python 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](https://github.com/strands-agents/sdk-python/releases/tag/v1.40.0) ```python from strands.models import OllamaModel ollama_model = OllamaModel( model_id="llama3.2", params={"temperature": 0.8} ) ``` ### LiteLLMModel LiteLLMModel 通过 LiteLLM 统一网关访问多种模型提供商，简化了多提供商切换的复杂度。 ```python 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` 参数，用于控制服务级别： ```python bedrock_model = BedrockModel( model_id="us.amazon.nova-pro-v1:0", service_tier="priority" # priority | standard | flex ) ``` | 级别 | 延迟 | 成本 | 适用场景 | |------|------|------|----------| | priority | 最低 | 最高 | 实时交互 | | standard | 中等 | 中等 | 常规使用 | | flex | 较高 | 最低 | 批量处理 | ## 使用模式 ### 默认模型配置 SDK 默认使用 BedrockModel 作为模型提供商，但开发者可以轻松切换到其他提供商： ```python 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("你好，请介绍一下你自己") ``` ### 模型混用 在复杂应用中，可以根据任务类型选择不同的模型： ```python 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` 启用： ```python 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](https://github.com/strands-agents/sdk-python/releases/tag/v1.38.0) ```python # 查询 token 使用量 token_count = model.count_token("要计算 token 的文本") ``` ### 上下文压缩 v1.40.0 版本引入了主动上下文压缩功能，Conversation Manager 会在上下文接近限制时自动压缩对话历史。资料来源：[v1.40.0 Release](https://github.com/strands-agents/sdk-python/releases/tag/v1.40.0) ## 常见问题与解决方案 ### 认证配置 **问题**：模型调用时出现认证错误 **解决方案**：确保正确配置凭证 ```python # 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 和区域配置 ```python # 使用正确的模型 ID 和区域 bedrock_model = BedrockModel( model_id="us.amazon.nova-pro-v1:0", # 确认区域前缀 ) ``` ### 连接超时 **问题**：远程模型响应超时 **解决方案**：配置适当的超时时间 ```python 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](https://github.com/strands-agents/sdk-python/issues/2347) ## 自定义模型提供商 开发者可以通过继承基础 `Model` 类创建自定义模型提供商： ```python 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 核心概念](../concepts/agent-loop) - 了解 Agent 如何使用模型提供商 - [MCP 工具集成](../tools/mcp) - 通过 MCP 扩展模型工具能力 - [多 Agent 系统](../multiagent/overview) - 多模型协作架构 - [配置与凭证管理](../guides/configuration) - 安全配置敏感信息 ## 更新日志 | 版本 | 更新内容 | |------|----------| | 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 支持 | --- ## 工具系统 ### 相关页面 相关主题：[MCP 集成](#mcp-integration), [Agent 系统架构](#agent-system)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/tools/decorator.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/decorator.py) - [strands-py/src/strands/tools/registry.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/registry.py) - [strands-py/src/strands/tools/tools.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/tools.py) - [strands-py/src/strands/tools/_tool_helpers.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/_tool_helpers.py) - [strands-py/src/strands/tools/executors/_executor.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/_executor.py) - [strands-py/src/strands/tools/executors/concurrent.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/concurrent.py) - [strands-py/src/strands/tools/executors/sequential.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/sequential.py) - [strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) - [strands-py/src/strands/tools/mcp/mcp_tasks.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_tasks.py)
# 工具系统 ## 概述 Strands Agents 的工具系统是 SDK 的核心组件之一，它为 AI Agent 提供了调用外部工具和执行特定任务的能力。工具系统采用模块化设计，支持多种工具类型、执行策略和协议集成，使开发者能够灵活地扩展 Agent 的功能。 工具系统的设计遵循以下核心原则： - **可组合性**：工具可以独立使用，也可以组合成复杂的工具链 - **灵活性**：支持同步和异步执行模式，提供多种执行策略 - **可扩展性**：通过 MCP（Model Context Protocol）协议支持第三方工具服务器 - **可观测性**：内置遥测和指标收集，便于监控和调试 资料来源：[strands-py/src/strands/tools/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/__init__.py) ## 架构概览 ```mermaid graph TD subgraph "工具系统架构" A["Agent"] --> B["工具注册表
(ToolRegistry)"] B --> C["工具执行器
(ToolExecutor)"] C --> D["顺序执行器
(SequentialToolExecutor)"] C --> E["并发执行器
(ConcurrentToolExecutor)"] F["内置工具
(strands_tools)"] --> B G["MCP 工具
(MCPClient)"] --> B H["自定义工具
(@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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/registry.py) --- ## 工具定义与注册 ### 使用 `@tool` 装饰器定义工具 Strands 提供 `@tool` 装饰器用于定义自定义工具。装饰器会自动处理参数解析、文档生成和类型验证。 ```python from strands import tool @tool def calculator(expression: str) -> str: """执行数学计算并返回结果。 Args: expression: 要计算的数学表达式，例如 "2 + 3 * 4" Returns: 计算结果的字符串表示 """ try: result = eval(expression) return str(result) except Exception as e: return f"计算错误: {str(e)}" ``` 装饰器支持丰富的元数据配置： ```python from strands import tool from typing import Annotated @tool( name="search_documents", description="搜索文档库获取相关信息", doc_uri="https://example.com/docs/search-api", ) def search_docs( query: Annotated[str, "搜索查询字符串"], limit: Annotated[int, "返回结果数量限制"] = 10, filters: Annotated[dict | None, "可选的过滤条件"] = None, ) -> str: """在文档库中搜索匹配的文档。 Args: query: 搜索查询字符串 limit: 返回结果的最大数量，默认10 filters: 可选的过滤条件字典 Returns: 搜索结果列表的字符串表示 """ # 实现搜索逻辑 pass ``` 资料来源：[strands-py/src/strands/tools/decorator.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/decorator.py) ### 工具注册机制 工具通过 `ToolRegistry` 类进行集中管理。注册表维护了工具的映射关系，支持按名称查找工具。 ```mermaid sequenceDiagram participant User as 用户代码 participant Registry as ToolRegistry participant Tool as Tool 对象 User->>Registry: register(tool) Registry->>Tool: validate() Tool-->>Registry: 验证通过 Registry->>Registry: 存储到 _tools 字典 Registry-->>User: 注册成功 User->>Registry: get_tool(name) Registry-->>User: 返回 Tool 对象 User->>Registry: list_tools() Registry-->>User: 返回所有工具列表 ``` `ToolRegistry` 的核心方法： | 方法 | 说明 | |------|------| | `register(tool, name=None)` | 注册工具到注册表 | | `get_tool(name)` | 根据名称获取工具 | | `list_tools()` | 列出所有已注册的工具 | | `unregister(name)` | 取消注册工具 | | `clear()` | 清空注册表 | 资料来源：[strands-py/src/strands/tools/registry.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/registry.py) ### 工具元数据结构 每个工具在注册时都会生成标准化的元数据： ```python @dataclass class ToolMetadata: """工具元数据""" name: str # 工具唯一名称 description: str # 工具功能描述 parameters: dict # JSON Schema 格式的参数定义 doc_uri: str | None = None # 文档链接 tags: list[str] = field(default_factory=list) # 标签 ``` 参数定义遵循 JSON Schema 规范，支持： - 基础类型：`string`、`number`、`integer`、`boolean` - 复杂类型：`array`、`object` - 必填/可选标记 - 默认值 - 描述文档 资料来源：[strands-py/src/strands/tools/tools.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/tools.py) --- ## 工具执行器 Strands 提供了两种主要的工具执行策略：顺序执行和并发执行。执行器负责接收工具调用请求、执行工具逻辑并返回结果。 ```mermaid graph LR subgraph "执行策略" A["工具调用请求"] --> B{执行策略选择} B -->|单个工具| C["顺序执行器
(Sequential)"] B -->|多个工具| D{"依赖分析"} D -->|有依赖| E["按依赖顺序
顺序执行"] D -->|无依赖| F["并发执行器
(Concurrent)"] C --> G["结果收集"] E --> G F --> G G --> H["返回结果"] end ``` ### 执行器基类 所有执行器都继承自 `ToolExecutor` 基类： ```python class ToolExecutor(ABC): """工具执行器抽象基类""" @abstractmethod def execute( self, tool: Tool, tool_kwargs: dict[str, Any], session: "CurrentSession | None" = None, ) -> ToolResult: """执行单个工具""" pass @abstractmethod async def execute_async( self, tool: Tool, tool_kwargs: dict[str, Any], session: "CurrentSession | None" = None, ) -> ToolResult: """异步执行单个工具""" pass ``` 资料来源：[strands-py/src/strands/tools/executors/_executor.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/_executor.py) ### 顺序执行器 `SequentialToolExecutor` 按顺序逐个执行工具调用，确保每次只有一个工具在执行。 ```python from strands.tools.executors import SequentialToolExecutor executor = SequentialToolExecutor() # 执行单个工具 result = executor.execute( tool=my_tool, tool_kwargs={"param1": "value1"}, session=session, ) ``` **适用场景**： - 工具之间存在依赖关系 - 需要保持执行顺序 - 资源受限环境 资料来源：[strands-py/src/strands/tools/executors/sequential.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/sequential.py) ### 并发执行器 `ConcurrentToolExecutor` 可以并行执行多个工具调用，提高执行效率。 ```python from strands.tools.executors import ConcurrentToolExecutor executor = ConcurrentToolExecutor( max_workers=4, # 最大并发数 return_exceptions=True, # 单个工具异常不影响其他工具 ) # 并发执行多个工具 results = executor.execute_tools( tools=[tool1, tool2, tool3], tool_kwargs_list=[kwargs1, kwargs2, kwargs3], session=session, ) ``` **配置参数**： | 参数 | 类型 | 默认值 | 说明 | |------|------|--------|------| | `max_workers` | `int` | `4` | 最大并发工作线程数 | | `return_exceptions` | `bool` | `False` | 是否在结果中包含异常 | **适用场景**： - 工具之间无依赖关系 - 需要最大化吞吐量 - 工具执行时间较长 资料来源：[strands-py/src/strands/tools/executors/concurrent.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/executors/concurrent.py) ### 执行器选择 ```python from strands.tools.executors import ConcurrentToolExecutor, SequentialToolExecutor # 自动选择策略 executor = ConcurrentToolExecutor() # 默认并发执行 # Agent 中使用 agent = Agent( tools=[my_tool], tool_executor=executor, ) ``` --- ## MCP 工具集成 ### MCP 概述 MCP（Model Context Protocol）是一个开放协议，允许 AI 应用连接到外部数据源和工具。Strands 内置了完整的 MCP 客户端支持。 ```mermaid graph LR subgraph "Strands Agent" A["Agent"] B["MCPClient"] end subgraph "MCP Server" C["stdio_server"] D["SSE Server"] end B -->|stdio| C B -->|SSE| D C --> E["工具 1"] C --> F["工具 2"] C --> G["资源"] D --> H["工具 N"] ``` 资料来源：[strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ### 连接 MCP 服务器 Strands 支持两种 MCP 服务器连接方式： #### 标准输入/输出（stdio）模式 ```python 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="uvx", args=["awslabs.aws-documentation-mcp-server@latest"], ) ) ) # 使用 MCP 工具 with mcp_client: agent = Agent(tools=mcp_client.list_tools_sync()) response = agent("Tell me about Amazon Bedrock") ``` #### SSE（Server-Sent Events）模式 ```python from strands.tools.mcp import MCPClient from mcp import sse_client mcp_client = MCPClient( lambda: sse_client(url="http://localhost:8080/sse") ) with mcp_client: tools = mcp_client.list_tools_sync() agent = Agent(tools=tools) ``` ### MCP 客户端配置 ```python from strands.tools.mcp import MCPClient from strands.tools.mcp.mcp_tasks import TasksConfig from datetime import timedelta mcp_client = MCPClient( transport_callable=..., # 启动超时 startup_timeout=timedelta(seconds=30), # 工具过滤 tool_filters=["calculator", "search_*"], # 工具名称前缀 prefix="mcp_", # 任务增强配置（实验性） tasks_config=TasksConfig( ttl=timedelta(minutes=1), poll_timeout=timedelta(minutes=5), ), ) ``` | 配置项 | 类型 | 说明 | |--------|------|------| | `transport_callable` | `Callable` | 传输层连接工厂函数 | | `startup_timeout` | `timedelta` | MCP 服务器启动超时时间 | | `tool_filters` | `list[str\|Pattern]` | 工具过滤规则 | | `prefix` | `str` | 工具名称前缀 | | `tasks_config` | `TasksConfig` | 任务增强执行配置（实验性） | 资料来源：[strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ### MCP 任务增强执行（实验性） MCP 支持任务增强的工具执行，适用于长时间运行的操作。该特性允许： 1. 创建任务并获取任务 ID 2. 轮询任务状态直到完成 3. 获取任务结果 ```mermaid sequenceDiagram participant Client as MCPClient participant Server as MCP Server Client->>Server: call_tool_as_task(name, args, ttl) Server-->>Client: task_id Note over Client: 轮询循环开始 Client->>Server: poll_task(task_id) Server-->>Client: status: running Client->>Server: poll_task(task_id) Server-->>Client: status: completed Client->>Server: get_task_result(task_id) Server-->>Client: result ``` **配置**： ```python from strands.tools.mcp.mcp_tasks import TasksConfig, DEFAULT_TASK_TTL, DEFAULT_TASK_POLL_TIMEOUT tasks_config = TasksConfig( ttl=DEFAULT_TASK_TTL, # 默认 1 分钟 poll_timeout=DEFAULT_TASK_POLL_TIMEOUT, # 默认 5 分钟 ) ``` **注意**：此特性是实验性的，MCP 规范（2025-11-25）和 Strands 实现都可能发生变化。 资料来源：[strands-py/src/strands/tools/mcp/mcp_tasks.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_tasks.py) --- ## 工具调用流程 ### 同步调用流程 ```mermaid sequenceDiagram participant Agent participant Registry as ToolRegistry participant Executor as ToolExecutor participant Tool as Tool Agent->>Registry: get_tool(tool_name) Registry-->>Agent: Tool 对象 Agent->>Executor: execute(tool, kwargs) Executor->>Tool: invoke(kwargs) alt 参数验证通过 Tool-->>Executor: ToolResult(content) Executor-->>Agent: ToolResult else 参数验证失败 Tool-->>Executor: ValidationError Executor-->>Agent: ToolResult(isError=True) end ``` ### 异步调用流程 ```python # 异步工具调用 async def call_tool_async( tool_use_id: str, name: str, arguments: dict[str, Any] | None = None, read_timeout_seconds: timedelta | None = None, meta: dict[str, Any] | None = None, ) -> MCPToolResult: """异步调用 MCP 工具""" # 根据工具支持情况选择执行模式 coro = self._create_call_tool_coroutine( name, arguments, read_timeout_seconds, meta=meta ) future = self._invoke_on_background_thread(coro) call_tool_result = await asyncio.wrap_future(future) return self._handle_tool_result(tool_use_id, call_tool_result) ``` 资料来源：[strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ### 错误处理 工具执行中的错误会被统一捕获并转换为 `MCPToolResult`： ```python def _create_task_error_result(self, message: str) -> MCPCallToolResult: """创建格式统一的错误结果""" return MCPCallToolResult( isError=True, content=[MCPTextContent(type="text", text=message)], ) def _handle_tool_execution_error( self, tool_use_id: str, exception: Exception ) -> MCPToolResult: """处理工具执行异常""" logger.exception("tool execution failed") # 返回错误结果，包含错误详情 ``` 资料来源：[strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) --- ## 遥测与监控 ### 工具执行指标 Strands 会自动收集工具执行相关的遥测数据： | 指标名称 | 类型 | 说明 | |----------|------|------| | `strands.tool.call_count` | Counter | 工具调用总次数 | | `strands.tool.success_count` | Counter | 成功调用次数 | | `strands.tool.error_count` | Counter | 失败调用次数 | | `strands.tool.duration` | Histogram | 工具执行耗时 | ```python # 指标常量定义 STRANDS_TOOL_CALL_COUNT = "strands.tool.call_count" STRANDS_TOOL_SUCCESS_COUNT = "strands.tool.success_count" STRANDS_TOOL_ERROR_COUNT = "strands.tool.error_count" STRANDS_TOOL_DURATION = "strands.tool.duration" ``` 资料来源：[strands-py/src/strands/telemetry/metrics_constants.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/metrics_constants.py) ### 配置遥测导出 ```python from strands.telemetry import StrandsTelemetry # 配置控制台导出 telemetry = StrandsTelemetry() telemetry.setup_console_exporter() # 配置 OTLP 导出 telemetry.setup_otlp_exporter(endpoint="http://localhost:4317") # 配置 Meter Provider telemetry.setup_meter(enable_console_exporter=True, enable_otlp_exporter=True) ``` --- ## 完整使用示例 ### 基本工具使用 ```python from strands import Agent, tool from strands.tools.executors import ConcurrentToolExecutor # 定义工具 @tool def calculator(expression: str) -> str: """执行数学计算""" result = eval(expression) return str(result) @tool def get_weather(city: str) -> str: """获取城市天气""" # 调用天气 API return f"{city} 天气晴朗，25°C" # 创建 Agent agent = Agent( tools=[calculator, get_weather], tool_executor=ConcurrentToolExecutor(max_workers=2), ) # 使用 Agent response = agent("计算 (15 + 25) * 3，然后告诉我北京天气") ``` ### MCP 工具集成 ```python 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"], env={"ROOT": "/tmp"}, ) ), prefix="fs_", # 添加前缀避免命名冲突 tool_filters=["read_file", "list_directory"], # 过滤需要的工具 ) with mcp_client: # 获取 MCP 工具 mcp_tools = mcp_client.list_tools_sync() # 组合使用 MCP 工具和内置工具 agent = Agent(tools=mcp_tools + [calculator]) response = agent("读取 /tmp/example.txt 并计算其中数字的总和") ``` ### 结构化输出工具 ```python from strands import Agent from strands.types import ToolResult from pydantic import BaseModel class SearchResult(BaseModel): title: str url: str snippet: str class SearchResponse(BaseModel): results: list[SearchResult] total_count: int # 使用结构化输出 agent = Agent( model=my_model, response_format=SearchResponse, ) result = agent("搜索关于 Python 异步编程的文章") # result.content 会被解析为 SearchResponse 对象 ``` --- ## 最佳实践 ### 1. 工具命名规范 - 使用清晰的动词前缀（如 `get_`、`search_`、`calculate_`） - 避免与内置工具名称冲突 - 使用下划线分隔单词（snake_case） ### 2. 参数设计 - 为参数提供详细的 docstring 描述 - 使用 `typing.Annotated` 添加参数说明 - 提供合理的默认值 - 使用类型提示提高可读性 ```python from typing import Annotated @tool def search_documents( query: Annotated[str, "搜索关键词，最少3个字符"], limit: Annotated[int, "结果数量限制"] = 10, include_metadata: Annotated[bool, "是否包含元数据"] = False, ) -> str: """搜索文档""" pass ``` ### 3. 错误处理 ```python @tool def risky_operation(data: str) -> str: """执行可能失败的操作""" try: result = perform_operation(data) return result except ValueError as e: return f"参数错误: {str(e)}" except Exception as e: return f"操作失败: {str(e)}" ``` ### 4. 执行策略选择 | 场景 | 推荐执行器 | |------|------------| | 工具间无依赖 | `ConcurrentToolExecutor` | | 工具间有依赖 | `SequentialToolExecutor` | | 资源受限环境 | `SequentialToolExecutor` | | 高吞吐量需求 | `ConcurrentToolExecutor(max_workers=N)` | --- ## 常见问题 ### Q: 工具调用超时如何处理？ A: 通过 `read_timeout_seconds` 参数设置超时： ```python result = executor.execute( tool=my_tool, tool_kwargs={"data": "value"}, read_timeout_seconds=timedelta(seconds=30), ) ``` ### Q: 如何调试工具执行？ A: 启用调试日志并检查遥测数据： ```python import logging logging.basicConfig(level=logging.DEBUG) # 查看工具调用详情 telemetry = StrandsTelemetry() telemetry.setup_console_exporter() ``` ### Q: MCP 服务器连接失败怎么办？ A: 检查以下几点： 1. 服务器命令是否正确安装 2. 网络连接是否正常 3. 启动超时设置是否足够 4. 查看服务器日志 ```python mcp_client = MCPClient( lambda: stdio_client(StdioServerParameters(...)), startup_timeout=timedelta(seconds=60), # 增加超时时间 ) ``` --- ## 相关文档 - [Agent 核心概念](agent-core-concepts) - [事件循环机制](event-loop) - [MCP 协议支持](mcp-support) - [遥测配置](telemetry) - [工具示例集](../examples/tools) --- ## 更新日志 | 版本 | 更新内容 | |------|----------| | v1.41.0 | 新增 MultiAgentPlugin 支持 | | v1.38.0 | 修复 MCP 工具错误标志传递 | | v1.36.0 | 支持在 Agent 构造函数中传递可调用钩子 | | v1.35.0 | Bedrock 服务层级支持 | --- ## 会话与上下文管理 ### 相关页面 相关主题：[Agent 系统架构](#agent-system), [遥测与可观测性](#telemetry)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/agent/conversation_manager/conversation_manager.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/conversation_manager/conversation_manager.py) - [strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py) - [strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py) - [strands-py/src/strands/session/session_manager.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/session/session_manager.py) - [strands-py/src/strands/session/file_session_manager.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/session/file_session_manager.py) - [strands-py/src/strands/event_loop/event_loop.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py)
# 会话与上下文管理 本文档详细介绍 Strands Agents SDK 中会话（Session）与上下文管理（Context Management）系统的架构、组件和使用方式。这些系统共同确保 AI Agent 能够有效地处理多轮对话、维护状态并智能地管理上下文窗口。 ## 概述 在构建 AI Agent 时，会话与上下文管理是核心能力之一。Strands Agents SDK 提供了完整的解决方案来处理： - **会话生命周期管理**：创建、持久化、恢复和清理会话状态 - **对话历史管理**：存储和检索多轮对话中的消息 - **上下文窗口优化**：通过滑动窗口和摘要等技术管理上下文长度 - **主动上下文压缩**：v1.40.0 引入的新特性，可预测性地压缩对话内容 资料来源：[event_loop.py:1-50](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ## 架构概览 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/session/session_manager.py) ## 会话管理（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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/session/session_manager.py) ### FileSessionManager 实现 `FileSessionManager` 是基于文件系统持久化的会话管理器实现： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/session/file_session_manager.py) ### 会话数据结构 会话存储的数据结构包含以下关键字段： | 字段 | 类型 | 说明 | |------|------|------| | `session_id` | str | 会话唯一标识符 | | `created_at` | datetime | 创建时间 | | `updated_at` | datetime | 最后更新时间 | | `state` | dict | 自定义状态数据 | | `messages` | list[Message] | 对话历史消息 | 资料来源：[session_manager.py:50-70](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/session/session_manager.py) ## 对话管理（Conversation Management） ### ConversationManager 基类 `ConversationManager` 是对话管理的核心抽象，定义了消息处理和上下文管理的标准接口： ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/conversation_manager/conversation_manager.py) ### 滑动窗口对话管理器 `SlidingWindowConversationManager` 使用滑动窗口策略管理对话历史： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py) ### 摘要对话管理器 `SummarizingConversationManager` 使用 AI 模型生成对话摘要来压缩上下文： ```python 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": "帮我分析销售数据"}) ``` **工作流程：** ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py) ## 主动上下文压缩（Proactive Context Compression） v1.40.0 引入了**主动上下文压缩**功能，这是对对话管理的重要增强。 ### 核心概念 主动上下文压缩与传统的事后处理不同，它在对话即将触及上下文窗口限制之前就开始准备压缩操作： ```python # 主动上下文压缩配置 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ### 错误缓存机制 v1.40.0 还引入了对 count tokens AccessDenied 错误的缓存机制： ```python # 当 IAM 权限不足时，系统会缓存失败状态 # 避免重复调用 count tokens API conversation_manager._access_denied_cache = {} ``` 这在 AWS 环境中尤其有用，可以减少不必要的 API 调用。 资料来源：[event_loop.py:200-250](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ## 上下文窗口限制 ### 上下文窗口查找表 v1.39.0 引入了上下文窗口限制的查找表功能： ```python 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 | 资料来源：[event_loop.py:100-150](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ### 超出上下文窗口异常 当输入内容超出模型上下文窗口限制时，系统会抛出 `ContextWindowOverflowException`： ```python try: response = agent("非常长的输入内容...") except ContextWindowOverflowException as e: # 处理上下文溢出 print(f"上下文窗口溢出: {e.message}") # 可以考虑使用摘要管理器压缩历史 ``` 资料来源：[event_loop.py:250-300](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ## 与 Agent 的集成 ### 在 Agent 中配置对话管理器 Agent 在初始化时可以选择使用不同的对话管理器： ```python 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 运行时的上下文管理 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/event_loop/event_loop.py) ## 最佳实践 ### 1. 选择合适的对话管理器 | 场景 | 推荐管理器 | 原因 | |------|------------|------| | 短对话 | 默认管理器 | 简单高效 | | 长对话（需要摘要） | SummarizingConversationManager | 保留关键信息 | | 实时性要求高 | SlidingWindowConversationManager | 无 AI 调用开销 | | 资源受限环境 | SlidingWindowConversationManager | 计算成本低 | ### 2. 合理配置窗口大小 ```python # 根据模型上下文窗口和平均消息长度配置 SlidingWindowConversationManager( window_size=50, # 假设每条消息平均约 100 tokens trim_point=10, # 工具密集对话保留更多锚点 ) ``` ### 3. 会话持久化策略 ```python # 生产环境建议使用数据库后端 class DatabaseSessionManager(SessionManager): def __init__(self, connection_pool): self.pool = connection_pool def create_session(self): # 使用数据库事务创建会话 pass ``` ## 常见问题与故障排除 ### 问题 1: 对话历史丢失 **症状**：重启后对话历史丢失 **可能原因**： - 未使用持久化的 SessionManager - 使用了内存会话管理器 **解决方案**：使用 FileSessionManager 或实现自定义的持久化管理器 ```python # 启用会话持久化 agent = Agent( session_manager=FileSessionManager(storage_dir="./sessions"), # ... ) ``` ### 问题 2: 上下文窗口频繁溢出 **症状**：对话进行到一定程度后抛出 ContextWindowOverflowException **可能原因**： - 滑动窗口设置过小 - 摘要生成策略不当 **解决方案**： 1. 增大滑动窗口大小 2. 切换到使用摘要策略的对话管理器 3. 启用主动上下文压缩 ### 问题 3: 摘要质量不佳 **症状**：生成的摘要丢失关键信息 **可能原因**： - 摘要模型选择不当 - summary_threshold 设置过低 **解决方案**： ```python 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-core-concepts.md) - Agent 的基本结构和运行机制 - [事件循环机制](event-loop.md) - 深入了解 Agent 的事件循环实现 - [工具系统](../tools/tool-system.md) - 工具的定义、注册和使用 - [MCP 集成](../mcp/mcp-integration.md) - MCP 服务器连接和工具调用 --- ## 多 Agent 系统 ### 相关页面 相关主题：[Agent 系统架构](#agent-system)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/multiagent/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/__init__.py) - [strands-py/src/strands/multiagent/graph.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/graph.py) - [strands-py/src/strands/multiagent/swarm.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/swarm.py) - [strands-py/src/strands/multiagent/a2a/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/__init__.py) - [strands-py/src/strands/multiagent/a2a/server.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/server.py) - [strands-py/src/strands/multiagent/a2a/executor.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/executor.py) - [strands-py/src/strands/plugins/multiagent_plugin.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/plugins/multiagent_plugin.py)
# 多 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/plugins/multiagent_plugin.py) ### 核心特性 | 特性 | 描述 | |------|------| | **Swarm 编排器** | 基于动态代理模式的去中心化编排，适合开放性任务 | | **Graph 编排器** | 基于有向图结构的确定性编排，适合结构化工作流 | | **A2A 协议支持** | 原生支持 Agent-to-Agent 通信协议 | | **插件集成** | 通过 MultiAgentPlugin 与 Agent 系统无缝集成 | ## 架构设计 ### 系统组件 多 Agent 系统由以下核心组件构成： ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/__init__.py) ### 模块导出 `strands.multiagent` 模块导出了以下核心类和函数： ```python from strands.multiagent import ( GraphOrchestrator, SwarmOrchestrator, AgentSkills, Skill, ) ``` 资料来源：[strands-py/src/strands/multiagent/__init__.py:1-30](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/__init__.py) ## Swarm 编排器 Swarm 编排器采用动态代理模式，实现了一种去中心化的多 Agent 协作方式。Agent 之间可以相互交接（handoff），实现任务的动态传递。 ### 工作原理 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/swarm.py) ### 交接机制（Handoff） Swarm 的核心机制是 **handoff**，允许一个 Agent 将控制权转移给另一个 Agent： | 参数 | 类型 | 描述 | |------|------|------| | `target` | str | 目标 Agent 的标识符 | | `description` | str | 交接原因的描述 | | `context` | dict | 传递给目标 Agent 的上下文 | 资料来源：[strands-py/src/strands/multiagent/swarm.py:50-80](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/swarm.py) ## Graph 编排器 Graph 编排器使用有向无环图（DAG）结构定义 Agent 之间的依赖关系，适合需要明确执行顺序的复杂工作流。 ### 工作原理 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/graph.py) ## A2A 协议支持 A2A（Agent-to-Agent）是一种标准化的 Agent 通信协议，允许不同实现、不同部署环境下的 Agent 之间进行互操作。 ### 架构组件 | 组件 | 描述 | 源码位置 | |------|------|----------| | `A2AServer` | A2A 服务器，处理入站请求 | [server.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/server.py) | | `A2AExecutor` | A2A 执行器，处理任务执行 | [executor.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/executor.py) | | `Converters` | 类型转换器，转换消息格式 | [_converters.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/_converters.py) | 资料来源：[strands-py/src/strands/multiagent/a2a/__init__.py:1-50](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/__init__.py) ### 任务状态转换 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/_converters.py) ## 使用方式 ### 通过 MultiAgentPlugin 集成 使用 `MultiAgentPlugin` 将多 Agent 系统集成到主 Agent 中： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/plugins/multiagent_plugin.py) ### 创建 A2A 服务器 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/server.py) ### 创建 A2A 执行器 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/multiagent/a2a/executor.py) ## 配置选项 ### 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 系统协同工作，实现更强大的功能： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/vended_plugins/skills/__init__.py) ## 常见问题与故障排除 ### Agent 交接失败 **问题**: Swarm 编排器中 Agent 交接失败 **可能原因**: 1. 目标 Agent 未在 `agents` 列表中注册 2. 交接上下文过大，超过了模型上下文限制 3. 目标 Agent 的 `max_tokens` 设置过小 **解决方案**: ```python # 确保所有 Agent 都已注册 orchestrator = SwarmOrchestrator( agents=[agent_alpha, agent_beta, agent_gamma] # 完整列表 ) # 检查交接上下文大小 orchestrator = SwarmOrchestrator( agents=agents, handoff_instructions="只传递必要的上下文信息" ) ``` ### Graph 执行死循环 **问题**: Graph 编排器进入死循环 **可能原因**: 图中存在环形依赖 **解决方案**: 确保使用有向无环图（DAG），检查 `edges` 配置中是否存在回路。 ### A2A 连接超时 **问题**: A2A 执行器连接超时 **解决方案**: ```python executor = A2AExecutor( agent=my_agent, client_kwargs={ "timeout": 120, # 增加超时时间 "retries": 3 # 增加重试次数 } ) ``` ## 版本历史 | 版本 | 特性 | |------|------| | v1.41.0 | 新增 `MultiAgentPlugin`，支持 Swarm 和 Graph 编排器 | | 早期版本 | 基础 Agent-to-Agent 支持 | ## 参见 - [Agent 系统](../agents/agent-loop) - [MCP 支持](./mcp) - [Skills 系统](./skills) - [模型配置](./models) - [工具系统](./tools) --- ## MCP 集成 ### 相关页面 相关主题：[工具系统](#tools-system), [Agent 系统架构](#agent-system)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/tools/mcp/mcp_client.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) - [strands-py/src/strands/tools/mcp/mcp_agent_tool.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_agent_tool.py) - [strands-py/src/strands/tools/mcp/mcp_tasks.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_tasks.py) - [strands-py/src/strands/tools/mcp/mcp_types.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_types.py) - [strands-py/src/strands/tools/mcp/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/__init__.py) - [strands-py/docs/MCP_CLIENT_ARCHITECTURE.md](https://github.com/strands-agents/sdk-python/blob/main/strands-py/docs/MCP_CLIENT_ARCHITECTURE.md)
# 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/docs/MCP_CLIENT_ARCHITECTURE.md) ## 架构设计 ### 整体架构 MCP 客户端采用双线程架构，主线程负责同步 API，后台线程处理异步 MCP 会话通信。 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ### 工具调用流程 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ## 核心组件 ### 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) #### 关键方法 | 方法 | 返回类型 | 说明 | |------|----------|------| | `start()` | `None` | 启动 MCP 会话并加载工具 | | `stop()` | `None` | 停止 MCP 会话 | | `tools()` | `list[MCPAgentTool]` | 获取可用工具列表 | | `get_server_instructions()` | `str \| None` | 获取服务器指令 | ### MCPAgentTool `MCPAgentTool` 是对 MCP 工具的封装，使其适配 Strands 的工具接口。 ```python class MCPAgentTool(BaseTool): """MCP 工具的 Strands 封装.""" mcp_tool: Tool mcp_client: MCPClient ``` 资料来源：[strands-py/src/strands/tools/mcp/mcp_agent_tool.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_agent_tool.py) ### TasksConfig 配置 任务增强执行配置，用于长时间运行工具的任务式管理。 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_tasks.py) #### 默认配置值 | 配置项 | 默认值 | 说明 | |--------|--------|------| | `ttl` | 1 分钟 | 任务在服务器端的有效期 | | `poll_timeout` | 5 分钟 | 客户端等待任务完成的最大时间 | ## 使用模式 ### 基本 stdio 连接 通过标准输入/输出连接本地 MCP 服务器： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/README.md) ### HTTP/SSE 传输 连接远程 MCP 服务器： ```python from strands.tools.mcp import MCPClient from mcp import http_client mcp_client = MCPClient( lambda: http_client("https://mcp-server.example.com/mcp") ) ``` ### 工具过滤 使用正则表达式或精确字符串过滤可用工具： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ### 任务增强执行 为长时间运行的工具启用任务增强模式： ```python 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 ) ``` 任务增强执行的完整流程： ```mermaid flowchart LR A[call_tool_as_task] --> B[创建任务
返回 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ## 工具结果类型 MCP 工具调用返回 `MCPCallToolResult` 类型： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_types.py) ## 错误处理 ### 常见错误场景 | 错误场景 | 错误类型 | 处理方式 | |----------|----------|----------| | MCP 服务器启动超时 | `MCPClientInitializationError` | 增加 `startup_timeout` 值 | | 工具调用超时 | 底层超时异常 | 设置合理的 `read_timeout_seconds` | | 任务结果获取失败 | 任务错误结果 | 检查 `MCPCallToolResult.isError` | | 任务已完成但结果过期 | 错误结果消息 | 重新调用工具 | ### 任务错误处理 当任务执行失败时，`_create_task_error_result` 方法生成标准化的错误响应： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ### 会话状态检查 在执行工具调用前，系统会验证会话状态： ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ## 高级配置 ### 自定义传输工厂 ```python 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) ) ``` ### 工具名称前缀 为来自特定服务器的工具有选择性地添加前缀，避免命名冲突： ```python mcp_client = MCPClient( transport_callable=transport, prefix="mcp_server_" # 所有工具名将以 mcp_server_ 开头 ) ``` ### 多 MCP 服务器集成 ```python 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](https://github.com/strands-agents/sdk-python/issues/1812)），当前版本的 MCP 任务实现尚未原生支持进度报告功能。FastMCP 已支持此特性，但 Strands SDK 仍在规划中。 2. **任务 TTL 限制**：任务结果在服务器端过期后，即使任务标记为 `completed`，客户端也可能无法获取结果。详见 [MCP 规范 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/tasks)。 ### 线程安全 - `MCPClient` 本身在单线程环境中使用是安全的 - `MetricsClient` 单例存在线程安全问题（[Issue #2342](https://github.com/strands-agents/sdk-python/issues/2342)），在多线程环境中使用指标功能时需注意 ### 兼容性 - Python 版本要求：3.10+ - 旧版 SDK 不支持 MCP Tasks 功能，使用任务配置时需确保 SDK 版本兼容 ## 调试与日志 启用调试日志以排查 MCP 集成问题： ```python import logging logging.getLogger("strands.tools.mcp").setLevel(logging.DEBUG) # 关键日志点包括： # - MCP 工具调用日志 # - 任务创建和状态轮询日志 # - 会话生命周期日志 ``` 资料来源：[strands-py/src/strands/tools/mcp/mcp_client.py:50-100](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/tools/mcp/mcp_client.py) ## 相关文档 - [MCP 客户端架构文档](strands-py/docs/MCP_CLIENT_ARCHITECTURE.md) - [Agent 循环概念](https://strandsagents.com/docs/user-guide/concepts/agents/agent-loop/) - [快速开始指南](https://strandsagents.com/docs/user-guide/quickstart/) ## 另请参阅 - [工具执行器](../tools/executors.md) - 了解并发和顺序工具执行策略 - [结构化输出工具](../tools/structured_output.md) - 使用 MCP 工具返回结构化数据 - [Agent Skills 集成](../vended_plugins/skills.md) - 集成 AgentSkills.io 技能 - [遥测配置](../telemetry/config.md) - 配置 MCP 操作的追踪和监控 --- ## 双向流式交互 ### 相关页面 相关主题：[Agent 系统架构](#agent-system)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/experimental/bidi/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/__init__.py) - [strands-py/src/strands/experimental/bidi/agent/agent.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/agent/agent.py) - [strands-py/src/strands/experimental/bidi/agent/loop.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/agent/loop.py) - [strands-py/src/strands/experimental/bidi/models/gemini_live.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/models/gemini_live.py) - [strands-py/src/strands/experimental/bidi/models/nova_sonic.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/models/nova_sonic.py) - [strands-py/src/strands/experimental/bidi/models/openai_realtime.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/models/openai_realtime.py) - [strands-py/src/strands/experimental/bidi/io/audio.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/io/audio.py) - [strands-py/src/strands/experimental/bidi/io/text.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/io/text.py)
# 双向流式交互 ## 概述 双向流式交互（Bidirectional Streaming，简称 bidi）是 Strands Agents SDK 中的一个**实验性功能模块**，允许 AI 智能体与用户之间进行实时的、双向的数据传输。与传统的请求-响应模式不同，双向流式交互支持持续的数据交换，使得对话更加自然流畅，特别适用于需要实时反馈的应用场景。 资料来源：[strands-py/src/strands/experimental/bidi/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/__init__.py) ## 架构设计 ### 模块结构 双向流式交互模块位于 `strands.experimental.bidi` 命名空间下，采用分层架构设计： ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/agent/agent.py) ## 支持的模型提供商 双向流式交互模块已实现对以下实时 API 的支持： ### 模型配置表 | 模型提供商 | 实现类 | API 类型 | 支持功能 | |------------|--------|----------|----------| | Google | `GeminiLiveModel` | Gemini Live API | 文本/音频流 | | Amazon | `NovaSonicModel` | Nova Sonic API | 文本/音频流 | | OpenAI | `OpenAIRealtimeModel` | Realtime API | 文本/音频流 | 资料来源：[strands-py/src/strands/experimental/bidi/models/gemini_live.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/models/gemini_live.py) 资料来源：[strands-py/src/strands/experimental/bidi/models/nova_sonic.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/models/nova_sonic.py) 资料来源：[strands-py/src/strands/experimental/bidi/models/openai_realtime.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/models/openai_realtime.py) ## 输入输出模式 ### 文本模式 文本模式是双向流式交互的基础模式，通过 `TextIO` 类实现简单的文本输入输出流。 ```python from strands.experimental.bidi.io.text import TextIO # 创建文本输入输出处理器 text_io = TextIO() # 文本模式适合命令行界面和简单的文本交互 ``` 资料来源：[strands-py/src/strands/experimental/bidi/io/text.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/io/text.py) ### 音频模式 音频模式支持实时语音交互，通过 `AudioIO` 类处理音频输入输出流。 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/io/audio.py) ## BidiAgent 事件循环 ### 工作流程 `BidiAgentLoop` 是双向流式交互的核心事件处理引擎，负责协调用户输入、模型调用和输出处理的整个流程。 ```mermaid 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/agent/loop.py) ### 事件类型 事件循环支持多种类型的事件处理： | 事件类型 | 描述 | 触发场景 | |----------|------|----------| | 文本输入 | 用户发送文本消息 | CLI 交互、文本聊天 | | 音频输入 | 用户发送语音数据 | 语音对话 | | 模型响应 | 模型返回的文本/音频流 | AI 回复生成中 | | 状态更新 | 连接状态变化 | 会话开始/结束/错误 | ## 使用示例 ### 基本用法 ```python 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() ``` ### 音频交互模式 ```python 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](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/agent/agent.py) ## 与标准 Agent 的对比 | 特性 | 标准 Agent | BidiAgent | |------|------------|-----------| | 交互模式 | 请求-响应 | 双向流式 | | 响应延迟 | 完整响应后返回 | 实时流式返回 | | 实时反馈 | 不支持 | 支持中间进度更新 | | 输入类型 | 文本为主 | 文本/音频 | | MCP 工具支持 | 完整支持 | 有限支持（实验性） | | 上下文管理 | 自动管理 | 实时流式管理 | ## 已知限制与注意事项 ### 实验性状态 双向流式交互模块标记为**实验性功能**，这意味着： 1. **API 可能变更**：接口和行为可能在后续版本中发生变化 2. **功能不完整**：部分功能仍在开发中 3. **稳定性风险**：不建议在生产环境中使用 资料来源：[strands-py/src/strands/experimental/bidi/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/experimental/bidi/__init__.py) ### 社区相关讨论 根据社区反馈，用户对双向流式交互有以下期待： - **MCP 任务进度更新** ([Issue #1812](https://github.com/strands-agents/sdk-python/issues/1812))：用户希望 MCP 工具能够支持长期运行任务的进度反馈机制，这与双向流式交互的实时反馈能力高度相关 - **AG-UI 协议支持** ([Issue #140](https://github.com/strands-agents/sdk-python/issues/140))：社区提议通过 AG-UI 协议将 Strands 事件映射到前端应用，这与实时双向交互的需求一致 ### 线程安全注意事项 根据 SDK 的其他模块经验（如 [Issue #2342](https://github.com/strands-agents/sdk-python/issues/2342) 报告的 MetricsClient 线程安全问题），在多线程环境中使用 BidiAgent 时需要注意线程安全。建议： - 每个 BidiAgent 实例在单一线程中使用 - 避免并发调用同一实例的 `run()` 方法 - 妥善管理 IO 处理器的生命周期 ## 未来发展方向 基于社区讨论和技术路线图，双向流式交互可能朝以下方向发展： 1. **MCP 工具集成**：将 MCP 工具调用能力扩展到双向流式场景 2. **进度报告机制**：实现 MCP 任务进度更新功能 3. **AG-UI 协议支持**：提供标准化的前端连接协议 4. **多模态支持**：扩展对视频等其他模态的支持 ## 相关文档 - [快速开始指南](https://strandsagents.com/) - [模型提供商配置](../models/providers.md) - [MCP 工具集成](../tools/mcp.md) - [实验性功能](../experimental/index.md) ## 另请参阅 - [标准 Agent 文档](./agent.md) - [模型层架构](../models/index.md) - [工具执行器](../tools/executors.md) - [多智能体系统](../multiagent/index.md) --- ## 遥测与可观测性 ### 相关页面 相关主题：[会话与上下文管理](#conversation-management), [Agent 系统架构](#agent-system)
相关源码文件 以下源码文件用于生成本页说明： - [strands-py/src/strands/telemetry/__init__.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/__init__.py) - [strands-py/src/strands/telemetry/tracer.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/tracer.py) - [strands-py/src/strands/telemetry/metrics.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/metrics.py) - [strands-py/src/strands/telemetry/config.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/config.py) - [strands-py/src/strands/telemetry/logging_config.py](https://github.com/strands-agents/sdk-python/blob/main/strands-py/src/strands/telemetry/logging_config.py)
# 遥测与可观测性 Strands Agents SDK 提供了一套完整的遥测（Telemetry）与可观测性（Observability）系统，用于监控、追踪和记录 Agent 运行时的各种指标和行为。这套系统帮助开发者理解 Agent 的执行流程、诊断问题并优化性能。 ## 概述 Strands 的遥测系统包含三大核心组件： | 组件 | 用途 | 主要功能 | |------|------|----------| | **追踪器（Tracer）** | 分布式追踪 | 记录请求在 Agent 内部的完整执行路径 | | **指标（Metrics）** | 性能指标 | 收集并报告各种运行时指标数据 | | **日志配置** | 日志记录 | 配置和管理应用程序日志 | 这套系统设计为可扩展的架构，支持与多种后端服务集成，包括 OpenTelemetry 标准兼容的追踪系统。 ## 架构设计 ### 整体架构 ```mermaid 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 ``` ### 追踪器架构 ```mermaid 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 | 是否使用非安全连接 | ### 使用方式 追踪器支持通过环境变量或代码配置： ```python # 环境变量配置 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 创建流程 ```mermaid 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）。在高并发场景下，多个线程同时访问指标收集器可能导致数据竞争或程序崩溃。此问题预计在后续版本中修复。 ### 使用示例 ```python 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 使用以下命名约定： ```python # 格式：__ # 示例： agent_calls_total # Agent 调用总次数 agent_duration_seconds # Agent 执行耗时 tool_calls_total # 工具调用总次数 model_tokens_total # 模型 token 使用量 ``` ## 日志系统 ### 日志配置 日志系统提供结构化日志输出，支持配置日志级别和输出格式： ```python 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 配置不会影响其他遥测组件。 ```python 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` | 服务地址 | ## 最佳实践 ### 生产环境配置 ```python 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" } ) ``` ### 调试配置 ```python 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. 考虑关闭高开销的直方图指标 ### 线程安全问题 **问题**：多线程环境下出现异常。 **解决方案**： 1. 升级到最新版本（v1.41.1+） 2. 避免在多线程环境中共享 MetricsClient 实例 3. 使用线程 Error with Openai API: output new_sensitive (1027) Please check that you have set the OPENAI_API_KEY environment variable with a valid API key. --- --- ## Doramagic 踩坑日志 项目：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