在AI Agent开发领域,Skill(技能)的编写质量直接决定了Agent的智能水平和任务完成能力。很多开发者在初次尝试编写Skill时,常常会陷入一些常见的“坑”中,导致Skill运行不稳定、意图识别不准或难以维护。本文基于大量实战经验,系统梳理了Skill编写中最常见的六个陷阱,并提供了一套完整的避坑指南和最佳实践。无论你是刚开始接触AI Agent的新手,还是希望优化现有Skill的开发者,都能从中获得实用的解决方案和工程化建议。

1. Skill编写核心概念与常见误区

在深入探讨具体“坑点”之前,我们有必要明确Skill在AI Agent体系中的定位及其编写的基本逻辑。

1.1 什么是AI Agent中的Skill?

Skill,通常被称为“技能”或“能力”,是赋予AI Agent执行特定任务的核心模块。你可以将其理解为一个封装好的功能函数或微服务,它接收自然语言指令或结构化输入,经过内部逻辑处理,最终输出结果或执行动作。

一个典型的Skill包含以下几个关键部分:

  • 意图识别(Intent Recognition) : 理解用户输入想要触发哪个Skill。例如,用户说“查一下北京的天气”,意图是“查询天气”。
  • 参数抽取(Slot Filling) : 从用户输入中提取执行任务所需的必要信息。例如,从“查一下北京的天气”中抽取地点参数“北京”。
  • 核心逻辑(Core Logic) : 执行任务的实际代码,可能是调用一个API、查询数据库或运行一段计算。
  • 响应格式化(Response Formatting) : 将处理结果转换成Agent能够理解或用户易于接受的格式(如自然语言、JSON、HTML等)。

1.2 Skill编写与普通脚本编程的本质区别

许多开发者第一个“坑”就是将Skill编写等同于编写普通函数。这是根本性的误解。

普通函数编程 关注的是输入、处理、输出的确定性和正确性。你传入参数 x ,函数必然返回 f(x)

Skill编程 则处于一个不确定的环境中,需要处理:

  1. 输入的模糊性 : 用户可能用多种方式表达同一意图(“播放音乐”、“来首歌”、“我想听歌”)。
  2. 上下文依赖性 : 当前对话的历史会影响Skill的理解(用户先说“我想听周杰伦的歌”,再说“播放七里香”,后者需要依赖前文的歌手信息)。
  3. 异常与降级处理 : 当API调用失败、参数缺失或结果为空时,Skill需要有能力进行友好地回复或尝试替代方案,而不是直接抛出异常让整个Agent崩溃。
  4. 可解释性 : Skill不仅要做,还要能向用户或开发者“解释”自己做了什么、为什么这么做、遇到了什么问题。

忽略这些区别,直接套用传统编程思维,是导致Skill脆弱、不智能的首要原因。

2. 环境准备与基础框架选择

在开始编写Skill之前,选择一个合适的开发框架和配置好环境至关重要。这能帮你规避后续许多工具链和集成问题。

2.1 主流AI Agent开发框架简介

目前社区有多种支持Skill开发的框架,各有侧重:

  • LangChain / LangGraph : 生态庞大,组件丰富,适合构建复杂的、有状态的Agent工作流。Skill在其中通常以 Tool Runnable 的形式存在。
  • Semantic Kernel : 微软推出,与.NET生态结合紧密,强调“规划器(Planner)”与“技能(Skill)”的分离,适合企业级应用。
  • AutoGen : 由微软推出,专注于多智能体对话,Skill可以作为Agent的能力被调用,适合研究型和复杂协作场景。
  • 自定义框架 : 许多项目也会基于OpenAI的Function Calling或Anthropic的Tools等功能,自行封装一套轻量级的Skill管理机制。

对于初学者和大多数应用场景, LangChain 因其丰富的文档、教程和社区支持,是入门和实战的首选。

2.2 基础环境搭建(以LangChain为例)

假设我们使用Python和LangChain来开发Skill。

1. 创建虚拟环境并安装依赖 强烈建议使用虚拟环境来隔离项目依赖。

# 创建并进入项目目录
mkdir my_agent_project && cd my_agent_project

# 创建Python虚拟环境(以venv为例)
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate

# 安装核心依赖
pip install langchain langchain-openai langchain-community
# 安装其他可能需要的工具库
pip install requests python-dotenv pydantic

2. 配置环境变量 将你的API密钥等敏感信息存储在环境变量中,不要硬编码在代码里。创建 .env 文件:

# .env
OPENAI_API_KEY=your_openai_api_key_here
WEATHER_API_KEY=your_weather_api_key_here  # 示例

在代码中通过 os.getenv 加载。

3. 项目结构建议 一个清晰的项目结构有助于Skill的管理和扩展。

my_agent_project/
├── .env                    # 环境变量
├── requirements.txt        # 依赖列表
├── main.py                 # Agent主入口
├── skills/                 # 技能包目录
│   ├── __init__.py
│   ├── base_skill.py      # 技能基类(可选)
│   ├── weather_skill.py   # 天气查询技能
│   ├── calculator_skill.py # 计算器技能
│   └── web_search_skill.py # 网络搜索技能
├── agents/                 # 智能体定义
│   └── main_agent.py
└── utils/                  # 工具函数
    └── config.py

做好这些准备工作,相当于为建造高楼打好了地基,能有效避免后续因环境混乱、依赖冲突导致的“隐形坑”。

3. Skill编写的六大常见“坑”及避坑指南

接下来,我们进入核心部分,逐一剖析Skill编写中最常见的六个陷阱,并提供具体的解决方案和代码示例。

3.1 坑一:意图描述模糊不清,导致Agent“听不懂”

问题现象 : 你写了一个“订餐”Skill,但用户说“帮我点个外卖”、“想吃披萨”或“饿了”时,Agent无法正确触发该技能。

根本原因 : Skill的意图(Intent)描述过于狭窄或抽象,没有覆盖用户可能的表达方式。在基于LLM的Agent中,意图通常通过函数(Function/Tool)的 description parameters description 来隐式定义。描述不清,LLM就无法准确匹配。

避坑方案

  1. 使用具体、场景化的描述 : 不要写“处理餐饮”,要写“根据用户提供的位置和食物偏好,推荐餐厅或下单外卖”。
  2. 在参数描述中举例 : 在描述参数时,可以加入常见示例值。
  3. 利用Few-shot示例 : 在Agent的系统提示词(System Prompt)中,提供几个该Skill的调用示例。

代码示例(LangChain Tool定义)

# skills/restaurant_skill.py - 错误示范(描述模糊)
from langchain.tools import Tool

bad_tool = Tool(
    name="order_food",
    func=order_pizza, # 实际执行函数
    description="Order food", # 描述太简单
)

# skills/restaurant_skill.py - 正确示范(描述具体)
from langchain.tools import Tool
from pydantic import BaseModel, Field
from typing import Optional

# 使用Pydantic模型定义清晰的输入结构
class RestaurantInput(BaseModel):
    location: str = Field(description="送餐地址或当前所在位置,例如:'北京中关村'、'上海市徐汇区'")
    cuisine: Optional[str] = Field(default=None, description="想吃的菜系,例如:'披萨'、'中餐'、'沙拉'")
    budget: Optional[str] = Field(default=None, description="预算范围,例如:'50元以下'、'100-200元'")

def find_restaurant(location: str, cuisine: str = None, budget: str = None) -> str:
    # 模拟查找餐厅的逻辑
    result = f"正在为您在{location}附近寻找"
    if cuisine:
        result += f"{cuisine}菜系的"
    if budget:
        result += f"预算{budget}的"
    result += "餐厅..."
    # 实际应调用API
    return result

good_tool = Tool.from_function(
    func=find_restaurant,
    name="find_restaurant",
    description="根据用户的位置、菜系偏好和预算,查找并推荐附近的餐厅。当用户表达饿了、想吃饭、点外卖、推荐美食等意图时使用。",
    args_schema=RestaurantInput # 使用Pydantic Schema明确参数
)

关键点 description 字段是给LLM看的“说明书”,写得越像真实用户场景和需求,匹配准确率越高。

3.2 坑二:参数处理缺乏校验与默认逻辑

问题现象 : Skill运行时崩溃,或因为用户未提供必要参数(如“查天气”但没说地点)而无法工作。

根本原因 : 函数内部假设参数总是存在且有效,没有进行空值校验、类型转换和提供合理的默认值或交互式补全。

避坑方案

  1. 强制校验 : 使用Pydantic等库在数据流入时就进行类型和约束校验。
  2. 优雅降级 : 对于非核心参数,提供默认值。
  3. 交互式补全 : 对于核心缺失参数,Skill应能通过多轮对话向用户询问。这通常需要Agent层面的状态管理来支持。
  4. 参数推理 : 利用上下文推断缺失参数。例如,用户之前说过“我在北京”,那么当他说“天气怎么样?”时,地点参数应默认为“北京”。

代码示例(带校验和默认值的Skill)

# skills/weather_skill.py
import requests
from pydantic import BaseModel, Field, validator
from typing import Optional
import os
from dotenv import load_dotenv

load_dotenv()

class WeatherInput(BaseModel):
    city: str = Field(description="需要查询天气的城市名称,例如:北京、上海、New York")
    date: Optional[str] = Field(default="today", description="查询日期,例如:'today'、'tomorrow'、'2024-10-01',默认为今天")

    @validator('city')
    def city_must_not_be_empty(cls, v):
        if not v or not v.strip():
            raise ValueError('城市名称不能为空')
        return v.strip()

def get_weather(city: str, date: str = "today") -> str:
    """
    获取指定城市天气的核心函数。
    注意:这里使用了模拟数据,真实场景应调用天气API。
    """
    # 1. 参数预处理(即使有校验,内部仍可做清洗)
    city_normalized = city.title() # 简单规范化

    # 2. 模拟根据日期调整逻辑
    if date not in ["today", "tomorrow"]:
        # 处理特定日期,这里简单模拟
        forecast_date = date
    else:
        forecast_date = date

    # 3. 模拟API调用和错误处理
    api_key = os.getenv("WEATHER_API_KEY")
    if not api_key:
        # 降级处理:返回模拟数据
        return f"[模拟数据] {city_normalized}在{forecast_date}的天气:晴,气温15-25℃。请注意,未配置真实API密钥。"

    # 4. 真实API调用(示例结构)
    try:
        # url = f"https://api.weatherapi.com/v1/forecast.json?key={api_key}&q={city}&days=1"
        # response = requests.get(url, timeout=10)
        # response.raise_for_status()
        # data = response.json()
        # ... 解析data ...
        # return f"{city}的天气是..."
        return f"[模拟API调用] 成功查询{city_normalized}在{forecast_date}的天气。"
    except requests.exceptions.RequestException as e:
        # 网络或API错误处理
        return f"抱歉,查询{city_normalized}天气时遇到网络或服务错误:{e}。请稍后再试。"
    except KeyError as e:
        # API响应格式异常处理
        return f"天气服务返回的数据格式异常,无法解析。请检查城市名称'{city_normalized}'是否正确。"

# 封装成Tool
weather_tool = Tool.from_function(
    func=get_weather,
    name="get_weather",
    description="查询指定城市在指定日期的天气情况。当用户询问天气、气温、是否下雨、穿什么衣服时使用。",
    args_schema=WeatherInput
)

3.3 坑三:技能功能过于庞杂,单一职责原则被破坏

问题现象 : 一个名为 handle_user_request 的Skill,内部包含了查询天气、计算数学、发送邮件、搜索网页等数十个功能,代码长达上千行,难以维护、测试,且意图识别极其困难。

根本原因 : 开发者试图用一个Skill解决所有问题,违背了软件工程的“单一职责原则”(SRP)。

避坑方案

  1. 原子化拆分 : 每个Skill只做一件明确的事情。 get_weather calculate_math send_email search_web 应该是四个独立的Skill。
  2. 功能聚合 : 如果确实存在一组紧密关联的操作,可以创建一个“协调器”Skill或利用Agent的“规划(Planning)”能力来按顺序调用多个原子Skill。
  3. 命名清晰 : Skill的名称应直接反映其功能,见名知意。

重构示例 : 将臃肿的 handle_user_request 拆解:

# 之前:一个庞大的skill
# def handle_user_request(request: str): # 内部通过if-else判断执行不同逻辑

# 之后:多个清晰的skill
skills = [
    Tool.from_function(func=get_weather, name="get_weather", description="..."),
    Tool.from_function(func=calculate, name="calculate", description="..."),
    Tool.from_function(func=send_email, name="send_email", description="..."),
    Tool.from_function(func=search_web, name="search_web", description="..."),
]

# Agent可以自动根据用户请求选择最合适的Tool

3.4 坑四:忽视异常处理与用户友好反馈

问题现象 : Skill内部调用失败时(如网络超时、API限流、数据解析错误),直接抛出异常,导致Agent会话终止,给用户的错误信息是晦涩的Python Traceback。

根本原因 : 仅考虑了“快乐路径”(Happy Path),未对可能出错的环节进行防御性编程和友好化处理。

避坑方案

  1. 全面Try-Catch : 在Skill内部,对所有外部依赖(网络IO、数据库、文件读写)的调用进行异常捕获。
  2. 分类处理异常 : 区分网络错误、权限错误、数据错误、逻辑错误等,并提供不同的恢复或反馈策略。
  3. 返回可理解的错误信息 : 将内部异常转换为对用户友好的自然语言描述,并可能给出建议操作。
  4. 记录日志 : 将详细的错误信息记录到日志系统,便于开发者排查,而不是展示给用户。

代码示例(完善的异常处理)

# skills/web_search_skill.py
import requests
import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_web_search(query: str) -> str:
    """带有重试和异常处理的网络搜索"""
    search_url = "https://api.search.example.com/v1/search"
    headers = {"Authorization": f"Bearer {os.getenv('SEARCH_API_KEY')}"}
    params = {"q": query, "limit": 5}

    try:
        response = requests.get(search_url, headers=headers, params=params, timeout=15)
        response.raise_for_status()  # 如果状态码不是200,抛出HTTPError
        data = response.json()

        # 检查API返回的业务状态码
        if data.get('status') != 'success':
            error_msg = data.get('message', '搜索服务返回未知错误')
            logger.warning(f"搜索API业务错误: {error_msg}, Query: {query}")
            return f"搜索服务暂时不可用,请稍后再试。错误码:{error_msg}"

        # 正常处理结果
        results = data.get('results', [])
        if not results:
            return f"关于'{query}',没有找到相关结果。"
        # ... 格式化结果 ...
        return formatted_result

    except requests.exceptions.Timeout:
        logger.error(f"搜索请求超时: {query}")
        return "搜索请求超时,可能是网络较慢或服务繁忙,请稍后重试。"
    except requests.exceptions.ConnectionError:
        logger.error(f"网络连接错误: {query}")
        return "无法连接到搜索服务,请检查您的网络连接。"
    except requests.exceptions.HTTPError as e:
        logger.error(f"HTTP错误 {e.response.status_code}: {query}")
        if e.response.status_code == 429:
            return "搜索请求过于频繁,已被限流,请一分钟后再试。"
        elif e.response.status_code == 403:
            return "搜索权限验证失败,请联系管理员检查API密钥。"
        else:
            return f"搜索服务出错(状态码:{e.response.status_code})。"
    except ValueError as e:  # 包括JSON解析错误
        logger.error(f"响应数据解析错误: {e}, Query: {query}")
        return "搜索服务返回的数据格式有误,无法处理。"
    except Exception as e:
        # 捕获其他所有未预见的异常
        logger.exception(f"搜索过程中发生未预期错误: {e}, Query: {query}") # 记录完整堆栈
        return "搜索过程发生意外错误,我们的技术团队已收到通知。"

3.5 坑五:技能输出格式混乱,不利于后续处理

问题现象 : Skill返回的数据有时是字符串,有时是字典,有时是包含HTML标签的文本,导致Agent无法稳定地解析结果并生成流畅的回复,或者后续的Skill无法利用前一个Skill的输出。

根本原因 : 没有定义清晰的Skill输出契约(Contract)。输出格式随意,缺乏结构化。

避坑方案

  1. 标准化输出 : 定义统一的输出格式。对于简单信息,返回纯文本字符串。对于复杂信息,返回结构化的字典或Pydantic模型。
  2. 包含元数据 : 在输出中除了核心数据,还可以包含状态码( success , partial_success , error )、错误信息、数据来源等元数据。
  3. 为链式调用设计 : 如果Skill的输出是另一个Skill的输入,那么输出格式必须严格匹配下游Skill的输入格式要求。

代码示例(结构化输出)

# skills/structured_skill.py
from pydantic import BaseModel
from enum import Enum
from typing import List, Optional

class ResultStatus(str, Enum):
    SUCCESS = "success"
    PARTIAL = "partial_success"
    ERROR = "error"

class SkillOutput(BaseModel):
    """Skill统一输出模型"""
    status: ResultStatus
    data: Optional[str] = None  # 主要给用户看的结果文本
    structured_data: Optional[dict] = None  # 可供其他Skill或程序使用的结构化数据
    error_message: Optional[str] = None
    source: Optional[str] = None  # 数据来源,如“weather_api_v1”

def get_stock_price(symbol: str) -> SkillOutput:
    """获取股票价格,返回结构化结果"""
    try:
        # 模拟API调用
        # price_data = call_stock_api(symbol)
        price_data = {"symbol": symbol, "price": 150.25, "change": "+2.5%", "currency": "USD"}

        # 准备用户友好的文本
        user_message = f"{symbol} 当前股价为 {price_data['price']} {price_data['currency']},涨跌幅 {price_data['change']}。"

        return SkillOutput(
            status=ResultStatus.SUCCESS,
            data=user_message,
            structured_data=price_data,  # 其他Skill可以使用这个结构化的数据
            source="mock_stock_api"
        )
    except Exception as e:
        return SkillOutput(
            status=ResultStatus.ERROR,
            data=f"无法获取{symbol}的股价信息。",
            error_message=str(e),
            source="mock_stock_api"
        )

# 使用示例
result = get_stock_price("AAPL")
if result.status == ResultStatus.SUCCESS:
    print(result.data)  # 输出给用户
    # 其他Skill可以访问 result.structured_data['price']
else:
    print(f"操作失败:{result.error_message}")

3.6 坑六:缺乏可测试性与版本管理

问题现象 : 修改Skill后,不确定是否会影响原有功能;线上Skill出现问题,难以定位是哪个版本的更改引入的;团队协作时,Skill的变更混乱。

根本原因 : 将Skill视为一次性脚本,没有为其建立测试用例、版本控制和持续集成流程。

避坑方案

  1. 编写单元测试 : 为每个Skill的核心逻辑函数编写测试,覆盖正常路径、边界情况和异常情况。
  2. 使用版本控制 : 用Git等工具管理Skill代码,提交信息清晰描述变更内容。
  3. 建立技能仓库(Registry) : 对于大型项目,可以建立一个中心化的Skill仓库,对Skill进行注册、版本管理和依赖声明。
  4. 集成测试 : 测试Skill与Agent的集成效果,确保意图识别和参数传递正常。

代码示例(Skill单元测试)

# tests/test_weather_skill.py
import pytest
from unittest.mock import patch, Mock
from skills.weather_skill import get_weather, WeatherInput

class TestWeatherSkill:
    """测试天气查询Skill"""

    def test_get_weather_success(self):
        """测试正常情况下的天气查询"""
        # 模拟一个成功的API响应
        mock_response = Mock()
        mock_response.json.return_value = {
            "location": {"name": "Beijing"},
            "current": {"temp_c": 20, "condition": {"text": "Sunny"}}
        }
        mock_response.raise_for_status = Mock()

        with patch('skills.weather_skill.requests.get', return_value=mock_response):
            with patch.dict('os.environ', {'WEATHER_API_KEY': 'test_key'}):
                result = get_weather("Beijing")
                assert "Beijing" in result
                assert "20" in result  # 检查温度信息

    def test_get_weather_missing_api_key(self):
        """测试缺少API密钥时的降级处理"""
        with patch.dict('os.environ', {}, clear=True): # 清空环境变量
            result = get_weather("Shanghai")
            assert "模拟数据" in result  # 应该触发降级逻辑
            assert "Shanghai" in result

    def test_get_weather_network_error(self):
        """测试网络异常"""
        with patch('skills.weather_skill.requests.get', side_effect=requests.exceptions.ConnectionError):
            with patch.dict('os.environ', {'WEATHER_API_KEY': 'test_key'}):
                result = get_weather("London")
                assert "网络连接错误" in result or "抱歉" in result

    def test_weather_input_validation(self):
        """测试输入模型的校验"""
        # 测试空城市名
        with pytest.raises(ValueError):
            WeatherInput(city="   ")
        # 测试正常输入
        valid_input = WeatherInput(city="New York", date="tomorrow")
        assert valid_input.city == "New York"
        assert valid_input.date == "tomorrow"

4. 构建一个健壮的Skill:完整实战案例

现在,我们将综合运用上述避坑指南,从头构建一个“新闻摘要”Skill。这个Skill将调用新闻API,获取头条新闻,并利用LLM生成简洁摘要。

4.1 需求分析与设计

  • 功能 : 根据用户指定的主题(如“科技”、“体育”)或关键词,获取最新新闻并生成摘要。
  • 输入 : 主题(可选,有默认值)、关键词(可选)、返回条数(可选)。
  • 输出 : 结构化数据,包含新闻标题、摘要、来源链接和原始Skill状态。
  • 避坑点
    • 清晰的意图描述。
    • 参数校验与默认值。
    • 外部API调用异常处理。
    • 结构化输出。
    • 可测试性。

4.2 项目结构与依赖

确保项目结构清晰,在 requirements.txt 中增加新依赖。

# requirements.txt 新增
langchain-openai
tenacity

4.3 编写核心Skill代码

# skills/news_summary_skill.py
import os
import requests
import logging
from typing import Optional, List
from pydantic import BaseModel, Field, validator
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- 1. 定义清晰、强校验的输入模型 ---
class NewsSummaryInput(BaseModel):
    topic: Optional[str] = Field(
        default="general",
        description="新闻主题类别,例如:'technology'(科技), 'sports'(体育), 'business'(商业), 'entertainment'(娱乐)。默认为'general'(综合)。"
    )
    keyword: Optional[str] = Field(
        default=None,
        description="搜索新闻的关键词,例如:'人工智能', '世界杯'。如果提供,将优先按关键词搜索。"
    )
    max_results: int = Field(
        default=3,
        ge=1,
        le=10,
        description="返回新闻的最大数量,范围在1到10之间。默认为3。"
    )

    @validator('topic')
    def topic_must_be_valid(cls, v):
        valid_topics = ['general', 'technology', 'sports', 'business', 'entertainment', 'science', 'health']
        if v.lower() not in valid_topics:
            logger.warning(f"收到非常规主题'{v}',将使用'general'。")
            return 'general'
        return v.lower()

# --- 2. 定义结构化的输出模型 ---
class NewsArticle(BaseModel):
    title: str
    summary: str
    source_url: Optional[str]
    published_at: Optional[str]

class NewsSummaryOutput(BaseModel):
    status: str  # "success", "partial_success", "error"
    articles: List[NewsArticle] = []
    message: str
    source: str = "news_summary_skill_v1"

# --- 3. 核心函数,包含完整异常处理 ---
@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_news_from_api(topic: str, keyword: Optional[str] = None, max_results: int = 3) -> List[dict]:
    """从模拟新闻API获取新闻列表。实际应替换为真实API调用。"""
    # 此处为模拟数据。真实情况应调用如 NewsAPI, GNews 等。
    # 示例: response = requests.get(f"{NEWS_API_URL}?topic={topic}&q={keyword}&pageSize={max_results}")
    logger.info(f"Fetching news for topic='{topic}', keyword='{keyword}', max={max_results}")
    # 模拟API延迟和可能的失败
    import random
    if random.random() < 0.1:  # 模拟10%的失败率
        raise requests.exceptions.Timeout("Simulated API timeout")

    mock_articles = [
        {"title": "OpenAI发布新模型,多模态能力再升级", "content": "内容预览...", "url": "https://example.com/1", "publishedAt": "2024-05-27"},
        {"title": "量子计算取得突破性进展", "content": "内容预览...", "url": "https://example.com/2", "publishedAt": "2024-05-26"},
        {"title": "全球科技巨头财报陆续公布", "content": "内容预览...", "url": "https://example.com/3", "publishedAt": "2024-05-25"},
    ]
    # 简单模拟根据关键词过滤
    filtered_articles = mock_articles
    if keyword:
        filtered_articles = [a for a in mock_articles if keyword.lower() in a['title'].lower()]
    return filtered_articles[:max_results]

def generate_summary_with_llm(text: str) -> str:
    """使用LLM生成新闻摘要。"""
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.2, api_key=os.getenv("OPENAI_API_KEY"))
    prompt = ChatPromptTemplate.from_template("""
    请将以下新闻内容浓缩为一句话摘要,要求简洁、客观、包含核心事实:
    新闻内容:{news_content}
    一句话摘要:
    """)
    chain = prompt | llm | StrOutputParser()
    try:
        summary = chain.invoke({"news_content": text[:500]})  # 限制输入长度
        return summary.strip()
    except Exception as e:
        logger.error(f"LLM摘要生成失败: {e}")
        return "(摘要生成失败)"

def get_news_summary(topic: str = "general", keyword: Optional[str] = None, max_results: int = 3) -> NewsSummaryOutput:
    """
    新闻摘要Skill主函数。
    1. 获取新闻列表
    2. 为每条新闻生成摘要
    3. 返回结构化结果
    """
    articles_for_output = []
    error_messages = []

    try:
        # 步骤1: 获取新闻
        raw_articles = fetch_news_from_api(topic, keyword, max_results)

        if not raw_articles:
            return NewsSummaryOutput(
                status="success",
                articles=[],
                message=f"未找到关于'{keyword or topic}'的新闻。"
            )

        # 步骤2: 处理每条新闻
        for raw_article in raw_articles:
            try:
                summary = generate_summary_with_llm(raw_article.get('content', raw_article['title']))
                article = NewsArticle(
                    title=raw_article['title'],
                    summary=summary,
                    source_url=raw_article.get('url'),
                    published_at=raw_article.get('publishedAt')
                )
                articles_for_output.append(article)
            except Exception as e:
                error_msg = f"处理新闻'{raw_article.get('title')}'时出错: {e}"
                logger.warning(error_msg)
                error_messages.append(error_msg)
                # 降级处理:使用标题作为摘要
                articles_for_output.append(NewsArticle(
                    title=raw_article['title'],
                    summary=raw_article['title'],  # 降级
                    source_url=raw_article.get('url'),
                    published_at=raw_article.get('publishedAt')
                ))

        # 步骤3: 确定最终状态和消息
        if error_messages and len(articles_for_output) > 0:
            status = "partial_success"
            message = f"已获取{len(articles_for_output)}条新闻摘要,但部分处理过程遇到问题。"
        elif not articles_for_output:
            status = "error"
            message = "未能成功获取或处理任何新闻。"
        else:
            status = "success"
            message = f"成功获取并生成了{len(articles_for_output)}条新闻摘要。"

        if error_messages:
            message += " 内部错误详情已记录。"

        return NewsSummaryOutput(
            status=status,
            articles=articles_for_output,
            message=message
        )

    except requests.exceptions.RequestException as e:
        logger.exception("新闻API网络请求失败")
        return NewsSummaryOutput(
            status="error",
            articles=[],
            message=f"无法连接新闻服务,请检查网络或稍后重试。错误类型:{type(e).__name__}"
        )
    except Exception as e:
        logger.exception("新闻摘要Skill发生未预期错误")
        return NewsSummaryOutput(
            status="error",
            articles=[],
            message=f"新闻摘要服务暂时不可用。"
        )

# --- 4. 封装为LangChain Tool ---
from langchain.tools import Tool

news_summary_tool = Tool.from_function(
    func=get_news_summary,
    name="get_news_summary",
    description="根据主题或关键词获取最新的新闻,并生成简洁的一句话摘要。当用户想了解最新动态、获取新闻简报、跟踪某个话题时使用。",
    args_schema=NewsSummaryInput
)

4.4 集成到Agent并测试

# main.py
import os
from dotenv import load_dotenv
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from skills.news_summary_skill import news_summary_tool
from skills.weather_skill import weather_tool

load_dotenv()

def main():
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, api_key=os.getenv("OPENAI_API_KEY"))

    tools = [news_summary_tool, weather_tool]  # 可以加入更多Skill

    agent = initialize_agent(
        tools,
        llm,
        agent=AgentType.OPENAI_FUNCTIONS,  # 或 AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION
        verbose=True,  # 打印详细思考过程,便于调试
    )

    # 测试几个查询
    queries = [
        "给我总结一下今天科技方面的新闻。",
        "我想了解人工智能领域的最新动态,找3条新闻。",
        "今天北京的天气怎么样?",
        "先看看体育新闻,然后告诉我上海明天天气。"
    ]

    for query in queries:
        print(f"\n=== 用户查询: {query} ===")
        try:
            response = agent.run(query)
            print(f"Agent回复: {response}")
        except Exception as e:
            print(f"Agent执行出错: {e}")

if __name__ == "__main__":
    main()

运行上述 main.py ,你可以看到Agent如何理解用户意图,选择正确的Skill( get_news_summary get_weather ),并执行它们。 verbose=True 会展示Agent的思考链,这对于调试Skill的匹配度非常有帮助。

5. Skill编写自检清单与最佳实践

在完成一个Skill的编写后,或者当Skill出现问题时,可以使用以下清单进行排查和优化。

5.1 Skill上线前自检清单

检查项 是/否 说明与改进建议
意图清晰度 Skill的 description 是否清晰、具体,覆盖了用户可能的表达方式?是否包含示例场景?
参数定义 所有参数是否都有清晰的 description ?是否设置了合理的默认值?是否进行了有效性校验(非空、类型、范围)?
单一职责 这个Skill是否只做一件明确的事情?如果功能超过一个,考虑拆分成多个Skill。
异常处理 是否对所有外部调用(API、DB、文件)进行了try-catch?是否将内部异常转换成了用户友好的消息?是否记录了详细的错误日志?
输出格式 Skill的输出是否稳定且结构化?是否便于其他Skill或主程序使用?是否包含了成功/失败状态?
依赖管理 Skill依赖的第三方库版本是否固定?API密钥等敏感信息是否通过环境变量管理?
性能考量 是否有耗时的操作?是否可以考虑异步或缓存?是否有超时设置?
测试覆盖 是否编写了单元测试,覆盖主流程、边界情况和异常情况?测试用例能否通过?

5.2 工程化最佳实践

  1. 配置化 : 将API端点、超时时间、重试次数、默认值等配置项提取到配置文件或环境变量中,避免硬编码。
  2. 日志标准化 : 使用结构化的日志格式(如JSON),统一记录Skill的调用开始、结束、输入、输出、耗时和错误,便于监控和排查。
  3. 监控与指标 : 为Skill添加关键指标,如调用次数、成功率、平均耗时、错误类型分布。这能帮助你发现性能瓶颈和潜在问题。
  4. 版本化与回滚 : 对Skill进行版本管理。当新版本Skill上线后出现问题,能快速回滚到旧版本。
  5. 文档化 : 为每个Skill编写清晰的文档,包括功能描述、输入输出格式、示例、错误码、依赖和更新日志。
  6. 技能仓库 : 当Skill数量增多时,建立内部技能仓库,实现Skill的发现、注册、版本管理和依赖解析。

编写高质量的Skill是构建强大AI Agent的基石。它要求开发者不仅要有扎实的编程能力,更要有产品思维和用户体验意识,能够预见到各种边界情况和失败模式。通过避开本文所述的六个常见陷阱,并遵循结构化的开发、测试和部署流程,你将能构建出稳定、可靠、易维护的Skill,从而让你的AI Agent真正具备解决复杂问题的能力。

更多推荐