1. Strands Agents SDK 与 Amazon Bedrock 技术解析

Strands Agents SDK 是一个专为构建生产级 AI Agent 系统设计的 Python 框架，而 Amazon Bedrock 是 AWS 提供的全托管基础模型服务。两者的结合为开发者提供了快速构建智能代理的能力。

1.1 Strands Agents SDK 核心特性

这个 SDK 的设计哲学体现在几个关键方面：

• 极简 API 设计：通过装饰器模式将复杂功能封装成简单接口
• 工具即代码：任何 Python 函数只需添加 @tool 装饰器即可成为 Agent 可调用的工具
• 多模型支持：统一接口适配不同基础模型，包括 Bedrock 上的 Claude、Llama 等

from strands import tool, Agent @tool def get_weather(city: str) -> dict: """获取城市天气信息""" # 实现天气查询逻辑 return weather_data weather_agent = Agent( tools=[get_weather], model="bedrock/anthropic.claude-v2" )

1.2 Amazon Bedrock 的核心价值

Bedrock 为 AI Agent 提供了三项关键能力：

1. 模型即服务：无需维护基础设施即可访问顶尖大模型
2. 知识库集成：支持将企业文档作为检索增强生成(RAG)的数据源
3. 安全护栏：内置内容过滤和 PII 保护机制

2. 20 行 Python 实现多功能 Agent

2.1 基础环境准备

首先确保环境配置正确：

pip install strands-agents boto3 export AWS_ACCESS_KEY_ID=your_key export AWS_SECRET_ACCESS_KEY=your_secret

2.2 核心代码实现

以下是一个完整的多功能 Agent 实现：

from strands import Agent, tool import math @tool def get_weather(city: str) -> str: """查询城市天气，示例返回模拟数据""" return f"{city}天气：晴，25℃" @tool def calculate(expression: str) -> str: """计算数学表达式""" try: return str(eval(expression)) except: return "计算失败，请检查表达式" @tool def read_file(path: str) -> str: """读取文件内容""" try: with open(path) as f: return f.read(1000) # 限制读取长度 except: return "文件读取失败" agent = Agent( tools=[get_weather, calculate, read_file], model="bedrock/anthropic.claude-v2", system_prompt="你是一个多功能助手，可以查天气、做数学计算和阅读文档" ) response = agent("北京今天天气如何？") print(response)

2.3 代码结构解析

这段代码实现了三个核心功能：

1. 天气查询：通过 get_weather 工具函数
2. 数学计算：利用 Python 的 eval 函数实现
3. 文档阅读：简单的文件读取功能

3. 功能扩展与实战技巧

3.1 增加实时数据查询

改造天气查询功能，接入真实 API：

import requests @tool def get_real_weather(city: str) -> str: """接入真实天气API""" api_key = "your_api_key" url = f"https://api.weather.com/v3/wx/conditions/current?city={city}&key={api_key}" try: res = requests.get(url, timeout=3) data = res.json() return f"{city}天气：{data['conditions']}，温度{data['temp']}℃" except: return "天气查询服务暂不可用"

3.2 增强文档处理能力

使用 PyPDF2 扩展文档阅读功能：

from PyPDF2 import PdfReader @tool def read_pdf(path: str) -> str: """读取PDF文件内容""" try: reader = PdfReader(path) text = "\n".join([page.extract_text() for page in reader.pages[:3]]) return text[:1000] # 限制返回长度 except: return "PDF文件读取失败"

3.3 错误处理最佳实践

完善的错误处理机制对生产环境至关重要：

@tool def safe_calculate(expression: str) -> str: """安全的数学计算""" allowed_chars = set("0123456789+-*/.() ") if not all(c in allowed_chars for c in expression): return "表达式包含非法字符" try: return str(eval(expression)) except ZeroDivisionError: return "错误：除数不能为零" except: return "计算失败，请检查表达式"

4. 生产环境部署方案

4.1 性能优化策略

对于高频调用的 Agent 需要实施优化：

1. 缓存机制：对天气查询等结果缓存 10 分钟
2. 异步处理：使用 async/await 提高并发能力
3. 批处理：支持多个表达式同时计算

from functools import lru_cache import asyncio @lru_cache(maxsize=100) @tool def cached_weather(city: str) -> str: """带缓存的天气查询""" return get_real_weather(city) async def async_calculate(exprs: list[str]) -> list[str]: """批量计算""" return [calculate(expr) for expr in exprs]

4.2 安全防护措施

必须考虑的安全防护层面：

1. 输入验证：过滤特殊字符和恶意输入
2. 权限控制：限制文件读取目录
3. 速率限制：防止API滥用

import os SAFE_DIR = "/var/www/data" @tool def safe_read(path: str) -> str: """安全的文件读取""" if not os.path.abspath(path).startswith(SAFE_DIR): return "无权访问该路径" if ".." in path: return "路径不合法" return read_file(path)

5. 典型问题排查指南

5.1 工具调用失败分析

常见工具调用问题及解决方案：

5.2 Bedrock 模型响应异常

模型相关问题的处理建议：

1. 响应不符合预期：优化 system_prompt 提供更明确指令
2. 输出被截断：调整 max_tokens 参数
3. 响应速度慢：尝试更轻量级的模型版本

optimized_agent = Agent( system_prompt=""" 你是一个专业助手，请严格遵守以下规则： 1. 只回答与工具功能相关的问题 2. 不确定时回答"我不知道" 3. 保持回答简洁专业 """, model="bedrock/anthropic.claude-instant-v1", # 更轻量模型 max_tokens=500 )

5.3 性能瓶颈定位

使用以下方法分析性能问题：

from datetime import datetime class Timer: def __enter__(self): self.start = datetime.now() def __exit__(self, *args): print(f"耗时：{(datetime.now()-self.start).total_seconds()}秒") with Timer(): response = agent("计算(3+4)*5的值")

6. 进阶开发技巧

6.1 多Agent协作模式

实现Agent间的协同工作：

from strands import Agent, tool @tool def math_agent(query: str) -> str: """专用数学计算Agent""" agent = Agent( system_prompt="你是一个数学专家，只回答数学问题", tools=[calculate] ) return agent(query) @tool def weather_agent(query: str) -> str: """专用天气查询Agent""" agent = Agent( system_prompt="你是一个天气助手，只回答天气相关问题", tools=[get_weather] ) return agent(query) coordinator = Agent( tools=[math_agent, weather_agent], system_prompt="分析用户问题并路由到合适的专业Agent" )

6.2 长期记忆实现

使用数据库保存会话上下文：

import sqlite3 from datetime import datetime conn = sqlite3.connect("memory.db") conn.execute("CREATE TABLE IF NOT EXISTS memory (key TEXT, value TEXT, time TEXT)") @tool def remember(key: str, value: str) -> bool: """存储信息到长期记忆""" conn.execute( "INSERT INTO memory VALUES (?, ?, ?)", (key, value, datetime.now().isoformat()) ) conn.commit() return True @tool def recall(key: str) -> str: """从长期记忆检索信息""" cur = conn.execute( "SELECT value FROM memory WHERE key=? ORDER BY time DESC LIMIT 1", (key,) ) if row := cur.fetchone(): return row[0] return "没有找到相关记忆"

6.3 流式响应处理

对于长时间任务实现流式输出：

async def stream_response(): agent = Agent(tools=[calculate]) async for chunk in agent.stream_async("计算圆周率前100位"): print(chunk["text"], end="", flush=True) import asyncio asyncio.run(stream_response())