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的出现，就是为了抽象掉这些底层细节。它的设计目标很明确：

1. 标准化：提供一套统一的、模型无关的技能描述规范。
2. 声明式：用Python代码而非手工JSON来定义技能，利用IDE的自动补全和类型检查。
3. 解耦：将技能的定义、描述（给AI看）和执行（实际运行）清晰分离。
4. 可发现：让智能体能轻松地知道自己有哪些技能可用。

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是如何实现上述设计的？我们来看几个关键技术点：

1. 基于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(description="The city name, e.g., 'Beijing'") country_code: str = Field("CN", description="ISO country code, default is CN") # 使用装饰器定义技能 @skill( name="get_weather", description="Get the current weather for a given city.", input_model=WeatherQueryInput ) async def get_weather_skill(query: WeatherQueryInput) -> str: # 这里是技能的实际实现 # 例如调用一个天气API return f"The weather in {query.city} is sunny."

   上面这段代码就完整定义了一个技能。@skill装饰器收集了所有元数据，而函数体是执行逻辑。ags会自动从WeatherQueryInput这个Pydantic模型生成对应的参数Schema。

2. 异步优先（Async-first）：现代AI应用和网络IO密集型操作（如调用API）普遍采用异步编程来提高并发性能。ags的技能执行函数默认支持async def，这意味着你可以在技能内部方便地使用aiohttp等异步库进行网络请求，而不会阻塞整个智能体的事件循环。

3. 灵活的配置管理：技能可能需要密钥、端点URL等配置。ags通过Pydantic的Settings管理理念，允许你将配置注入到技能中，而不是硬编码在函数里。这既保证了安全性（密钥不进入代码仓库），也提高了技能的可移植性。

4. 技能组合与路由：ags提供了将多个技能聚合在一起的能力。你可以创建一个SkillRegistry（技能注册表）来管理一整套技能。当智能体传来一个工具调用请求时，ags能根据工具名称快速路由到正确的技能并执行。这个注册表也可以方便地导入导出，实现技能的共享。

3. 从零开始：定义与实现你的第一个技能

理论说了不少，现在我们来动手创建一个实实在在的技能。假设我们要为智能体添加一个“查询当前时间”的技能。这个技能很简单，但它能完整走通ags的工作流程。

3.1 环境准备与安装

首先，确保你有一个Python环境（3.8以上）。然后安装ags。由于它是一个较新的开源项目，通常直接从GitHub安装最新版本。

# 推荐使用uv或pip进行安装 pip install "ags[all]" # 安装ags及其常用依赖 # 或者从源码安装 pip install git+https://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( default="UTC", description="The 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( name="get_current_time", description="Get the current date and time for a specified timezone.", input_model=GetTimeInput ) 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 f"Unknown 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 f"The current time in {tz_str} is: {formatted_time}"

代码解读：

1. @skill装饰器：我们提供了技能的名称(name)、给AI看的描述(description)，以及输入模型(input_model)。
2. 函数get_current_time：这是一个异步函数，接收一个GetTimeInput实例。函数内部：
   • 尝试用pytz.timezone解析用户传入的时区字符串。
   • 如果时区无效，则捕获异常，使用UTC时区并返回提示信息。
   • 如果有效，则获取该时区的当前时间并格式化。
   • 最后返回一个字符串结果。
3. 错误处理：在技能实现中考虑边界情况和错误处理非常重要。这里我们对无效时区做了优雅降级，而不是抛出异常导致智能体崩溃。在实际技能中，如调用外部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, indent=2)) # 输出大致如下： # [ # { # "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_name=ai_tool_call["name"], tool_arguments=ai_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 CST

skillset.execute_tool_call方法完成了路由和执行的脏活累活：它根据tool_name找到对应的技能，解析tool_arguments（JSON字符串）为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(description="The search query string.") num_results: int = Field(default=5, ge=1, le=10, description="Number of search results to return, between 1 and 10.") class SearchResult(BaseModel): title: str = Field(description="The title of the search result.") link: HttpUrl = Field(description="The URL of the search result.") snippet: str = Field(description="A brief summary or snippet of the result.") class WebSearchOutput(BaseModel): results: List[SearchResult] = Field(description="List of search results.") total_estimated: int = Field(description="Estimated 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( name="web_search", description="Perform a web search and return structured results.", input_model=WebSearchInput, output_model=WebSearchOutput # 声明输出模型，有助于AI理解返回结构 ) async def web_search_skill( input: WebSearchInput, config: Annotated[SearchConfig, SkillParam(scope="skill")] # 通过依赖注入获取配置 ) -> 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, json=payload, headers=headers) as response: response.raise_for_status() # 如果状态码不是2xx，抛出异常 data = await response.json() except aiohttp.ClientError as e: # 网络或客户端错误 return WebSearchOutput( results=[], total_estimated=0, error_message=f"Network error during search: {str(e)}" ) except Exception as e: # 其他未知错误 return WebSearchOutput( results=[], total_estimated=0, error_message=f"An 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( title=item.get("title", "No title"), link=item.get("link", ""), snippet=item.get("snippet", "") ) ) return WebSearchOutput( results=search_results, total_estimated=data.get("searchInformation", {}).get("totalResults", 0) )

关键点解析：

1. 依赖注入：config: Annotated[SearchConfig, SkillParam(scope="skill")]这行是ags的一个强大特性。它告诉ags，当执行这个技能时，请自动为我提供一个SearchConfig的实例。SkillParam(scope="skill")表示这个配置是在技能级别共享的。ags会在背后管理配置的初始化和注入。
2. 异步HTTP客户端：使用aiohttp.ClientSession进行异步请求，避免阻塞。
3. 全面的错误处理：网络请求可能失败，API可能返回错误。我们用try...except捕获aiohttp.ClientError和其他异常，并返回一个包含错误信息的WebSearchOutput，而不是让异常向上传播导致智能体崩溃。这是生产级技能的必要考虑。
4. 输出模型：最终我们返回一个WebSearchOutput实例，其中包含了结构化的结果列表。AI在收到这个结果后，可以清晰地引用其中的title和link。

4.3 配置与运行

在运行前，需要设置环境变量。

export SEARCH_SERPER_API_KEY="your_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(f"Search 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_value=fake_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(query="test query", num_results=2) # 4. 执行技能 (需要手动注入被mock的config) # 注意：这里需要根据ags的具体测试方式调整，可能需要使用ags的测试工具 # 假设我们直接调用函数（在知道如何注入config的情况下） result = await web_search_skill(input=search_input, config=mock_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"] == 5

5.3 性能、安全与错误处理最佳实践

1. 超时与重试：所有涉及网络IO的技能都必须设置超时，并考虑重试逻辑。可以在技能内部实现，或者使用像tenacity这样的重试库。aiohttp本身支持超时设置。

   timeout = aiohttp.ClientTimeout(total=10) # 10秒总超时 async with session.post(url, json=payload, timeout=timeout) as response: ...

2. 速率限制：如果调用的是有速率限制的第三方API（如OpenAI、Serper），需要在技能或调用层面实现限流，避免触发限制。可以使用asyncio.Semaphore或专门的限流库。

3. 输入验证与净化：Pydantic提供了强大的输入验证，但有时需要额外处理。例如，对于接收URL并下载内容的技能，需要验证URL的协议（只允许http/https），防止SSRF攻击。对于接收用户输入并用于数据库查询的技能，要警惕SQL注入。

4. 敏感信息处理：

   • 永远不要在技能代码、日志或返回给AI的结果中硬编码或泄露API密钥、密码。
   • 使用SecretStr、SecretBytes等类型。
   • 配置从环境变量或安全的配置管理服务（如Vault）中读取。
   • 在日志中，对敏感参数进行脱敏处理。

5. 错误的分类与反馈：技能执行失败时，返回给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( message=f"API rate limited: {e}", user_friendly_msg="The search service is currently busy. Please try again in a moment.", error_code="RATE_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( name="web_search", description="Useful for searching the web for current information.", skill_name="web_search", skill_set=my_skillset ) time_tool = AGSAdapter( name="get_current_time", description="Useful for getting the current time in any timezone.", skill_name="get_current_time", skill_set=my_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(model="gpt-4", temperature=0) memory = ConversationBufferMemory(memory_key="chat_history") tools = [search_tool, time_tool] # 我们的ags技能工具 agent = initialize_agent( tools, llm, agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 适合对话的Agent类型 memory=memory, verbose=True ) # 运行Agent response = agent.run("What's 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 性能优化要点

1. 技能懒加载：如果技能数量很多，但一次对话只用其中几个，可以考虑懒加载机制。即不在启动时初始化所有技能（特别是那些依赖重型资源或慢速连接的技能），而是在第一次被调用时再初始化。这可以通过自定义Skill类或使用工厂模式实现。

2. 连接池与缓存：对于频繁调用外部API的技能（如数据库查询、向量检索），使用连接池（如aiohttp.ClientSession重用）和适当的缓存策略可以极大提升性能。例如，对某些只读的、更新不频繁的查询结果缓存几分钟。

   from functools import lru_cache import asyncio @lru_cache(maxsize=128) 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

   注意：缓存需要设计合理的失效策略，避免返回过期数据。

3. 异步并发执行：如果一个智能体需要并行执行多个独立技能（例如同时查询天气和新闻），可以利用asyncio.gather。但需要注意，这需要智能体框架本身支持并行工具调用，目前大多数框架是顺序执行的。

4. 监控与日志：为关键技能添加详细的日志，记录输入、输出、执行时间。这有助于性能分析和故障排查。可以使用结构化日志（如structlog）方便后续聚合分析。

7.3 设计模式：技能编排与组合

有时，一个复杂任务需要按顺序调用多个技能。虽然这通常由智能体的“大脑”来规划，但也可以将固定的工作流封装成一个“复合技能”。

@skill(name="research_and_summarize", description="Research 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(query=topic, num_results=3)) 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(url=str(result.link))) fetch_tasks.append(task) webpage_contents = await asyncio.gather(*fetch_tasks, return_exceptions=True) # 处理抓取成功和失败的内容... # 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(text=combined_text, max_length=500)) return summary

这种模式将多个原子技能组合成一个更高级别的技能，对AI来说是一个“宏操作”，简化了智能体的规划复杂度。但要注意控制复合技能的复杂度和执行时间。