1. 项目概述从零理解一个AI智能体技能库最近在折腾AI智能体开发的朋友可能都绕不开一个核心问题如何让一个AI模型比如GPT-4、Claude或者开源的Llama不仅能和你聊天还能真正“动手”帮你做事比如让它查查天气、发封邮件、或者从网上抓取点信息。这背后需要的就是所谓的“技能”或“工具”。今天要聊的这个项目agentskill-sh/ags就是一个专门为AI智能体打造的、开源的技能库。你可以把它想象成一个“瑞士军刀”的刀架它本身不提供具体的刀片比如开瓶器、小刀但它定义了一套标准让你可以轻松地把各种功能刀片插上去交给你的AI智能体使用。我第一次接触这类项目是因为在构建一个自动化客服助手时需要让AI能根据用户问题自动查询订单状态。当时面临的选择是自己从头写一套调用后端API的代码并费力地让AI理解如何调用还是找一个现成的框架来管理这些“技能”。ags走的是后一条路它试图解决一个非常实际的问题技能管理的标准化和易用性。对于开发者而言这意味着你不用再为每个智能体项目重复发明轮子去设计技能的描述格式、调用协议和结果处理逻辑。ags提供了一套统一的接口无论是调用一个简单的计算器还是集成一个复杂的企业内部系统都可以用同一种方式“告诉”AI并由AI以同一种方式触发。这个项目的价值尤其体现在当前“智能体即应用”的趋势下。当AI不再仅仅是对话界面而要成为工作流的核心调度者时一个可靠、可扩展的技能底座就至关重要。ags瞄准的正是这个生态位。它不绑定任何特定的AI模型或运行时框架这意味着你可以把它接入LangChain、AutoGen、或者你自己写的智能体循环中。它的核心思想是“声明式”的技能定义你通过代码主要是Python清晰地描述一个技能能做什么、需要什么参数、会返回什么结果然后ags负责将其打包成AI模型能理解的格式比如OpenAI的Function Calling格式并处理实际的调用执行。接下来我们就深入拆解一下它的设计思路和到底该怎么用。2. 核心设计理念与架构拆解2.1 为什么需要专门的技能库在深入ags的代码之前我们先想想如果没有它我们通常怎么给AI加功能一个典型的做法是在提示词里用自然语言描述“你可以调用get_weather函数来查询天气它需要一个city参数...”。这种方法在小规模时勉强可行但问题很多描述容易歧义AI可能误解参数格式每增加一个技能提示词就膨胀一段影响上下文窗口技能的逻辑和执行代码混杂在智能体主循环里难以维护和测试。另一种进阶做法是使用模型原生的工具调用功能比如OpenAI的Function Calling。你需要按照其规定的JSON Schema格式定义每个函数的名称、描述和参数。这解决了格式标准化的问题但管理起来依然繁琐你需要手动维护一堆JSON定义确保它们与后端实现同步并且处理函数注册和路由的逻辑。ags的出现就是为了抽象掉这些底层细节。它的设计目标很明确标准化提供一套统一的、模型无关的技能描述规范。声明式用Python代码而非手工JSON来定义技能利用IDE的自动补全和类型检查。解耦将技能的定义、描述给AI看和执行实际运行清晰分离。可发现让智能体能轻松地知道自己有哪些技能可用。2.2 核心架构Skill, Tool, Agent 的三层关系ags的架构围绕几个核心概念展开理解它们就理解了整个项目Skill技能这是最核心的抽象。一个Skill代表一个具体的、可执行的能力。它包含三部分信息定义技能的元数据如名称、描述、所需的输入参数及其类型、返回值的类型。这部分是给AI模型“看”的用于让AI理解何时以及如何调用该技能。实现具体的执行函数一个Python可调用对象。当AI决定调用该技能时ags会运行这个函数。配置一些运行时参数比如API密钥、服务地址等这些通常不暴露给AI但为技能执行所必需。Tool在ags的语境中Tool通常是Skill适配成特定AI模型如OpenAI所需格式后的产物。例如一个WeatherSkill会被转换成符合OpenAI Function Calling规范的tool字典。ags内部帮你完成了这个转换你通常不需要直接操作Tool。Agent智能体。ags本身不实现完整的智能体逻辑如思考、记忆、规划它专注于为智能体提供“技能装备”。你的智能体系统无论用什么框架会从ags中加载所需的技能获取这些技能的“工具描述”并注入给AI模型然后在AI模型返回工具调用请求时委托ags来执行对应的技能。这种分层带来了极大的灵活性。你可以独立开发、测试每一个Skill就像开发一个独立的微服务。然后像搭积木一样根据不同智能体的职责组合不同的技能集。例如一个客服机器人可能只需要QueryOrderSkill和SubmitTicketSkill而一个个人办公助手则需要SendEmailSkill、ScheduleMeetingSkill和WebSearchSkill。2.3 关键技术实现解析ags是如何实现上述设计的我们来看几个关键技术点基于Pydantic的类型驱动定义ags重度依赖Pydantic这个库。Pydantic利用Python类型注解type hints来进行数据验证和设置管理。在定义技能参数时你可以使用str,int,Literal[option1, option2]等丰富的类型。ags会利用这些类型注解自动生成准确、结构化的JSON Schema这比手动写JSON描述要可靠和高效得多并且能享受到静态类型检查的好处。from pydantic import BaseModel, Field from ags import skill, SkillParam # 定义输入参数的模型 class WeatherQueryInput(BaseModel): city: str Field(descriptionThe city name, e.g., Beijing) country_code: str Field(CN, descriptionISO country code, default is CN) # 使用装饰器定义技能 skill( nameget_weather, descriptionGet the current weather for a given city., input_modelWeatherQueryInput ) async def get_weather_skill(query: WeatherQueryInput) - str: # 这里是技能的实际实现 # 例如调用一个天气API return fThe weather in {query.city} is sunny.上面这段代码就完整定义了一个技能。skill装饰器收集了所有元数据而函数体是执行逻辑。ags会自动从WeatherQueryInput这个Pydantic模型生成对应的参数Schema。异步优先Async-first现代AI应用和网络IO密集型操作如调用API普遍采用异步编程来提高并发性能。ags的技能执行函数默认支持async def这意味着你可以在技能内部方便地使用aiohttp等异步库进行网络请求而不会阻塞整个智能体的事件循环。灵活的配置管理技能可能需要密钥、端点URL等配置。ags通过Pydantic的Settings管理理念允许你将配置注入到技能中而不是硬编码在函数里。这既保证了安全性密钥不进入代码仓库也提高了技能的可移植性。技能组合与路由ags提供了将多个技能聚合在一起的能力。你可以创建一个SkillRegistry技能注册表来管理一整套技能。当智能体传来一个工具调用请求时ags能根据工具名称快速路由到正确的技能并执行。这个注册表也可以方便地导入导出实现技能的共享。3. 从零开始定义与实现你的第一个技能理论说了不少现在我们来动手创建一个实实在在的技能。假设我们要为智能体添加一个“查询当前时间”的技能。这个技能很简单但它能完整走通ags的工作流程。3.1 环境准备与安装首先确保你有一个Python环境3.8以上。然后安装ags。由于它是一个较新的开源项目通常直接从GitHub安装最新版本。# 推荐使用uv或pip进行安装 pip install ags[all] # 安装ags及其常用依赖 # 或者从源码安装 pip install githttps://github.com/agentskill-sh/ags.git注意开源项目迭代可能较快API可能会有变动。如果遇到问题查看项目README或pyproject.toml文件中的依赖说明是很好的习惯。生产环境建议锁定版本。3.2 技能定义三步走第一步设计输入输出思考技能需要什么信息输入以及会返回什么信息输出。对于“查询时间”我们可能希望它能根据时区来返回时间。所以输入可以是一个可选的时区参数比如Asia/Shanghai输出是一个表示时间的字符串。第二步编写Pydantic模型这是将思考规范化的关键一步。我们为输入创建一个模型。from pydantic import BaseModel, Field from typing import Optional import pytz # 需要安装 pytz: pip install pytz from datetime import datetime class GetTimeInput(BaseModel): timezone: Optional[str] Field( defaultUTC, descriptionThe IANA timezone name, e.g., Asia/Shanghai, America/New_York. Default is UTC. )这里我们定义了一个GetTimeInput类它有一个timezone字段类型是可选字符串默认值是UTC并附上了描述。这个描述至关重要AI模型会阅读它来理解这个参数的意义。第三步用装饰器创建技能现在我们实现技能逻辑并用skill装饰器将其包装。from ags import skill import pytz from datetime import datetime skill( nameget_current_time, descriptionGet the current date and time for a specified timezone., input_modelGetTimeInput ) async def get_current_time(input: GetTimeInput) - str: 获取指定时区的当前时间。 tz_str input.timezone try: # 获取时区对象 tz pytz.timezone(tz_str) except pytz.exceptions.UnknownTimeZoneError: # 如果时区无效回退到UTC并在结果中说明 tz pytz.UTC current_time_utc datetime.now(pytz.UTC) return fUnknown timezone {tz_str}. Current time in UTC is: {current_time_utc.strftime(%Y-%m-%d %H:%M:%S %Z)} # 获取该时区的当前时间 current_time datetime.now(tz) # 格式化为易读的字符串 formatted_time current_time.strftime(%Y-%m-%d %H:%M:%S %Z) return fThe current time in {tz_str} is: {formatted_time}代码解读skill装饰器我们提供了技能的名称(name)、给AI看的描述(description)以及输入模型(input_model)。函数get_current_time这是一个异步函数接收一个GetTimeInput实例。函数内部尝试用pytz.timezone解析用户传入的时区字符串。如果时区无效则捕获异常使用UTC时区并返回提示信息。如果有效则获取该时区的当前时间并格式化。最后返回一个字符串结果。错误处理在技能实现中考虑边界情况和错误处理非常重要。这里我们对无效时区做了优雅降级而不是抛出异常导致智能体崩溃。在实际技能中如调用外部API更需要进行完善的错误处理和重试逻辑。3.3 将技能“装备”给智能体技能定义好了但孤零零的一个函数没法用。我们需要创建一个技能集合并暴露给智能体框架。这里以最简单的交互为例演示如何获取技能的“工具描述”并手动模拟一次调用。from ags import SkillSet # 1. 创建技能集合 skillset SkillSet() # 2. 将技能添加到集合中 skillset.add_skill(get_current_time) # 3. 获取该技能对应的“工具”定义例如给OpenAI的格式 tools_for_ai skillset.to_openai_tools() print(Tool definition for AI:) import json print(json.dumps(tools_for_ai, indent2)) # 输出大致如下 # [ # { # type: function, # function: { # name: get_current_time, # description: Get the current date and time for a specified timezone., # parameters: { # type: object, # properties: { # timezone: { # type: string, # description: The IANA timezone name, e.g., Asia/Shanghai, America/New_York. Default is UTC. # } # }, # required: [] # } # } # } # ]这个tools_for_ai列表就是你可以直接填入OpenAI API调用中tools参数的内容。AI模型在看到这个定义后就学会了在合适的时候调用get_current_time。模拟一次AI调用与技能执行 假设AI模型经过思考决定调用这个技能并生成了调用参数。# 模拟AI返回的工具调用请求 ai_tool_call { name: get_current_time, arguments: json.dumps({timezone: Asia/Shanghai}) # AI可能会生成这个JSON字符串 } # 4. 技能集合执行工具调用 import asyncio async def run_demo(): result await skillset.execute_tool_call( tool_nameai_tool_call[name], tool_argumentsai_tool_call[arguments] ) print(Skill execution result:, result) asyncio.run(run_demo()) # 输出: Skill execution result: The current time in Asia/Shanghai is: 2023-10-27 15:30:00 CSTskillset.execute_tool_call方法完成了路由和执行的脏活累活它根据tool_name找到对应的技能解析tool_argumentsJSON字符串为Pydantic模型然后执行我们定义的get_current_time函数最后返回结果。4. 进阶实战构建一个实用的网络搜索技能单一技能威力有限现在我们挑战一个更实用、也更复杂的技能网络搜索。这涉及到调用外部API如Serper、Google Search API、处理API密钥、解析返回结果等。通过这个例子你将掌握ags处理配置、异步IO和复杂返回类型的技巧。4.1 设计技能与配置管理一个网络搜索技能需要输入搜索查询词query可能还有结果数量num_results。输出结构化的搜索结果列表包含标题、链接、摘要等。配置API密钥、搜索端点URL。这些绝不能硬编码而应通过配置注入。首先定义输入模型和输出模型。输出模型用于告诉AI返回值的结构。from pydantic import BaseModel, Field, HttpUrl from typing import List class WebSearchInput(BaseModel): query: str Field(descriptionThe search query string.) num_results: int Field(default5, ge1, le10, descriptionNumber of search results to return, between 1 and 10.) class SearchResult(BaseModel): title: str Field(descriptionThe title of the search result.) link: HttpUrl Field(descriptionThe URL of the search result.) snippet: str Field(descriptionA brief summary or snippet of the result.) class WebSearchOutput(BaseModel): results: List[SearchResult] Field(descriptionList of search results.) total_estimated: int Field(descriptionEstimated total number of results found.)这里我们定义了嵌套的Pydantic模型。HttpUrl类型会自动验证字符串是否为有效的URL。ge和le用于限制num_results的范围。接下来处理配置。我们创建一个专门的配置类并从环境变量读取敏感信息。from pydantic import SecretStr from pydantic_settings import BaseSettings class SearchConfig(BaseSettings): 网络搜索技能的配置 serper_api_key: SecretStr # 使用SecretStr隐藏密钥 search_endpoint: str https://google.serper.dev/search # 默认端点 class Config: env_prefix SEARCH_ # 环境变量前缀例如 SEARCH_SERPER_API_KEY使用pydantic-settings可以方便地从环境变量、.env文件等加载配置。SecretStr类型在打印或日志中会自动显示为********增加了安全性。4.2 实现技能逻辑集成外部API现在实现技能函数。我们将使用aiohttp进行异步HTTP请求。import aiohttp from ags import skill, SkillParam from typing import Annotated skill( nameweb_search, descriptionPerform a web search and return structured results., input_modelWebSearchInput, output_modelWebSearchOutput # 声明输出模型有助于AI理解返回结构 ) async def web_search_skill( input: WebSearchInput, config: Annotated[SearchConfig, SkillParam(scopeskill)] # 通过依赖注入获取配置 ) - WebSearchOutput: 使用Serper API执行网络搜索。 headers { X-API-KEY: config.serper_api_key.get_secret_value(), Content-Type: application/json } payload { q: input.query, num: input.num_results } async with aiohttp.ClientSession() as session: try: async with session.post(config.search_endpoint, jsonpayload, headersheaders) as response: response.raise_for_status() # 如果状态码不是2xx抛出异常 data await response.json() except aiohttp.ClientError as e: # 网络或客户端错误 return WebSearchOutput( results[], total_estimated0, error_messagefNetwork error during search: {str(e)} ) except Exception as e: # 其他未知错误 return WebSearchOutput( results[], total_estimated0, error_messagefAn unexpected error occurred: {str(e)} ) # 解析Serper API的返回结果 (实际结构需参考API文档) # 这里是一个示例解析逻辑 organic_results data.get(organic, []) search_results [] for item in organic_results[:input.num_results]: search_results.append( SearchResult( titleitem.get(title, No title), linkitem.get(link, ), snippetitem.get(snippet, ) ) ) return WebSearchOutput( resultssearch_results, total_estimateddata.get(searchInformation, {}).get(totalResults, 0) )关键点解析依赖注入config: Annotated[SearchConfig, SkillParam(scopeskill)]这行是ags的一个强大特性。它告诉ags当执行这个技能时请自动为我提供一个SearchConfig的实例。SkillParam(scopeskill)表示这个配置是在技能级别共享的。ags会在背后管理配置的初始化和注入。异步HTTP客户端使用aiohttp.ClientSession进行异步请求避免阻塞。全面的错误处理网络请求可能失败API可能返回错误。我们用try...except捕获aiohttp.ClientError和其他异常并返回一个包含错误信息的WebSearchOutput而不是让异常向上传播导致智能体崩溃。这是生产级技能的必要考虑。输出模型最终我们返回一个WebSearchOutput实例其中包含了结构化的结果列表。AI在收到这个结果后可以清晰地引用其中的title和link。4.3 配置与运行在运行前需要设置环境变量。export SEARCH_SERPER_API_KEYyour_actual_api_key_here然后在你的智能体主程序中import asyncio from ags import SkillSet async def main(): # 技能集会自动从环境变量读取配置并注入 skillset SkillSet() skillset.add_skill(web_search_skill) # 获取工具定义 tools skillset.to_openai_tools() # ... 将tools提供给AI模型 ... # 模拟AI调用搜索 ai_call { name: web_search, arguments: json.dumps({query: latest developments in AI agents, num_results: 3}) } result await skillset.execute_tool_call(**ai_call) print(fSearch completed. Found {result.total_estimated} results.) for res in result.results: print(f- {res.title}: {res.link}) if __name__ __main__: asyncio.run(main())5. 工程化实践技能管理、测试与最佳实践当技能数量增多后如何有效地组织、测试和集成它们就成为了工程上的挑战。这一部分分享一些在实战中积累的经验。5.1 技能的组织与发现不建议把所有技能都写在一个文件里。一个好的实践是按领域或功能模块组织技能。my_agent_project/ ├── skills/ │ ├── __init__.py │ ├── base.py # 基础配置、公共工具 │ ├── web.py # 网络相关技能搜索、爬虫 │ ├── productivity.py # 效率工具日历、邮件 │ └── data.py # 数据处理技能 ├── config/ │ └── settings.py # 统一配置管理 ├── agent_main.py # 智能体主程序 └── requirements.txt在skills/__init__.py中可以集中导出所有技能方便主程序导入。# skills/__init__.py from .web import web_search_skill, fetch_webpage_skill from .productivity import send_email_skill, create_calendar_event_skill __all__ [ web_search_skill, fetch_webpage_skill, send_email_skill, create_calendar_event_skill, ]主程序只需要from skills import *然后添加到SkillSet中。对于更复杂的项目可以考虑使用SkillRegistry类它提供了更丰富的技能管理功能如按标签过滤、动态加载等。5.2 技能的单元测试技能也是代码必须可测试。由于技能通常是异步函数且可能依赖外部服务测试时需要一些技巧。1. 模拟Mock外部依赖对于网络请求、数据库访问等使用unittest.mock或pytest-mock来模拟。import pytest from unittest.mock import AsyncMock, patch from skills.web import web_search_skill, WebSearchInput pytest.mark.asyncio async def test_web_search_success(mocker): # 1. Mock配置 mock_config mocker.MagicMock() mock_config.serper_api_key.get_secret_value.return_value fake_key mock_config.search_endpoint https://fake.endpoint # 2. Mock aiohttp的响应 fake_json { organic: [ {title: Test Result 1, link: https://example.com/1, snippet: Snippet 1}, {title: Test Result 2, link: https://example.com/2, snippet: Snippet 2}, ], searchInformation: {totalResults: 100} } mock_response AsyncMock() mock_response.json AsyncMock(return_valuefake_json) mock_response.raise_for_status AsyncMock() mock_session mocker.patch(aiohttp.ClientSession) mock_session_instance AsyncMock() mock_session.return_value.__aenter__.return_value mock_session_instance mock_session_instance.post.return_value.__aenter__.return_value mock_response # 3. 准备输入 search_input WebSearchInput(querytest query, num_results2) # 4. 执行技能 (需要手动注入被mock的config) # 注意这里需要根据ags的具体测试方式调整可能需要使用ags的测试工具 # 假设我们直接调用函数在知道如何注入config的情况下 result await web_search_skill(inputsearch_input, configmock_config) # 5. 断言 assert len(result.results) 2 assert result.results[0].title Test Result 1 assert result.total_estimated 100 # 验证mock是否被以预期方式调用 mock_session_instance.post.assert_called_once_with( https://fake.endpoint, json{q: test query, num: 2}, headers{X-API-KEY: fake_key, Content-Type: application/json} )2. 测试技能的工具描述生成确保技能生成的JSON Schema符合预期。def test_skill_tool_schema(): tools web_search_skill.to_tools() # 假设skill对象有to_tools方法 tool_def tools[0] assert tool_def[function][name] web_search assert query in tool_def[function][parameters][properties] assert tool_def[function][parameters][properties][num_results][default] 55.3 性能、安全与错误处理最佳实践超时与重试所有涉及网络IO的技能都必须设置超时并考虑重试逻辑。可以在技能内部实现或者使用像tenacity这样的重试库。aiohttp本身支持超时设置。timeout aiohttp.ClientTimeout(total10) # 10秒总超时 async with session.post(url, jsonpayload, timeouttimeout) as response: ...速率限制如果调用的是有速率限制的第三方API如OpenAI、Serper需要在技能或调用层面实现限流避免触发限制。可以使用asyncio.Semaphore或专门的限流库。输入验证与净化Pydantic提供了强大的输入验证但有时需要额外处理。例如对于接收URL并下载内容的技能需要验证URL的协议只允许http/https防止SSRF攻击。对于接收用户输入并用于数据库查询的技能要警惕SQL注入。敏感信息处理永远不要在技能代码、日志或返回给AI的结果中硬编码或泄露API密钥、密码。使用SecretStr、SecretBytes等类型。配置从环境变量或安全的配置管理服务如Vault中读取。在日志中对敏感参数进行脱敏处理。错误的分类与反馈技能执行失败时返回给AI的错误信息应具有指导性但又不能泄露内部细节。可以定义一套错误码和用户友好的错误信息。class SkillError(Exception): 技能基础异常 def __init__(self, message: str, user_friendly_msg: str, error_code: str): super().__init__(message) self.user_friendly_msg user_friendly_msg self.error_code error_code class ExternalServiceError(SkillError): 外部服务错误 pass # 在技能中 try: await call_external_api() except aiohttp.ClientResponseError as e: if e.status 429: raise ExternalServiceError( messagefAPI rate limited: {e}, user_friendly_msgThe search service is currently busy. Please try again in a moment., error_codeRATE_LIMIT )然后在智能体层面可以捕获这些自定义异常并将user_friendly_msg传递给AI或最终用户。6. 集成到智能体框架以LangChain为例ags技能最终要服务于智能体。这里以流行的LangChain框架为例展示如何无缝集成。假设我们已经有了一个SkillSet对象my_skillset里面包含了web_search_skill和get_current_time_skill。第一步将ags技能转换为LangChain ToolLangChain有自己的Tool抽象。我们需要一个简单的适配器。from langchain.tools import BaseTool from langchain.callbacks.manager import CallbackManagerForToolRun from typing import Optional, Type from pydantic import BaseModel, Field class AGSAdapter(BaseTool): 适配器将ags技能包装成LangChain Tool skill_name: str skill_set: SkillSet # 持有ags的技能集合 def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] None) - str: # LangChain的Tool默认同步调用但ags技能是异步的。 # 这里需要在异步上下文中运行或者使用同步执行方法如果ags提供。 # 假设skill_set有同步执行方法例如通过asyncio.run在内部处理 # 注意这只是一个示例实际集成可能需要处理异步到同步的转换。 result asyncio.run(self.skill_set.execute_tool_call(self.skill_name, query)) return str(result) # 将结果转为字符串 async def _arun(self, query: str, run_manager: Optional[CallbackManagerForToolRun] None) - str: # 异步版本 result await self.skill_set.execute_tool_call(self.skill_name, query) return str(result) # 为每个技能创建适配器 search_tool AGSAdapter( nameweb_search, descriptionUseful for searching the web for current information., skill_nameweb_search, skill_setmy_skillset ) time_tool AGSAdapter( nameget_current_time, descriptionUseful for getting the current time in any timezone., skill_nameget_current_time, skill_setmy_skillset )第二步在LangChain Agent中使用现在可以将这些Tool提供给LangChain的Agent。from langchain.agents import initialize_agent, AgentType from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory llm ChatOpenAI(modelgpt-4, temperature0) memory ConversationBufferMemory(memory_keychat_history) tools [search_tool, time_tool] # 我们的ags技能工具 agent initialize_agent( tools, llm, agentAgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 适合对话的Agent类型 memorymemory, verboseTrue ) # 运行Agent response agent.run(Whats the current time in Tokyo, and then search for the latest news about it?) print(response)在这个流程中LangChain Agent负责与用户对话、规划思考过程ReAct模式当它认为需要搜索或查时间时会调用我们提供的Tool而这个Tool背后实际执行的是ags技能。更优雅的集成上述适配器方法比较直接。ags项目未来可能会提供官方的LangChain集成或者社区会有更成熟的方案。核心思想是ags负责技能的标准化定义和执行而智能体框架负责高层的推理、规划和工具调用调度两者各司其职通过一个轻量级的适配层连接。7. 常见问题、排查与性能调优在实际开发和运维中你肯定会遇到各种问题。这里记录一些典型场景和解决思路。7.1 技能执行失败排查清单问题现象可能原因排查步骤AI模型不调用技能1. 技能描述不清晰。2. 技能名称/参数名与AI训练数据不匹配。3. 智能体配置未正确加载工具。1. 检查skill中的description是否准确描述了技能功能和适用场景。2. 检查生成的Tool定义to_openai_tools()输出确保JSON Schema格式正确。3. 在智能体初始化后打印其可用工具列表确认。技能被调用但参数解析错误1. AI生成的参数JSON格式错误。2. Pydantic模型验证失败类型不符、必填项缺失。1. 在skillset.execute_tool_call前后打印tool_arguments检查是否为合法JSON。2. 查看Pydantic抛出的验证错误详情调整模型定义或给AI更明确的参数描述。技能执行超时或挂起1. 外部API响应慢或无响应。2. 技能函数内有阻塞操作或死循环。3. 异步事件循环被阻塞。1. 为网络请求添加超时如aiohttp.ClientTimeout。2. 检查技能代码确保IO操作都是异步的使用await。3. 考虑在技能级别或调用侧设置整体超时。返回结果AI无法理解1. 返回类型过于复杂或非结构化。2. 返回了AI无法解析的二进制数据或特殊对象。1. 尽量返回字符串或简单的字典/列表。使用output_model明确输出结构。2. 对于复杂对象在技能内将其转换为文本描述。例如将Pandas DataFrame转为Markdown表格字符串。配置注入失败1. 环境变量未设置或名称错误。2.SkillParam依赖注入作用域配置错误。1. 确认环境变量前缀如SEARCH_和变量名正确。2. 在技能函数内打印或日志记录传入的config对象检查其属性是否为预期值。3. 查阅ags关于依赖注入和作用域的文档。7.2 性能优化要点技能懒加载如果技能数量很多但一次对话只用其中几个可以考虑懒加载机制。即不在启动时初始化所有技能特别是那些依赖重型资源或慢速连接的技能而是在第一次被调用时再初始化。这可以通过自定义Skill类或使用工厂模式实现。连接池与缓存对于频繁调用外部API的技能如数据库查询、向量检索使用连接池如aiohttp.ClientSession重用和适当的缓存策略可以极大提升性能。例如对某些只读的、更新不频繁的查询结果缓存几分钟。from functools import lru_cache import asyncio lru_cache(maxsize128) def _sync_expensive_call(param): # 同步的昂贵调用 pass async def expensive_skill(param: str): # 在异步函数中运行同步缓存调用使用线程池执行器避免阻塞事件循环 loop asyncio.get_event_loop() result await loop.run_in_executor(None, _sync_expensive_call, param) return result注意缓存需要设计合理的失效策略避免返回过期数据。异步并发执行如果一个智能体需要并行执行多个独立技能例如同时查询天气和新闻可以利用asyncio.gather。但需要注意这需要智能体框架本身支持并行工具调用目前大多数框架是顺序执行的。监控与日志为关键技能添加详细的日志记录输入、输出、执行时间。这有助于性能分析和故障排查。可以使用结构化日志如structlog方便后续聚合分析。7.3 设计模式技能编排与组合有时一个复杂任务需要按顺序调用多个技能。虽然这通常由智能体的“大脑”来规划但也可以将固定的工作流封装成一个“复合技能”。skill(nameresearch_and_summarize, descriptionResearch a topic and provide a summary.) async def research_skill(topic: str) - str: 1. 搜索主题 2. 抓取前3个结果的内容 3. 调用LLM进行总结 # 1. 搜索 search_results await web_search_skill(WebSearchInput(querytopic, num_results3)) if not search_results.results: return No relevant information found. # 2. 并发抓取网页内容 (假设有fetch_webpage_skill) fetch_tasks [] for result in search_results.results: task fetch_webpage_skill(FetchWebpageInput(urlstr(result.link))) fetch_tasks.append(task) webpage_contents await asyncio.gather(*fetch_tasks, return_exceptionsTrue) # 处理抓取成功和失败的内容... # 3. 组合内容并总结 (假设有summarize_text_skill) combined_text \n---\n.join([c for c in webpage_contents if isinstance(c, str)]) summary await summarize_text_skill(SummarizeInput(textcombined_text, max_length500)) return summary这种模式将多个原子技能组合成一个更高级别的技能对AI来说是一个“宏操作”简化了智能体的规划复杂度。但要注意控制复合技能的复杂度和执行时间。 AI智能体技能库开发实战：从Pydantic定义到LangChain集成
1. 项目概述从零理解一个AI智能体技能库最近在折腾AI智能体开发的朋友可能都绕不开一个核心问题如何让一个AI模型比如GPT-4、Claude或者开源的Llama不仅能和你聊天还能真正“动手”帮你做事比如让它查查天气、发封邮件、或者从网上抓取点信息。这背后需要的就是所谓的“技能”或“工具”。今天要聊的这个项目agentskill-sh/ags就是一个专门为AI智能体打造的、开源的技能库。你可以把它想象成一个“瑞士军刀”的刀架它本身不提供具体的刀片比如开瓶器、小刀但它定义了一套标准让你可以轻松地把各种功能刀片插上去交给你的AI智能体使用。我第一次接触这类项目是因为在构建一个自动化客服助手时需要让AI能根据用户问题自动查询订单状态。当时面临的选择是自己从头写一套调用后端API的代码并费力地让AI理解如何调用还是找一个现成的框架来管理这些“技能”。ags走的是后一条路它试图解决一个非常实际的问题技能管理的标准化和易用性。对于开发者而言这意味着你不用再为每个智能体项目重复发明轮子去设计技能的描述格式、调用协议和结果处理逻辑。ags提供了一套统一的接口无论是调用一个简单的计算器还是集成一个复杂的企业内部系统都可以用同一种方式“告诉”AI并由AI以同一种方式触发。这个项目的价值尤其体现在当前“智能体即应用”的趋势下。当AI不再仅仅是对话界面而要成为工作流的核心调度者时一个可靠、可扩展的技能底座就至关重要。ags瞄准的正是这个生态位。它不绑定任何特定的AI模型或运行时框架这意味着你可以把它接入LangChain、AutoGen、或者你自己写的智能体循环中。它的核心思想是“声明式”的技能定义你通过代码主要是Python清晰地描述一个技能能做什么、需要什么参数、会返回什么结果然后ags负责将其打包成AI模型能理解的格式比如OpenAI的Function Calling格式并处理实际的调用执行。接下来我们就深入拆解一下它的设计思路和到底该怎么用。2. 核心设计理念与架构拆解2.1 为什么需要专门的技能库在深入ags的代码之前我们先想想如果没有它我们通常怎么给AI加功能一个典型的做法是在提示词里用自然语言描述“你可以调用get_weather函数来查询天气它需要一个city参数...”。这种方法在小规模时勉强可行但问题很多描述容易歧义AI可能误解参数格式每增加一个技能提示词就膨胀一段影响上下文窗口技能的逻辑和执行代码混杂在智能体主循环里难以维护和测试。另一种进阶做法是使用模型原生的工具调用功能比如OpenAI的Function Calling。你需要按照其规定的JSON Schema格式定义每个函数的名称、描述和参数。这解决了格式标准化的问题但管理起来依然繁琐你需要手动维护一堆JSON定义确保它们与后端实现同步并且处理函数注册和路由的逻辑。ags的出现就是为了抽象掉这些底层细节。它的设计目标很明确标准化提供一套统一的、模型无关的技能描述规范。声明式用Python代码而非手工JSON来定义技能利用IDE的自动补全和类型检查。解耦将技能的定义、描述给AI看和执行实际运行清晰分离。可发现让智能体能轻松地知道自己有哪些技能可用。2.2 核心架构Skill, Tool, Agent 的三层关系ags的架构围绕几个核心概念展开理解它们就理解了整个项目Skill技能这是最核心的抽象。一个Skill代表一个具体的、可执行的能力。它包含三部分信息定义技能的元数据如名称、描述、所需的输入参数及其类型、返回值的类型。这部分是给AI模型“看”的用于让AI理解何时以及如何调用该技能。实现具体的执行函数一个Python可调用对象。当AI决定调用该技能时ags会运行这个函数。配置一些运行时参数比如API密钥、服务地址等这些通常不暴露给AI但为技能执行所必需。Tool在ags的语境中Tool通常是Skill适配成特定AI模型如OpenAI所需格式后的产物。例如一个WeatherSkill会被转换成符合OpenAI Function Calling规范的tool字典。ags内部帮你完成了这个转换你通常不需要直接操作Tool。Agent智能体。ags本身不实现完整的智能体逻辑如思考、记忆、规划它专注于为智能体提供“技能装备”。你的智能体系统无论用什么框架会从ags中加载所需的技能获取这些技能的“工具描述”并注入给AI模型然后在AI模型返回工具调用请求时委托ags来执行对应的技能。这种分层带来了极大的灵活性。你可以独立开发、测试每一个Skill就像开发一个独立的微服务。然后像搭积木一样根据不同智能体的职责组合不同的技能集。例如一个客服机器人可能只需要QueryOrderSkill和SubmitTicketSkill而一个个人办公助手则需要SendEmailSkill、ScheduleMeetingSkill和WebSearchSkill。2.3 关键技术实现解析ags是如何实现上述设计的我们来看几个关键技术点基于Pydantic的类型驱动定义ags重度依赖Pydantic这个库。Pydantic利用Python类型注解type hints来进行数据验证和设置管理。在定义技能参数时你可以使用str,int,Literal[option1, option2]等丰富的类型。ags会利用这些类型注解自动生成准确、结构化的JSON Schema这比手动写JSON描述要可靠和高效得多并且能享受到静态类型检查的好处。from pydantic import BaseModel, Field from ags import skill, SkillParam # 定义输入参数的模型 class WeatherQueryInput(BaseModel): city: str Field(descriptionThe city name, e.g., Beijing) country_code: str Field(CN, descriptionISO country code, default is CN) # 使用装饰器定义技能 skill( nameget_weather, descriptionGet the current weather for a given city., input_modelWeatherQueryInput ) async def get_weather_skill(query: WeatherQueryInput) - str: # 这里是技能的实际实现 # 例如调用一个天气API return fThe weather in {query.city} is sunny.上面这段代码就完整定义了一个技能。skill装饰器收集了所有元数据而函数体是执行逻辑。ags会自动从WeatherQueryInput这个Pydantic模型生成对应的参数Schema。异步优先Async-first现代AI应用和网络IO密集型操作如调用API普遍采用异步编程来提高并发性能。ags的技能执行函数默认支持async def这意味着你可以在技能内部方便地使用aiohttp等异步库进行网络请求而不会阻塞整个智能体的事件循环。灵活的配置管理技能可能需要密钥、端点URL等配置。ags通过Pydantic的Settings管理理念允许你将配置注入到技能中而不是硬编码在函数里。这既保证了安全性密钥不进入代码仓库也提高了技能的可移植性。技能组合与路由ags提供了将多个技能聚合在一起的能力。你可以创建一个SkillRegistry技能注册表来管理一整套技能。当智能体传来一个工具调用请求时ags能根据工具名称快速路由到正确的技能并执行。这个注册表也可以方便地导入导出实现技能的共享。3. 从零开始定义与实现你的第一个技能理论说了不少现在我们来动手创建一个实实在在的技能。假设我们要为智能体添加一个“查询当前时间”的技能。这个技能很简单但它能完整走通ags的工作流程。3.1 环境准备与安装首先确保你有一个Python环境3.8以上。然后安装ags。由于它是一个较新的开源项目通常直接从GitHub安装最新版本。# 推荐使用uv或pip进行安装 pip install ags[all] # 安装ags及其常用依赖 # 或者从源码安装 pip install githttps://github.com/agentskill-sh/ags.git注意开源项目迭代可能较快API可能会有变动。如果遇到问题查看项目README或pyproject.toml文件中的依赖说明是很好的习惯。生产环境建议锁定版本。3.2 技能定义三步走第一步设计输入输出思考技能需要什么信息输入以及会返回什么信息输出。对于“查询时间”我们可能希望它能根据时区来返回时间。所以输入可以是一个可选的时区参数比如Asia/Shanghai输出是一个表示时间的字符串。第二步编写Pydantic模型这是将思考规范化的关键一步。我们为输入创建一个模型。from pydantic import BaseModel, Field from typing import Optional import pytz # 需要安装 pytz: pip install pytz from datetime import datetime class GetTimeInput(BaseModel): timezone: Optional[str] Field( defaultUTC, descriptionThe IANA timezone name, e.g., Asia/Shanghai, America/New_York. Default is UTC. )这里我们定义了一个GetTimeInput类它有一个timezone字段类型是可选字符串默认值是UTC并附上了描述。这个描述至关重要AI模型会阅读它来理解这个参数的意义。第三步用装饰器创建技能现在我们实现技能逻辑并用skill装饰器将其包装。from ags import skill import pytz from datetime import datetime skill( nameget_current_time, descriptionGet the current date and time for a specified timezone., input_modelGetTimeInput ) async def get_current_time(input: GetTimeInput) - str: 获取指定时区的当前时间。 tz_str input.timezone try: # 获取时区对象 tz pytz.timezone(tz_str) except pytz.exceptions.UnknownTimeZoneError: # 如果时区无效回退到UTC并在结果中说明 tz pytz.UTC current_time_utc datetime.now(pytz.UTC) return fUnknown timezone {tz_str}. Current time in UTC is: {current_time_utc.strftime(%Y-%m-%d %H:%M:%S %Z)} # 获取该时区的当前时间 current_time datetime.now(tz) # 格式化为易读的字符串 formatted_time current_time.strftime(%Y-%m-%d %H:%M:%S %Z) return fThe current time in {tz_str} is: {formatted_time}代码解读skill装饰器我们提供了技能的名称(name)、给AI看的描述(description)以及输入模型(input_model)。函数get_current_time这是一个异步函数接收一个GetTimeInput实例。函数内部尝试用pytz.timezone解析用户传入的时区字符串。如果时区无效则捕获异常使用UTC时区并返回提示信息。如果有效则获取该时区的当前时间并格式化。最后返回一个字符串结果。错误处理在技能实现中考虑边界情况和错误处理非常重要。这里我们对无效时区做了优雅降级而不是抛出异常导致智能体崩溃。在实际技能中如调用外部API更需要进行完善的错误处理和重试逻辑。3.3 将技能“装备”给智能体技能定义好了但孤零零的一个函数没法用。我们需要创建一个技能集合并暴露给智能体框架。这里以最简单的交互为例演示如何获取技能的“工具描述”并手动模拟一次调用。from ags import SkillSet # 1. 创建技能集合 skillset SkillSet() # 2. 将技能添加到集合中 skillset.add_skill(get_current_time) # 3. 获取该技能对应的“工具”定义例如给OpenAI的格式 tools_for_ai skillset.to_openai_tools() print(Tool definition for AI:) import json print(json.dumps(tools_for_ai, indent2)) # 输出大致如下 # [ # { # type: function, # function: { # name: get_current_time, # description: Get the current date and time for a specified timezone., # parameters: { # type: object, # properties: { # timezone: { # type: string, # description: The IANA timezone name, e.g., Asia/Shanghai, America/New_York. Default is UTC. # } # }, # required: [] # } # } # } # ]这个tools_for_ai列表就是你可以直接填入OpenAI API调用中tools参数的内容。AI模型在看到这个定义后就学会了在合适的时候调用get_current_time。模拟一次AI调用与技能执行 假设AI模型经过思考决定调用这个技能并生成了调用参数。# 模拟AI返回的工具调用请求 ai_tool_call { name: get_current_time, arguments: json.dumps({timezone: Asia/Shanghai}) # AI可能会生成这个JSON字符串 } # 4. 技能集合执行工具调用 import asyncio async def run_demo(): result await skillset.execute_tool_call( tool_nameai_tool_call[name], tool_argumentsai_tool_call[arguments] ) print(Skill execution result:, result) asyncio.run(run_demo()) # 输出: Skill execution result: The current time in Asia/Shanghai is: 2023-10-27 15:30:00 CSTskillset.execute_tool_call方法完成了路由和执行的脏活累活它根据tool_name找到对应的技能解析tool_argumentsJSON字符串为Pydantic模型然后执行我们定义的get_current_time函数最后返回结果。4. 进阶实战构建一个实用的网络搜索技能单一技能威力有限现在我们挑战一个更实用、也更复杂的技能网络搜索。这涉及到调用外部API如Serper、Google Search API、处理API密钥、解析返回结果等。通过这个例子你将掌握ags处理配置、异步IO和复杂返回类型的技巧。4.1 设计技能与配置管理一个网络搜索技能需要输入搜索查询词query可能还有结果数量num_results。输出结构化的搜索结果列表包含标题、链接、摘要等。配置API密钥、搜索端点URL。这些绝不能硬编码而应通过配置注入。首先定义输入模型和输出模型。输出模型用于告诉AI返回值的结构。from pydantic import BaseModel, Field, HttpUrl from typing import List class WebSearchInput(BaseModel): query: str Field(descriptionThe search query string.) num_results: int Field(default5, ge1, le10, descriptionNumber of search results to return, between 1 and 10.) class SearchResult(BaseModel): title: str Field(descriptionThe title of the search result.) link: HttpUrl Field(descriptionThe URL of the search result.) snippet: str Field(descriptionA brief summary or snippet of the result.) class WebSearchOutput(BaseModel): results: List[SearchResult] Field(descriptionList of search results.) total_estimated: int Field(descriptionEstimated total number of results found.)这里我们定义了嵌套的Pydantic模型。HttpUrl类型会自动验证字符串是否为有效的URL。ge和le用于限制num_results的范围。接下来处理配置。我们创建一个专门的配置类并从环境变量读取敏感信息。from pydantic import SecretStr from pydantic_settings import BaseSettings class SearchConfig(BaseSettings): 网络搜索技能的配置 serper_api_key: SecretStr # 使用SecretStr隐藏密钥 search_endpoint: str https://google.serper.dev/search # 默认端点 class Config: env_prefix SEARCH_ # 环境变量前缀例如 SEARCH_SERPER_API_KEY使用pydantic-settings可以方便地从环境变量、.env文件等加载配置。SecretStr类型在打印或日志中会自动显示为********增加了安全性。4.2 实现技能逻辑集成外部API现在实现技能函数。我们将使用aiohttp进行异步HTTP请求。import aiohttp from ags import skill, SkillParam from typing import Annotated skill( nameweb_search, descriptionPerform a web search and return structured results., input_modelWebSearchInput, output_modelWebSearchOutput # 声明输出模型有助于AI理解返回结构 ) async def web_search_skill( input: WebSearchInput, config: Annotated[SearchConfig, SkillParam(scopeskill)] # 通过依赖注入获取配置 ) - WebSearchOutput: 使用Serper API执行网络搜索。 headers { X-API-KEY: config.serper_api_key.get_secret_value(), Content-Type: application/json } payload { q: input.query, num: input.num_results } async with aiohttp.ClientSession() as session: try: async with session.post(config.search_endpoint, jsonpayload, headersheaders) as response: response.raise_for_status() # 如果状态码不是2xx抛出异常 data await response.json() except aiohttp.ClientError as e: # 网络或客户端错误 return WebSearchOutput( results[], total_estimated0, error_messagefNetwork error during search: {str(e)} ) except Exception as e: # 其他未知错误 return WebSearchOutput( results[], total_estimated0, error_messagefAn unexpected error occurred: {str(e)} ) # 解析Serper API的返回结果 (实际结构需参考API文档) # 这里是一个示例解析逻辑 organic_results data.get(organic, []) search_results [] for item in organic_results[:input.num_results]: search_results.append( SearchResult( titleitem.get(title, No title), linkitem.get(link, ), snippetitem.get(snippet, ) ) ) return WebSearchOutput( resultssearch_results, total_estimateddata.get(searchInformation, {}).get(totalResults, 0) )关键点解析依赖注入config: Annotated[SearchConfig, SkillParam(scopeskill)]这行是ags的一个强大特性。它告诉ags当执行这个技能时请自动为我提供一个SearchConfig的实例。SkillParam(scopeskill)表示这个配置是在技能级别共享的。ags会在背后管理配置的初始化和注入。异步HTTP客户端使用aiohttp.ClientSession进行异步请求避免阻塞。全面的错误处理网络请求可能失败API可能返回错误。我们用try...except捕获aiohttp.ClientError和其他异常并返回一个包含错误信息的WebSearchOutput而不是让异常向上传播导致智能体崩溃。这是生产级技能的必要考虑。输出模型最终我们返回一个WebSearchOutput实例其中包含了结构化的结果列表。AI在收到这个结果后可以清晰地引用其中的title和link。4.3 配置与运行在运行前需要设置环境变量。export SEARCH_SERPER_API_KEYyour_actual_api_key_here然后在你的智能体主程序中import asyncio from ags import SkillSet async def main(): # 技能集会自动从环境变量读取配置并注入 skillset SkillSet() skillset.add_skill(web_search_skill) # 获取工具定义 tools skillset.to_openai_tools() # ... 将tools提供给AI模型 ... # 模拟AI调用搜索 ai_call { name: web_search, arguments: json.dumps({query: latest developments in AI agents, num_results: 3}) } result await skillset.execute_tool_call(**ai_call) print(fSearch completed. Found {result.total_estimated} results.) for res in result.results: print(f- {res.title}: {res.link}) if __name__ __main__: asyncio.run(main())5. 工程化实践技能管理、测试与最佳实践当技能数量增多后如何有效地组织、测试和集成它们就成为了工程上的挑战。这一部分分享一些在实战中积累的经验。5.1 技能的组织与发现不建议把所有技能都写在一个文件里。一个好的实践是按领域或功能模块组织技能。my_agent_project/ ├── skills/ │ ├── __init__.py │ ├── base.py # 基础配置、公共工具 │ ├── web.py # 网络相关技能搜索、爬虫 │ ├── productivity.py # 效率工具日历、邮件 │ └── data.py # 数据处理技能 ├── config/ │ └── settings.py # 统一配置管理 ├── agent_main.py # 智能体主程序 └── requirements.txt在skills/__init__.py中可以集中导出所有技能方便主程序导入。# skills/__init__.py from .web import web_search_skill, fetch_webpage_skill from .productivity import send_email_skill, create_calendar_event_skill __all__ [ web_search_skill, fetch_webpage_skill, send_email_skill, create_calendar_event_skill, ]主程序只需要from skills import *然后添加到SkillSet中。对于更复杂的项目可以考虑使用SkillRegistry类它提供了更丰富的技能管理功能如按标签过滤、动态加载等。5.2 技能的单元测试技能也是代码必须可测试。由于技能通常是异步函数且可能依赖外部服务测试时需要一些技巧。1. 模拟Mock外部依赖对于网络请求、数据库访问等使用unittest.mock或pytest-mock来模拟。import pytest from unittest.mock import AsyncMock, patch from skills.web import web_search_skill, WebSearchInput pytest.mark.asyncio async def test_web_search_success(mocker): # 1. Mock配置 mock_config mocker.MagicMock() mock_config.serper_api_key.get_secret_value.return_value fake_key mock_config.search_endpoint https://fake.endpoint # 2. Mock aiohttp的响应 fake_json { organic: [ {title: Test Result 1, link: https://example.com/1, snippet: Snippet 1}, {title: Test Result 2, link: https://example.com/2, snippet: Snippet 2}, ], searchInformation: {totalResults: 100} } mock_response AsyncMock() mock_response.json AsyncMock(return_valuefake_json) mock_response.raise_for_status AsyncMock() mock_session mocker.patch(aiohttp.ClientSession) mock_session_instance AsyncMock() mock_session.return_value.__aenter__.return_value mock_session_instance mock_session_instance.post.return_value.__aenter__.return_value mock_response # 3. 准备输入 search_input WebSearchInput(querytest query, num_results2) # 4. 执行技能 (需要手动注入被mock的config) # 注意这里需要根据ags的具体测试方式调整可能需要使用ags的测试工具 # 假设我们直接调用函数在知道如何注入config的情况下 result await web_search_skill(inputsearch_input, configmock_config) # 5. 断言 assert len(result.results) 2 assert result.results[0].title Test Result 1 assert result.total_estimated 100 # 验证mock是否被以预期方式调用 mock_session_instance.post.assert_called_once_with( https://fake.endpoint, json{q: test query, num: 2}, headers{X-API-KEY: fake_key, Content-Type: application/json} )2. 测试技能的工具描述生成确保技能生成的JSON Schema符合预期。def test_skill_tool_schema(): tools web_search_skill.to_tools() # 假设skill对象有to_tools方法 tool_def tools[0] assert tool_def[function][name] web_search assert query in tool_def[function][parameters][properties] assert tool_def[function][parameters][properties][num_results][default] 55.3 性能、安全与错误处理最佳实践超时与重试所有涉及网络IO的技能都必须设置超时并考虑重试逻辑。可以在技能内部实现或者使用像tenacity这样的重试库。aiohttp本身支持超时设置。timeout aiohttp.ClientTimeout(total10) # 10秒总超时 async with session.post(url, jsonpayload, timeouttimeout) as response: ...速率限制如果调用的是有速率限制的第三方API如OpenAI、Serper需要在技能或调用层面实现限流避免触发限制。可以使用asyncio.Semaphore或专门的限流库。输入验证与净化Pydantic提供了强大的输入验证但有时需要额外处理。例如对于接收URL并下载内容的技能需要验证URL的协议只允许http/https防止SSRF攻击。对于接收用户输入并用于数据库查询的技能要警惕SQL注入。敏感信息处理永远不要在技能代码、日志或返回给AI的结果中硬编码或泄露API密钥、密码。使用SecretStr、SecretBytes等类型。配置从环境变量或安全的配置管理服务如Vault中读取。在日志中对敏感参数进行脱敏处理。错误的分类与反馈技能执行失败时返回给AI的错误信息应具有指导性但又不能泄露内部细节。可以定义一套错误码和用户友好的错误信息。class SkillError(Exception): 技能基础异常 def __init__(self, message: str, user_friendly_msg: str, error_code: str): super().__init__(message) self.user_friendly_msg user_friendly_msg self.error_code error_code class ExternalServiceError(SkillError): 外部服务错误 pass # 在技能中 try: await call_external_api() except aiohttp.ClientResponseError as e: if e.status 429: raise ExternalServiceError( messagefAPI rate limited: {e}, user_friendly_msgThe search service is currently busy. Please try again in a moment., error_codeRATE_LIMIT )然后在智能体层面可以捕获这些自定义异常并将user_friendly_msg传递给AI或最终用户。6. 集成到智能体框架以LangChain为例ags技能最终要服务于智能体。这里以流行的LangChain框架为例展示如何无缝集成。假设我们已经有了一个SkillSet对象my_skillset里面包含了web_search_skill和get_current_time_skill。第一步将ags技能转换为LangChain ToolLangChain有自己的Tool抽象。我们需要一个简单的适配器。from langchain.tools import BaseTool from langchain.callbacks.manager import CallbackManagerForToolRun from typing import Optional, Type from pydantic import BaseModel, Field class AGSAdapter(BaseTool): 适配器将ags技能包装成LangChain Tool skill_name: str skill_set: SkillSet # 持有ags的技能集合 def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] None) - str: # LangChain的Tool默认同步调用但ags技能是异步的。 # 这里需要在异步上下文中运行或者使用同步执行方法如果ags提供。 # 假设skill_set有同步执行方法例如通过asyncio.run在内部处理 # 注意这只是一个示例实际集成可能需要处理异步到同步的转换。 result asyncio.run(self.skill_set.execute_tool_call(self.skill_name, query)) return str(result) # 将结果转为字符串 async def _arun(self, query: str, run_manager: Optional[CallbackManagerForToolRun] None) - str: # 异步版本 result await self.skill_set.execute_tool_call(self.skill_name, query) return str(result) # 为每个技能创建适配器 search_tool AGSAdapter( nameweb_search, descriptionUseful for searching the web for current information., skill_nameweb_search, skill_setmy_skillset ) time_tool AGSAdapter( nameget_current_time, descriptionUseful for getting the current time in any timezone., skill_nameget_current_time, skill_setmy_skillset )第二步在LangChain Agent中使用现在可以将这些Tool提供给LangChain的Agent。from langchain.agents import initialize_agent, AgentType from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory llm ChatOpenAI(modelgpt-4, temperature0) memory ConversationBufferMemory(memory_keychat_history) tools [search_tool, time_tool] # 我们的ags技能工具 agent initialize_agent( tools, llm, agentAgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 适合对话的Agent类型 memorymemory, verboseTrue ) # 运行Agent response agent.run(Whats the current time in Tokyo, and then search for the latest news about it?) print(response)在这个流程中LangChain Agent负责与用户对话、规划思考过程ReAct模式当它认为需要搜索或查时间时会调用我们提供的Tool而这个Tool背后实际执行的是ags技能。更优雅的集成上述适配器方法比较直接。ags项目未来可能会提供官方的LangChain集成或者社区会有更成熟的方案。核心思想是ags负责技能的标准化定义和执行而智能体框架负责高层的推理、规划和工具调用调度两者各司其职通过一个轻量级的适配层连接。7. 常见问题、排查与性能调优在实际开发和运维中你肯定会遇到各种问题。这里记录一些典型场景和解决思路。7.1 技能执行失败排查清单问题现象可能原因排查步骤AI模型不调用技能1. 技能描述不清晰。2. 技能名称/参数名与AI训练数据不匹配。3. 智能体配置未正确加载工具。1. 检查skill中的description是否准确描述了技能功能和适用场景。2. 检查生成的Tool定义to_openai_tools()输出确保JSON Schema格式正确。3. 在智能体初始化后打印其可用工具列表确认。技能被调用但参数解析错误1. AI生成的参数JSON格式错误。2. Pydantic模型验证失败类型不符、必填项缺失。1. 在skillset.execute_tool_call前后打印tool_arguments检查是否为合法JSON。2. 查看Pydantic抛出的验证错误详情调整模型定义或给AI更明确的参数描述。技能执行超时或挂起1. 外部API响应慢或无响应。2. 技能函数内有阻塞操作或死循环。3. 异步事件循环被阻塞。1. 为网络请求添加超时如aiohttp.ClientTimeout。2. 检查技能代码确保IO操作都是异步的使用await。3. 考虑在技能级别或调用侧设置整体超时。返回结果AI无法理解1. 返回类型过于复杂或非结构化。2. 返回了AI无法解析的二进制数据或特殊对象。1. 尽量返回字符串或简单的字典/列表。使用output_model明确输出结构。2. 对于复杂对象在技能内将其转换为文本描述。例如将Pandas DataFrame转为Markdown表格字符串。配置注入失败1. 环境变量未设置或名称错误。2.SkillParam依赖注入作用域配置错误。1. 确认环境变量前缀如SEARCH_和变量名正确。2. 在技能函数内打印或日志记录传入的config对象检查其属性是否为预期值。3. 查阅ags关于依赖注入和作用域的文档。7.2 性能优化要点技能懒加载如果技能数量很多但一次对话只用其中几个可以考虑懒加载机制。即不在启动时初始化所有技能特别是那些依赖重型资源或慢速连接的技能而是在第一次被调用时再初始化。这可以通过自定义Skill类或使用工厂模式实现。连接池与缓存对于频繁调用外部API的技能如数据库查询、向量检索使用连接池如aiohttp.ClientSession重用和适当的缓存策略可以极大提升性能。例如对某些只读的、更新不频繁的查询结果缓存几分钟。from functools import lru_cache import asyncio lru_cache(maxsize128) def _sync_expensive_call(param): # 同步的昂贵调用 pass async def expensive_skill(param: str): # 在异步函数中运行同步缓存调用使用线程池执行器避免阻塞事件循环 loop asyncio.get_event_loop() result await loop.run_in_executor(None, _sync_expensive_call, param) return result注意缓存需要设计合理的失效策略避免返回过期数据。异步并发执行如果一个智能体需要并行执行多个独立技能例如同时查询天气和新闻可以利用asyncio.gather。但需要注意这需要智能体框架本身支持并行工具调用目前大多数框架是顺序执行的。监控与日志为关键技能添加详细的日志记录输入、输出、执行时间。这有助于性能分析和故障排查。可以使用结构化日志如structlog方便后续聚合分析。7.3 设计模式技能编排与组合有时一个复杂任务需要按顺序调用多个技能。虽然这通常由智能体的“大脑”来规划但也可以将固定的工作流封装成一个“复合技能”。skill(nameresearch_and_summarize, descriptionResearch a topic and provide a summary.) async def research_skill(topic: str) - str: 1. 搜索主题 2. 抓取前3个结果的内容 3. 调用LLM进行总结 # 1. 搜索 search_results await web_search_skill(WebSearchInput(querytopic, num_results3)) if not search_results.results: return No relevant information found. # 2. 并发抓取网页内容 (假设有fetch_webpage_skill) fetch_tasks [] for result in search_results.results: task fetch_webpage_skill(FetchWebpageInput(urlstr(result.link))) fetch_tasks.append(task) webpage_contents await asyncio.gather(*fetch_tasks, return_exceptionsTrue) # 处理抓取成功和失败的内容... # 3. 组合内容并总结 (假设有summarize_text_skill) combined_text \n---\n.join([c for c in webpage_contents if isinstance(c, str)]) summary await summarize_text_skill(SummarizeInput(textcombined_text, max_length500)) return summary这种模式将多个原子技能组合成一个更高级别的技能对AI来说是一个“宏操作”简化了智能体的规划复杂度。但要注意控制复合技能的复杂度和执行时间。