基于strands-agents的AI代理开发：从工具调用到生产部署

1. 项目概述一个面向AI代理开发的Python SDK如果你最近在尝试构建一个能够自主执行复杂任务的AI代理比如让它帮你分析数据、自动回复邮件甚至管理一个项目流程那你大概率会遇到一个核心难题如何让大语言模型LLM不只是“聊天”而是能“动手做事”。你需要让它能调用工具、管理状态、记住上下文并协调多个步骤。这正是strands-agents/sdk-python这个项目要解决的核心问题。它不是一个简单的API封装而是一个专为构建生产级、可扩展的AI代理Agent而设计的Python软件开发工具包。简单来说strands-agents/sdk-python提供了一套标准化的“积木”和“蓝图”让你能快速、优雅地将大语言模型如GPT-4、Claude、本地部署的模型与外部工具API、数据库、函数连接起来组装成一个能感知环境、规划行动、执行任务并从中学习的智能体。它抽象了代理开发中的通用模式比如工具调用、记忆管理、工作流编排和错误处理让你能专注于业务逻辑本身而不是重复造轮子。无论你是想做一个智能客服机器人、一个自动化数据分析助手还是一个复杂的业务流程自动化代理这个SDK都能为你提供一个坚实且灵活的起点。2. 核心设计理念与架构拆解2.1 为什么需要专门的Agent SDK在深入代码之前我们先聊聊“为什么”。直接用OpenAI的API写个函数调用Function Calling不也能让模型使用工具吗确实可以但对于构建一个稍复杂的代理系统你会很快陷入泥潭。你需要手动管理对话历史记忆处理工具调用的输入输出解析编排多个工具调用的顺序规划处理执行中的异常还要考虑如何将代理的状态持久化。这些“胶水代码”会迅速膨胀变得难以维护和测试。strands-agents/sdk-python的核心理念是关注点分离和可组合性。它将一个代理的典型生命周期分解为几个清晰的组件代理Agent核心决策者根据目标、记忆和当前观察决定下一步行动调用哪个工具或直接输出。工具Tool代理可以调用的具体能力如搜索网络、查询数据库、执行代码。记忆Memory存储和检索代理与用户的历史交互、工具执行结果等上下文信息。执行器Executor驱动代理运行循环的引擎负责调用模型、执行工具、更新状态。状态State代理在单次运行会话中的所有上下文数据是连接各个组件的纽带。这种架构让你可以像搭乐高一样独立地更换或升级每个部件。例如你可以轻松地将记忆后端从简单的内存列表切换到Redis或者为同一个代理更换不同的LLM提供商而无需重写核心逻辑。2.2 SDK的核心抽象State与Message这是理解该SDK的关键。与许多其他框架不同strands-agents重度依赖一个名为State的字典状对象来贯穿整个执行流程。State包含了当前会话的所有信息用户输入、模型响应、工具调用结果、自定义数据等。更具体地说State中最重要的部分之一是messages列表。这里存储着所有类型的Message对象例如HumanMessage用户输入、AIMessageAI回复、ToolMessage工具执行结果。代理的决策基于State中的messages历史而每次行动无论是思考还是工具调用都会产生新的Message并追加到State中从而推动状态演进。这种设计带来了极大的灵活性。你可以编写中间件Middleware在代理执行的各个阶段如调用模型前、执行工具后注入逻辑来修改或观察State。例如实现一个日志中间件来记录所有交互或者一个验证中间件来检查工具输入的合法性。3. 从零开始构建你的第一个智能代理3.1 环境准备与基础安装首先确保你的Python环境在3.8以上。创建一个新的虚拟环境是一个好习惯。python -m venv agent-env source agent-env/bin/activate # Linux/macOS # 或 agent-env\Scripts\activate # Windows然后安装strands-agentsSDK。由于它可能处于活跃开发中建议从GitHub仓库直接安装最新版本并同时安装一些常用的额外依赖。pip install “strands-agents[openai]” # 安装核心库及OpenAI集成 # 或者如果你想从源码安装最新开发版 # pip install githttps://github.com/strands-ai/strands-agents.git这里我们安装了[openai]额外依赖因为它包含了与OpenAI API交互的必要组件。如果你计划使用Anthropic的Claude或其他模型可能需要安装对应的集成包如[anthropic]。3.2 定义你的第一个工具Tool工具是代理的手和脚。让我们创建一个简单的工具让代理能获取当前时间。from datetime import datetime from strands.agents.tools import tool tool def get_current_time(timezone: str “UTC”) - str: “”” 获取指定时区的当前时间。 Args: timezone: 时区名称例如 ‘Asia/Shanghai’。默认为 ‘UTC’。 Returns: 格式化后的当前时间字符串。 “”” # 这是一个简化示例实际应用中应使用pytz等库处理时区 now datetime.utcnow() if timezone ! “UTC”: # 提示生产环境请使用pytz或zoneinfo return f“当前 {timezone} 时间约为: {now.strftime(‘%Y-%m-%d %H:%M:%S’)} (示例未真实转换时区)” return f“当前UTC时间: {now.strftime(‘%Y-%m-%d %H:%M:%S’)}”关键点解析使用tool装饰器将一个普通Python函数声明为代理可用的工具。函数的文档字符串“””至关重要LLM会阅读它来理解工具的功能、参数和返回值。描述务必清晰准确。输入参数最好有类型注解如str和默认值这有助于SDK和LLM进行更好的解析。3.3 组装代理并运行现在我们将工具、LLM和记忆组合成一个可运行的代理。import asyncio from strands.agents import Agent, OpenAIChat from strands.agents.memory import SimpleMemory # 1. 初始化LLM这里使用OpenAI GPT-4 # 请确保已设置环境变量 OPENAI_API_KEY llm OpenAIChat(model“gpt-4-turbo-preview”) # 2. 初始化记忆这里使用简单的内存记忆会话结束后消失 memory SimpleMemory() # 3. 创建代理并传入我们定义的工具列表 agent Agent( llmllm, tools[get_current_time], # 将工具函数放入列表 memorymemory, system_prompt“你是一个乐于助人的助手可以告诉用户时间。”, # 系统指令设定代理角色 ) # 4. 运行代理异步方式 async def main(): # 初始化状态包含用户的初始消息 initial_state {“messages”: [{“role”: “user”, “content”: “现在上海是几点钟了”}]} # 运行代理它会自动处理思考、工具调用、响应的全过程 final_state await agent.run(stateinitial_state) # 从最终状态中提取AI的最后一条消息 last_message final_state[“messages”][-1] print(f“助手回复: {last_message[‘content’]}”) # 执行异步函数 if __name__ “__main__”: asyncio.run(main())当你运行这段代码时代理会经历以下内部流程接收包含用户消息的初始State。Agent将State中的消息历史这里只有用户消息和系统提示一起发送给LLM。LLM分析后认为需要调用get_current_time工具并生成一个符合格式的工具调用请求。Agent解析这个请求执行对应的get_current_time函数并将结果封装为一个ToolMessage添加到State中。Agent再次将更新后的State包含用户消息、工具调用请求、工具结果发送给LLM。LLM根据工具返回的结果生成最终的自然语言回复作为一个AIMessage添加到State。流程结束返回最终的State。在控制台你可能会看到类似这样的输出助手回复: 根据工具返回的信息当前上海时间约为: 2024-05-27 03:14:22 (示例未真实转换时区)。4. 核心进阶深入理解与定制化4.1 状态State的精细化管理State是SDK的血液。理解如何操作它是进行高级定制的基础。State本质上是一个字典但SDK提供了一些工具函数来安全地读写它。from strands.agents import State # 假设我们有一个初始状态 state State({“messages”: [], “user_id”: “123”, “session_id”: “abc”}) # 安全地更新状态 - 推荐方式避免直接修改 from strands.agents import append_to_state new_state append_to_state(state, {“new_key”: “value”}) # new_state 现在包含 {“messages”: [], “user_id”: “123”, “session_id”: “abc”, “new_key”: “value”} # 专门用于添加消息的方法 from strands.agents import add_message state_with_msg add_message(state, {“role”: “user”, “content”: “你好”}) # state_with_msg[“messages”] 现在包含一条用户消息 # 在工具函数中访问状态 tool def check_status(state: State) - str: user_id state.get(“user_id”, “unknown”) return f“当前用户 {user_id} 的状态正常。”实操心得在编写自定义工具或中间件时尽量使用SDK提供的工具函数如append_to_state,add_message来操作状态而不是直接修改字典。这能保证状态更新的可预测性并兼容未来可能的内部变更。4.2 构建复杂的工具Tool生态一个强大的代理离不开丰富的工具集。除了简单的函数工具还可以是类方法、异步函数甚至是对其他服务的复杂封装。import aiohttp from pydantic import BaseModel, Field from strands.agents.tools import tool # 示例1使用Pydantic模型定义复杂输入的工具 class WeatherQuery(BaseModel): city: str Field(description“城市名称例如北京”) unit: str Field(default“celsius”, description“温度单位’celsius‘ 或 ‘fahrenheit‘”) tool(args_schemaWeatherQuery) async def get_weather(query: WeatherQuery) - str: “””获取指定城市的天气信息。””” # 注意这是一个模拟函数实际需要调用天气API async with aiohttp.ClientSession() as session: # 模拟API调用 await asyncio.sleep(0.1) return f“{query.city}的天气是晴朗温度25{‘°C’ if query.unit ‘celsius’ else ‘°F’}。” # 示例2一个需要访问状态上下文的工具 tool def summarize_conversation(state: State) - str: “””总结当前的对话内容。””” messages state.get(“messages”, []) user_msgs [msg[“content”] for msg in messages if msg[“role”] “user”] summary f“用户共询问了{len(user_msgs)}个问题最新问题是{user_msgs[-1][:50]}...” return summary # 将工具列表传给Agent时 agent Agent( llmllm, tools[get_weather, summarize_conversation, get_current_time], # … 其他配置 )注意事项异步工具如果工具涉及网络I/O如调用API务必将其定义为async函数并使用await。这能避免阻塞整个代理执行循环提升并发性能。参数验证使用args_schema并传入一个Pydantic模型可以自动对工具输入进行验证和解析确保LLM提供的参数格式正确大大减少运行时错误。工具描述工具的文档字符串和Pydantic模型的字段描述是LLM理解工具的“说明书”。写得越清晰、越具体LLM调用工具的准确率就越高。4.3 记忆Memory系统的选择与集成SimpleMemory适合快速原型但对于需要持久化或复杂检索的生产环境你需要更强大的记忆系统。from strands.agents.memory import BufferMemory, RedisMemory # 1. BufferMemory保留最近N条交互的滑动窗口 buffer_memory BufferMemory(max_messages20) # 只记住最近20条消息 # 适用于关注近期对话的场景避免上下文过长导致模型性能下降或API费用过高。 # 2. RedisMemory使用Redis进行持久化存储 # 首先安装 redis 包: pip install redis redis_memory RedisMemory( redis_url“redis://localhost:6379/0”, session_ttl3600, # 会话数据过期时间秒 ) # 适用于多实例部署、需要会话持久化如用户下次回来继续对话的场景。 # 使用记忆 agent_with_memory Agent(llmllm, tools[…], memoryredis_memory)内存管理心得记忆不仅仅是存储消息。BufferMemory的max_messages参数需要仔细权衡。设置太小代理可能忘记重要的早期信息设置太大会增加每次调用LLM的令牌消耗和成本。一个常见的策略是结合使用用BufferMemory管理最近对话同时用另一个工具如summarize_conversation定期将长程记忆总结后以系统消息的形式注入上下文。5. 实战构建一个自动化数据分析代理让我们综合运用以上知识构建一个稍微复杂点的代理它能根据用户的自然语言描述对一份给定的CSV数据进行简单的分析和可视化。5.1 定义数据分析工具集我们假设数据已经以Pandas DataFrame的形式存在于状态中。import pandas as pd import matplotlib.pyplot as plt import io import base64 from strands.agents.tools import tool tool def load_csv_data(state: State, csv_content: str) - str: “”” 将CSV字符串加载到Pandas DataFrame中并存储到状态。 后续工具可以操作这个DataFrame。 “”” try: from io import StringIO df pd.read_csv(StringIO(csv_content)) # 将DataFrame存储到状态中供其他工具使用 state[“dataframe”] df state[“dataframe_name”] “loaded_data” return f“数据加载成功共 {len(df)} 行{len(df.columns)} 列。列名{‘ ‘.join(df.columns.tolist())}” except Exception as e: return f“加载CSV数据失败{str(e)}” tool def describe_data(state: State) - str: “””提供当前加载数据的基本统计描述。””” df state.get(“dataframe”) if df is None: return “未找到已加载的数据。请先使用 ‘load_csv_data‘ 工具加载数据。” buffer io.StringIO() df.describe(include‘all’).to_string(buffer) return buffer.getvalue() tool def plot_histogram(state: State, column: str) - str: “””为指定数值列生成直方图并返回图片的Base64编码。””” df state.get(“dataframe”) if df is None: return “错误数据未加载。” if column not in df.columns: return f“错误数据中不存在列 ‘{column}‘。” if not pd.api.types.is_numeric_dtype(df[column]): return f“错误列 ‘{column}‘ 不是数值类型无法绘制直方图。” plt.figure(figsize(10, 6)) df[column].hist(bins20, edgecolor‘black’) plt.title(f‘Distribution of {column}‘) plt.xlabel(column) plt.ylabel(‘Frequency’) # 将图形保存到内存缓冲区并转换为Base64 img_buffer io.BytesIO() plt.savefig(img_buffer, format‘png’, bbox_inches‘tight’) plt.close() # 关闭图形释放内存 img_buffer.seek(0) img_base64 base64.b64encode(img_buffer.read()).decode(‘utf-8’) # 将图片存储到状态并返回一个可读的消息 state[“last_plot”] img_base64 return f“已生成列 ‘{column}‘ 的直方图。图片已保存你可以使用 ‘show_last_plot‘ 工具查看。” tool def show_last_plot(state: State) - str: “””返回最近生成的图片的HTML展示代码适用于Jupyter环境。””” img_base64 state.get(“last_plot”) if not img_base64: return “没有可显示的图片。请先使用 ‘plot_histogram‘ 生成一张图。” # 返回一个HTML img标签在支持HTML的环境如Jupyter中可以直接渲染 html_img f‘img src“data:image/png;base64,{img_base64}” alt“Last Plot”’ state[“last_plot_html”] html_img return f“图片已准备就绪。在支持HTML的环境下以下代码将显示图片\n\n{html_img}”5.2 创建并运行数据分析代理async def run_data_analysis_agent(): llm OpenAIChat(model“gpt-4”) memory BufferMemory(max_messages30) agent Agent( llmllm, tools[load_csv_data, describe_data, plot_histogram, show_last_plot], memorymemory, system_prompt“”” 你是一个数据分析助手。用户会提供CSV格式的数据你需要根据他们的要求调用相应的工具来加载、描述或可视化数据。 请按步骤思考清晰地告诉用户你每一步要做什么。 如果用户的要求不明确请主动询问细节例如要绘制直方图的列名是什么。 “””, ) # 模拟一份CSV数据例如身高体重数据 sample_csv “””height,weight,age 175,70,25 180,80,30 165,55,28 190,85,35 170,65,22 “”” # 初始状态用户要求分析数据 initial_state State({ “messages”: [ {“role”: “user”, “content”: f“我这里有一份数据\n{sample_csv}\n请帮我分析一下先告诉我数据的基本情况然后为‘height’列画个分布图。”} ] }) print(“用户提问”, initial_state[“messages”][-1][“content”][:50], “…) print(“\n--- 代理开始执行 ---\n”) final_state await agent.run(stateinitial_state) # 打印整个对话过程 for msg in final_state[“messages”]: if msg[“role”] “assistant”: print(f“助手: {msg[‘content’]}”) elif msg[“role”] “tool”: print(f“[工具 {msg[‘name’]}]: {msg[‘content’]}”) print(“\n--- 执行结束 ---”) # 此时final_state中应该包含了直方图的Base64数据 if __name__ “__main__”: asyncio.run(run_data_analysis_agent())这个代理将能够理解用户的复合指令先描述再画图自动按顺序调用load_csv_data、describe_data和plot_histogram工具并将结果反馈给用户。在Jupyter Notebook环境中show_last_plot工具返回的HTML可以直接渲染出图片。6. 生产环境部署与性能调优6.1 错误处理与鲁棒性增强在生产中代理必须能优雅地处理各种异常。from strands.agents import Agent from strands.agents.exceptions import ToolExecutionError, MaxStepsExceededError # 方案1在Agent级别配置重试和超时 robust_agent Agent( llmllm, tools[…], max_steps10, # 限制单次运行的最大步骤思考工具调用数防止死循环 retry_policy{ “max_attempts”: 2, # 工具调用失败重试次数 “delay”: 0.5, # 重试延迟秒 } ) # 方案2在工具函数内部进行细粒度错误捕获 tool def call_unstable_api(state: State, param: str) - str: “””调用一个可能不稳定的外部API。””” try: # … 模拟API调用 if random.random() 0.3: # 30%概率模拟失败 raise ConnectionError(“API暂时不可用”) return “调用成功” except ConnectionError as e: # 返回清晰的错误信息让LLM能理解并可能采取补救措施 return f“工具调用失败网络错误{str(e)}。建议用户稍后重试。” except ValueError as e: return f“工具调用失败参数错误{str(e)}。请检查输入参数。” # 方案3使用中间件进行全局错误监控 from strands.agents.middleware import Middleware class ErrorLoggingMiddleware(Middleware): async def on_tool_error(self, state: State, error: Exception): # 在这里将错误记录到日志系统如Sentry, Logstash print(f“工具执行出错: {error}, 状态: {state.get(‘session_id’, ‘unknown’)}”) # 可以选择修改state例如添加一个错误标记 state[“last_error”] str(error) # 继续传递错误或者返回一个修复后的state return state agent_with_middleware Agent(llmllm, tools[…], middleware[ErrorLoggingMiddleware()])6.2 性能优化策略随着工具变多、逻辑变复杂代理的响应速度可能成为瓶颈。工具调用的并行化默认情况下代理是顺序执行工具的。如果多个工具调用之间没有依赖关系可以考虑在代理的“规划”阶段进行优化或者设计一个能并行调用多个子工具的高级工具。LLM调用优化缓存对相同的提示词进行缓存可以大幅减少对LLM API的调用和成本。可以集成像langchain.cache这样的缓存层。流式响应对于生成内容较长的场景使用流式响应Streaming可以提升用户体验让用户尽快看到部分结果。模型选择在非核心推理步骤使用更小、更快的模型如gpt-3.5-turbo来处理简单任务将大模型如gpt-4留给需要复杂思考和规划的任务。状态序列化与持久化对于长时间运行的会话定期将State对象序列化如使用pickle或json并保存到数据库可以支持会话恢复和分布式部署。6.3 测试与评估如何确保你的代理行为符合预期单元测试和评估框架必不可少。import pytest from strands.agents import Agent, State from strands.agents.evaluation import run_agent_test # 为工具编写单元测试 def test_get_current_time_tool(): from your_tools import get_current_time result get_current_time(timezone“UTC”) assert “UTC” in result assert “:” in result # 简单的时间格式检查 # 编写针对代理端到端行为的测试 pytest.mark.asyncio async def test_agent_weather_query(): llm OpenAIChat(model“gpt-3.5-turbo”) # 测试时可用轻量级模型 agent Agent(llmllm, tools[get_weather]) test_state State({ “messages”: [{“role”: “user”, “content”: “北京天气怎么样”}] }) final_state await agent.run(statetest_state) last_message final_state[“messages”][-1][“content”] # 断言代理最终给出了包含“北京”和“天气”的回复 assert “北京” in last_message assert “天气” in last_message # 更复杂的测试可以检查是否调用了正确的工具测试心得代理的测试比普通软件更复杂因为LLM的输出具有非确定性。策略是工具测试确保每个工具函数在各种边界条件下都能正确工作。集成测试使用固定的系统提示和种子并在测试中Mock LLM的响应使其返回预定义的内容从而确保代理的执行流程符合预期。评估集构建一个包含各种典型和边缘案例用户查询的评估集定期运行监控代理整体性能的变化。7. 常见问题与排查技巧实录在实际开发中你肯定会遇到各种问题。以下是一些典型问题及其解决方案。7.1 代理陷入循环或行为异常问题表现代理不停地调用同一个工具或者输出的内容与预期完全不符。排查步骤检查系统提示System Prompt这是最常见的原因。系统提示定义了代理的“角色”和“行为准则”。确保它清晰、无歧义并明确规定了代理的目标和限制。例如加上“如果无法通过现有工具满足用户请求请直接告知用户不要尝试调用不存在的工具。”审查工具描述工具函数的文档字符串是否准确描述了功能、参数和返回值LLM完全依赖这个描述来决策。模糊的描述会导致错误的调用。启用调试日志strands-agentsSDK通常有日志功能。设置日志级别为DEBUG查看代理内部的决策过程、收到的提示词和生成的响应。import logging logging.basicConfig(levellogging.DEBUG)限制最大步数在创建Agent时设置max_steps参数如max_steps15防止因逻辑错误导致无限循环。简化复现用一个最小化的、可复现的示例来测试排除是复杂状态或记忆导致的问题。7.2 工具调用参数解析错误问题表现LLM生成了工具调用请求但SDK解析时抛出验证错误如“缺少必需参数”或“参数类型错误”。解决方案使用Pydantic Schema如前所述为工具定义args_schema。Pydantic会进行严格的类型验证并给LLM提供清晰的结构化参数定义。在工具描述中提供示例在工具文档字符串中给出参数的具体示例。例如city (str): 城市名称例如 ‘北京‘ 或 ‘New York‘。检查LLM的响应格式确保LLM被正确配置为支持“工具调用”或“函数调用”格式。对于OpenAI需要使用OpenAIChat类并确保模型版本支持此功能如gpt-3.5-turbo-1106及以上。7.3 处理长上下文与令牌限制问题表现随着对话进行State中的messages越来越长最终超过LLM的上下文窗口限制导致API调用失败或性能下降。应对策略使用BufferMemory这是最直接的方法只保留最近N条消息。总结压缩实现一个自定义的记忆类或中间件定期例如每10轮对话后将早期的对话历史总结成一段简短的文本然后替换掉旧消息。可以将总结作为一个SystemMessage插入到上下文开头。选择性记忆不是所有消息都同等重要。可以设计逻辑只保留包含工具调用结果或关键决策点的消息。分页查询对于需要参考大量背景知识的场景如知识库问答不要将所有知识塞进上下文。而是让代理学会调用一个“搜索”工具按需获取相关信息片段。7.4 如何集成自定义LLM或非OpenAI模型strands-agents的设计是开放的。虽然它提供了OpenAIChat等便捷类但集成任何符合其LLM接口的模型都是可行的。from abc import ABC, abstractmethod from strands.agents import BaseLLM, State from typing import List, Dict, Any class MyCustomLLM(BaseLLM, ABC): def __init__(self, model_endpoint: str): self.endpoint model_endpoint async def agenerate(self, messages: List[Dict], tools: List[Any] None) - Dict[str, Any]: “”” 必须实现的方法。 根据消息历史和可选工具调用你的模型返回模型原始响应。 响应格式必须包含 ‘content‘ 和可选的 ‘tool_calls‘。 “”” # 1. 将 messages 和 tools 格式化为你的模型所需的提示 prompt self._format_prompt(messages, tools) # 2. 调用你的模型API同步或异步 import aiohttp async with aiohttp.ClientSession() as session: async with session.post(self.endpoint, json{“prompt”: prompt}) as resp: raw_response await resp.json() # 3. 将你的模型响应解析为 strands-agents 期望的格式 # 假设你的模型返回 {“text”: “…”, “function_call”: {…}} result {“content”: raw_response.get(“text”, “”)} if “function_call” in raw_response: # 将你的 function_call 格式转换为标准的 tool_calls 格式 result[“tool_calls”] [self._parse_function_call(raw_response[“function_call”])] return result def _format_prompt(self, messages, tools): # … 实现你的提示词工程逻辑 pass def _parse_function_call(self, fc): # … 实现格式转换逻辑 pass # 使用自定义LLM my_llm MyCustomLLM(model_endpoint“http://localhost:8000/v1/completions”) agent Agent(llmmy_llm, tools[…])这个过程需要你深入了解你的模型API的输入输出格式并完成相应的适配工作。核心是确保agenerate方法返回的字典包含content文本回复和可选的tool_calls工具调用列表字段。