OA0 = Omni AI 0
OA0 是一个探索 AI 的论坛
现在注册
已注册用户请  登录
OA0  ›  技能包  ›  aws-strands:使用 AWS Strands SDK 构建并运行 Python 智能体

aws-strands:使用 AWS Strands SDK 构建并运行 Python 智能体

 
  user ·  2026-02-23 14:39:47 · 2 次点击  · 0 条评论  

名称: strands
版本: 2.0.0
描述: 使用 AWS Strands SDK 构建和运行基于 Python 的 AI 智能体。适用于创建自主智能体、多智能体工作流、自定义工具或与 MCP 服务器集成。支持 Ollama(本地)、Anthropic、OpenAI、Bedrock 等多种模型提供商。可用于智能体脚手架搭建、工具创建和以编程方式运行智能体任务。
主页: https://github.com/strands-agents/sdk-python
元数据:
openclaw:
emoji: 🧬
requires:
bins: [python3]
packages: [strands-agents]


Strands 智能体 SDK

使用 Strands SDK(Apache-2.0 许可,来自 AWS)在 Python 中构建 AI 智能体。

已验证版本:strands-agents==1.23.0strands-agents-tools==0.2.19

先决条件

# 安装 SDK + 工具(推荐使用 pipx 进行隔离安装)
pipx install strands-agents-builder  # 包含 strands-agents、strands-agents-tools 和 CLI

# 或直接安装
pip install strands-agents strands-agents-tools

核心概念:Bedrock 是默认模型

不指定 model= 参数的 Agent() 默认使用 Amazon Bedrock —— 具体是 us-west-2 区域的 us.anthropic.claude-sonnet-4-20250514-v1:0。这需要 AWS 凭证。要使用其他提供商,请显式传递 model= 参数。

默认模型常量:strands.models.bedrock.DEFAULT_BEDROCK_MODEL_ID

快速开始 — 本地智能体 (Ollama)

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

# host 是必需的位置参数
model = OllamaModel("http://localhost:11434", model_id="qwen3:latest")
agent = Agent(model=model)
result = agent("法国的首都是什么?")
print(result)

注意: 并非所有开源模型都支持工具调用。经过“消融”处理的模型通常在处理过程中会失去函数调用能力。建议先用标准模型(如 qwen3, llama3.x, mistral)进行测试。

快速开始 — Bedrock(默认提供商)

from strands import Agent

# 未指定模型 → 使用 BedrockModel(Claude Sonnet 4, us-west-2)
# 需要 AWS 凭证(~/.aws/credentials 或环境变量)
agent = Agent()
result = agent("解释一下量子计算")

# 显式指定 Bedrock 模型:
from strands.models import BedrockModel
model = BedrockModel(model_id="us.anthropic.claude-sonnet-4-20250514-v1:0")
agent = Agent(model=model)

快速开始 — Anthropic(直接 API)

from strands import Agent
from strands.models.anthropic import AnthropicModel

# max_tokens 是 Required[int] — 必须提供
model = AnthropicModel(model_id="claude-sonnet-4-20250514", max_tokens=4096)
agent = Agent(model=model)
result = agent("解释一下量子计算")

需要设置 ANTHROPIC_API_KEY 环境变量。

快速开始 — OpenAI

from strands import Agent
from strands.models.openai import OpenAIModel

model = OpenAIModel(model_id="gpt-4.1")
agent = Agent(model=model)

需要设置 OPENAI_API_KEY 环境变量。

创建自定义工具

使用 @tool 装饰器。类型提示将转换为模式;文档字符串将转换为描述:

from strands import Agent, tool

@tool
def read_file(path: str) -> str:
    """读取给定路径的文件内容。

    Args:
        path: 要读取的文件系统路径。
    """
    with open(path) as f:
        return f.read()

@tool
def write_file(path: str, content: str) -> str:
    """将内容写入文件。

    Args:
        path: 要写入的文件系统路径。
        content: 要写入的文本内容。
    """
    with open(path, 'w') as f:
        f.write(content)
    return f"已将 {len(content)} 字节写入 {path}"

agent = Agent(model=model, tools=[read_file, write_file])
agent("读取 /tmp/test.txt 并总结其内容")

ToolContext

工具可以通过 ToolContext 访问智能体状态:

from strands import tool
from strands.types.tools import ToolContext

@tool
def stateful_tool(query: str, tool_context: ToolContext) -> str:
    """一个可以访问智能体状态的工具。

    Args:
        query: 输入查询。
    """
    # 访问共享的智能体状态
    count = tool_context.state.get("call_count", 0) + 1
    tool_context.state["call_count"] = count
    return f"调用 #{count}: {query}"

内置工具(46 个可用)

strands-agents-tools 提供了预构建的工具:

from strands_tools import calculator, file_read, file_write, shell, http_request
agent = Agent(model=model, tools=[calculator, file_read, shell])

完整列表:calculator, file_read, file_write, shell, http_request, editor, image_reader, python_repl, current_time, think, stop, sleep, environment, retrieve, search_video, chat_video, speak, generate_image, generate_image_stability, diagram, journal, memory, agent_core_memory, elasticsearch_memory, mongodb_memory, mem0_memory, rss, cron, batch, workflow, use_agent, use_llm, use_aws, use_computer, load_tool, handoff_to_user, slack, swarm, graph, a2a_client, mcp_client, exa, tavily, bright_data, nova_reels.

热重载:Agent(load_tools_from_directory=True) 会监视 ./tools/ 目录的更改。

MCP 集成

连接到任何模型上下文协议(MCP)服务器。MCPClient 实现了 ToolProvider —— 可以直接将其传入工具列表:

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

# MCPClient 接收一个返回传输层的可调用对象
mcp = MCPClient(lambda: stdio_client(StdioServerParameters(
    command="uvx",
    args=["some-mcp-server@latest"]
)))

# 用作上下文管理器 — MCPClient 是一个 ToolProvider
with mcp:
    agent = Agent(model=model, tools=[mcp])
    agent("使用 MCP 工具来完成某项任务")

SSE 传输:

from mcp.client.sse import sse_client
mcp = MCPClient(lambda: sse_client("http://localhost:8080/sse"))

多智能体模式

智能体作为工具

嵌套智能体 —— 内部智能体成为外部智能体的工具:

researcher = Agent(model=model, system_prompt="你是一个研究助手。")
writer = Agent(model=model, system_prompt="你是一个作家。")

orchestrator = Agent(
    model=model,
    tools=[researcher, writer],
    system_prompt="你负责协调研究和写作任务。"
)
orchestrator("研究量子计算并撰写一篇博客文章")

群集模式

具有共享上下文和自主交接协调能力的自组织智能体团队:

from strands.multiagent.swarm import Swarm

# 智能体需要 name 和 description 以便进行交接识别
researcher = Agent(
    model=model,
    name="研究员",
    description="查找并总结信息"
)
writer = Agent(
    model=model,
    name="作家",
    description="创作精炼的内容"
)

swarm = Swarm(
    nodes=[researcher, writer],
    entry_point=researcher,    # 可选 — 默认为第一个智能体
    max_handoffs=20,           # 默认值
    max_iterations=20,         # 默认值
    execution_timeout=900.0,   # 默认 15 分钟
    node_timeout=300.0         # 每个节点默认 5 分钟
)
result = swarm("研究 AI 智能体,然后交接给作家撰写博客文章")

群集会自动注入一个 handoff_to_agent 工具。智能体通过调用此工具并指定目标智能体的名称来进行交接。支持中断/恢复、会话持久化和重复交接检测。

图模式(DAG)

通过 GraphBuilder 实现基于依赖关系的确定性执行:

from strands.multiagent.graph import GraphBuilder

builder = GraphBuilder()
research_node = builder.add_node(researcher, node_id="research")
writing_node = builder.add_node(writer, node_id="writing")
builder.add_edge("research", "writing")
builder.set_entry_point("research")

# 可选:条件边
# builder.add_edge("research", "writing",
#     condition=lambda state: "complete" in str(state.completed_nodes))

graph = builder.build()
result = graph("撰写一篇关于 AI 智能体的博客文章")

支持循环(反馈循环)使用 builder.reset_on_revisit(True)、执行超时和嵌套图(Graph 作为另一个 Graph 的节点)。

A2A 协议(智能体到智能体)

将 Strands 智能体暴露为 A2A 兼容服务器,用于智能体间通信:

from strands.multiagent.a2a import A2AServer

server = A2AServer(
    agent=my_agent,
    host="127.0.0.1",
    port=9000,
    version="0.0.1"
)
server.start()  # 运行 uvicorn

使用 strands-agents-tools 中的 a2a_client 工具连接到 A2A 智能体。A2A 实现了 Google 的智能体到智能体协议,用于标准化的跨进程/跨网络智能体通信。

会话持久化

跨智能体运行持久化对话:

from strands.session.file_session_manager import FileSessionManager

session = FileSessionManager(session_file_path="./sessions/my_session.json")
agent = Agent(model=model, session_manager=session)

# 也可用:
from strands.session.s3_session_manager import S3SessionManager
session = S3SessionManager(bucket_name="my-bucket", session_id="session-1")

群集和图都支持会话管理器,用于持久化多智能体状态。

双向流式传输(实验性)

具有持久音频流的实时语音/文本对话:

from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models.nova_sonic import NovaSonicModel

# 支持:NovaSonicModel, GeminiLiveModel, OpenAIRealtimeModel
model = NovaSonicModel(region="us-east-1")
agent = BidiAgent(model=model, tools=[my_tool])

支持中断检测、并发工具执行和连续来回音频。实验性功能 —— API 可能变更。

系统提示词

agent = Agent(
    model=model,
    system_prompt="你是 Hex,一个敏锐而机智的 AI 助手。",
    tools=[read_file, write_file]
)

Strands 也支持 list[SystemContentBlock] 用于具有缓存控制的结构化系统提示词。

可观测性

原生 OpenTelemetry 追踪:

agent = Agent(
    model=model,
    trace_attributes={"project": "my-agent", "environment": "dev"}
)

每个工具调用、模型调用、交接和生命周期事件都是可插装的。

Bedrock 特定功能

  • 护栏:在 BedrockModel 配置中设置 guardrail_id + guardrail_version —— 内容过滤、PII 检测、输入/输出编辑
  • 缓存点:系统提示词和工具定义缓存,用于成本优化
  • 流式传输:默认开启,可通过 streaming=False 禁用
  • 区域:默认为 us-west-2,可通过 region_name 参数或 AWS_REGION 环境变量覆盖
  • 跨区域推理:以 us. 为前缀的模型 ID 使用跨区域推理配置文件

搭建新智能体脚手架

python3 {baseDir}/scripts/create-agent.py my-agent --provider ollama --model qwen3:latest
python3 {baseDir}/scripts/create-agent.py my-agent --provider anthropic
python3 {baseDir}/scripts/create-agent.py my-agent --provider bedrock
python3 {baseDir}/scripts/create-agent.py my-agent --provider openai --model gpt-4.1

创建一个包含工具、配置和入口点的可运行智能体目录。

运行智能体

python3 {baseDir}/scripts/run-agent.py path/to/agent.py "你的提示词"
python3 {baseDir}/scripts/run-agent.py path/to/agent.py --interactive

模型提供商参考(共 11 个)

提供商 初始化 备注
Bedrock BedrockModel BedrockModel(model_id=...) 默认,急切导入
Ollama OllamaModel OllamaModel("http://host:11434", model_id=...) host 是位置参数
Anthropic AnthropicModel AnthropicModel(model_id=..., max_tokens=4096) max_tokens 必需
OpenAI OpenAIModel OpenAIModel(model_id=...) OPENAI_API_KEY
Gemini GeminiModel GeminiModel(model_id=...) api_key 在 client_args 中
Mistral MistralModel MistralModel(model_id=...) Mistral API 密钥
LiteLLM LiteLLMModel LiteLLMModel(model_id=...) 元提供商(Cohere, Groq 等)
LlamaAPI LlamaAPIModel LlamaAPIModel(model_id=...) Meta Llama API
llama.cpp LlamaCppModel LlamaCppModel(...) 本地服务器,OpenAI 兼容
SageMaker SageMakerAIModel SageMakerAIModel(...) 自定义 AWS 端点
Writer WriterModel WriterModel(model_id=...) Writer 平台

所有非 Bedrock 提供商都是懒加载的 —— 仅在引用时导入依赖项。

导入模式:from strands.models.<provider> import <Class>(或使用 from strands.models import <Class> 进行懒加载)。

提示与技巧

  • 不带 model=Agent() 需要 AWS 凭证(默认使用 Bedrock)
  • AnthropicModel 需要 max_tokens —— 省略会导致运行时错误
  • OllamaModelhost 是位置参数:OllamaModel("http://...", model_id="...")
  • 经过“消融”处理的 Ollama 模型通常会失去工具调用支持 —— 对于使用工具的智能体,请使用标准模型
  • 群集智能体需要 name=description= 以便进行交接路由
  • Agent(load_tools_from_directory=True) 会监视 ./tools/ 目录以热重载工具文件
  • 使用 agent.tool.my_tool() 可以直接调用工具,无需 LLM 路由
  • MCPClient 是一个 ToolProvider —— 直接在 tools=[mcp] 中传递,与 Agent 一起使用时不要手动调用 list_tools_sync()
  • 会话管理器可与 Agent、Swarm 和 Graph 一起使用
  • 固定你的 strands-agents 版本 —— SDK 较新,API 在版本间可能会演变
2 次点击  ∙  0 人收藏  
登录后收藏  
目前尚无回复
0 条回复
About   ·   Help   ·    
OA0 - Omni AI 0 一个探索 AI 的社区
沪ICP备2024103595号-2
Developed with Cursor