基于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的核心理念是关注点分离和可组合性。它将一个代理的典型生命周期分解为几个清晰的组件：

1. 代理（Agent）：核心决策者，根据目标、记忆和当前观察，决定下一步行动（调用哪个工具，或直接输出）。
2. 工具（Tool）：代理可以调用的具体能力，如搜索网络、查询数据库、执行代码。
3. 记忆（Memory）：存储和检索代理与用户的历史交互、工具执行结果等上下文信息。
4. 执行器（Executor）：驱动代理运行循环的引擎，负责调用模型、执行工具、更新状态。
5. 状态（State）：代理在单次运行会话中的所有上下文数据，是连接各个组件的纽带。

这种架构让你可以像搭乐高一样，独立地更换或升级每个部件。例如，你可以轻松地将记忆后端从简单的内存列表切换到Redis，或者为同一个代理更换不同的LLM提供商，而无需重写核心逻辑。

2.2 SDK的核心抽象：State与Message

这是理解该SDK的关键。与许多其他框架不同，strands-agents重度依赖一个名为State的字典状对象来贯穿整个执行流程。State包含了当前会话的所有信息：用户输入、模型响应、工具调用结果、自定义数据等。

更具体地说，State中最重要的部分之一是messages列表。这里存储着所有类型的Message对象，例如HumanMessage（用户输入）、AIMessage（AI回复）、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 git+https://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( llm=llm, tools=[get_current_time], # 将工具函数放入列表 memory=memory, system_prompt=“你是一个乐于助人的助手，可以告诉用户时间。”, # 系统指令，设定代理角色 ) # 4. 运行代理（异步方式） async def main(): # 初始化状态，包含用户的初始消息 initial_state = {“messages”: [{“role”: “user”, “content”: “现在上海是几点钟了？”}]} # 运行代理，它会自动处理思考、工具调用、响应的全过程 final_state = await agent.run(state=initial_state) # 从最终状态中提取AI的最后一条消息 last_message = final_state[“messages”][-1] print(f“助手回复: {last_message[‘content’]}”) # 执行异步函数 if __name__ == “__main__”: asyncio.run(main())

当你运行这段代码时，代理会经历以下内部流程：

1. 接收包含用户消息的初始State。
2. Agent将State中的消息历史（这里只有用户消息）和系统提示一起发送给LLM。
3. LLM分析后，认为需要调用get_current_time工具，并生成一个符合格式的工具调用请求。
4. Agent解析这个请求，执行对应的get_current_time函数，并将结果封装为一个ToolMessage，添加到State中。
5. Agent再次将更新后的State（包含用户消息、工具调用请求、工具结果）发送给LLM。
6. LLM根据工具返回的结果，生成最终的自然语言回复，作为一个AIMessage添加到State。
7. 流程结束，返回最终的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_schema=WeatherQuery) 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( llm=llm, 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_messages=20) # 只记住最近20条消息 # 适用于关注近期对话的场景，避免上下文过长导致模型性能下降或API费用过高。 # 2. RedisMemory：使用Redis进行持久化存储 # 首先安装 redis 包: pip install redis redis_memory = RedisMemory( redis_url=“redis://localhost:6379/0”, session_ttl=3600, # 会话数据过期时间（秒） ) # 适用于多实例部署、需要会话持久化（如用户下次回来继续对话）的场景。 # 使用记忆 agent_with_memory = Agent(llm=llm, tools=[…], memory=redis_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(bins=20, 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_messages=30) agent = Agent( llm=llm, tools=[load_csv_data, describe_data, plot_histogram, show_last_plot], memory=memory, 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(state=initial_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( llm=llm, tools=[…], max_steps=10, # 限制单次运行的最大步骤（思考+工具调用）数，防止死循环 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(llm=llm, tools=[…], middleware=[ErrorLoggingMiddleware()])

6.2 性能优化策略

随着工具变多、逻辑变复杂，代理的响应速度可能成为瓶颈。

1. 工具调用的并行化：默认情况下，代理是顺序执行工具的。如果多个工具调用之间没有依赖关系，可以考虑在代理的“规划”阶段进行优化，或者设计一个能并行调用多个子工具的高级工具。

2. LLM调用优化：

   • 缓存：对相同的提示词进行缓存可以大幅减少对LLM API的调用和成本。可以集成像langchain.cache这样的缓存层。
   • 流式响应：对于生成内容较长的场景，使用流式响应（Streaming）可以提升用户体验，让用户尽快看到部分结果。
   • 模型选择：在非核心推理步骤，使用更小、更快的模型（如gpt-3.5-turbo）来处理简单任务，将大模型（如gpt-4）留给需要复杂思考和规划的任务。

3. 状态序列化与持久化：对于长时间运行的会话，定期将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(llm=llm, tools=[get_weather]) test_state = State({ “messages”: [{“role”: “user”, “content”: “北京天气怎么样？”}] }) final_state = await agent.run(state=test_state) last_message = final_state[“messages”][-1][“content”] # 断言代理最终给出了包含“北京”和“天气”的回复 assert “北京” in last_message assert “天气” in last_message # 更复杂的测试可以检查是否调用了正确的工具

测试心得：代理的测试比普通软件更复杂，因为LLM的输出具有非确定性。策略是：

• 工具测试：确保每个工具函数在各种边界条件下都能正确工作。
• 集成测试：使用固定的系统提示和种子，并在测试中Mock LLM的响应，使其返回预定义的内容，从而确保代理的执行流程符合预期。
• 评估集：构建一个包含各种典型和边缘案例用户查询的评估集，定期运行，监控代理整体性能的变化。

7. 常见问题与排查技巧实录

在实际开发中，你肯定会遇到各种问题。以下是一些典型问题及其解决方案。

7.1 代理陷入循环或行为异常

问题表现：代理不停地调用同一个工具，或者输出的内容与预期完全不符。

排查步骤：

1. 检查系统提示（System Prompt）：这是最常见的原因。系统提示定义了代理的“角色”和“行为准则”。确保它清晰、无歧义，并明确规定了代理的目标和限制。例如，加上“如果无法通过现有工具满足用户请求，请直接告知用户，不要尝试调用不存在的工具。”
2. 审查工具描述：工具函数的文档字符串是否准确描述了功能、参数和返回值？LLM完全依赖这个描述来决策。模糊的描述会导致错误的调用。
3. 启用调试日志：strands-agentsSDK通常有日志功能。设置日志级别为DEBUG，查看代理内部的决策过程、收到的提示词和生成的响应。

   import logging logging.basicConfig(level=logging.DEBUG)

4. 限制最大步数：在创建Agent时设置max_steps参数（如max_steps=15），防止因逻辑错误导致无限循环。
5. 简化复现：用一个最小化的、可复现的示例来测试，排除是复杂状态或记忆导致的问题。

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调用失败或性能下降。

应对策略：

1. 使用BufferMemory：这是最直接的方法，只保留最近N条消息。
2. 总结压缩：实现一个自定义的记忆类或中间件，定期（例如每10轮对话后）将早期的对话历史总结成一段简短的文本，然后替换掉旧消息。可以将总结作为一个SystemMessage插入到上下文开头。
3. 选择性记忆：不是所有消息都同等重要。可以设计逻辑，只保留包含工具调用结果或关键决策点的消息。
4. 分页查询：对于需要参考大量背景知识的场景（如知识库问答），不要将所有知识塞进上下文。而是让代理学会调用一个“搜索”工具，按需获取相关信息片段。

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(llm=my_llm, tools=[…])

这个过程需要你深入了解你的模型API的输入输出格式，并完成相应的适配工作。核心是确保agenerate方法返回的字典包含content（文本回复）和可选的tool_calls（工具调用列表）字段。