Agent技能开发入门指南：从零构建你的第一个Skill

共计 1977 个字符，预计需要花费 5 分钟才能阅读完成。

什么是 Agent Skill？

在 Agent 开发体系中，Skill 是完成特定任务的最小功能单元（如查天气、订机票）。它与 Action 的区别在于：
– Action 是原子操作（如调用一次 API）
– Skill 是包含完整业务逻辑的服务（如处理用户问询→调用 API→格式化回复）
– Flow 则是多个 Skill 的编排组合（如订酒店→租车→行程提醒）

新手开发三大痛点

1. 接口规范不符：Agent 框架要求技能必须实现特定生命周期方法（如initialize、execute）
2. 状态管理混乱：未正确处理对话上下文导致多轮对话中断
3. 异常处理缺失：网络超时、API 限流等场景下直接崩溃

天气查询 Skill 实战

以下是一个 Python 实现的完整示例（使用 Rasa NLU 进行意图识别）：

from agent_framework.skill import BaseSkill
from agent_framework.models import Message, Response
import requests

class WeatherSkill(BaseSkill):
    def __init__(self):
        # 技能注册时必须声明支持的意图
        self.supported_intents = ["query_weather"]

    async def execute(self, msg: Message) -> Response:
        """
        核心业务逻辑：1. 解析用户消息中的城市参数
        2. 调用天气 API
        3. 构造自然语言响应
        """
        try:
            # 从 NLU 结果提取实体
            city = next((e['value'] for e in msg.entities 
                 if e['entity'] == "city"),
                None
            )

            if not city:
                return Response(
                    text="请告诉我您想查询哪个城市的天气",
                    need_followup=True  # 需要用户补充信息
                )

            # 调用第三方 API（示例使用假想天气服务）resp = requests.get(f"https://api.weather.example?city={city}",
                timeout=3
            )
            data = resp.json()

            # 构造自然语言响应
            return Response(text=f"{city}当前天气：{data['temp']}℃，{data['condition']}",
                session=msg.session  # 保持对话上下文
            )

        except requests.Timeout:
            return Response("天气服务暂时不可用，请稍后再试")
        except Exception as e:
            self.logger.error(f"天气查询失败: {e}")
            return Response("遇到了一点技术问题，正在紧急修复中...")

# 注册到 Agent 系统（框架提供）register_skill(WeatherSkill())

关键代码说明：
1. 意图声明 ：supported_intents 定义技能处理的用户意图
2. 消息处理 ：execute() 是必须实现的入口方法
3. 上下文保持 ：通过session 字段实现多轮对话
4. 健壮性设计：对网络请求进行超时捕获和降级处理

四大避坑指南

1. 幂等性设计
2. 对相同参数请求返回缓存结果

3. 使用 request_id 防止重复处理

   # 在 execute 方法开头添加
   if msg.metadata.get('is_retry'):
       return cached_response

4. 上下文管理

5. 短期记忆：使用 session 存储当前对话状态

6. 长期记忆：通过用户 ID 关联数据库

   # 存储上一轮选择的城市
   msg.session["last_city"] = city

7. 性能优化

8. 冷启动优化：异步加载资源

   async def initialize(self):
       self.cities = await load_city_data_async()  

9. 并发控制：限制最大并行请求数

   from asyncio import Semaphore
   self.sem = Semaphore(10)  # 最大 10 并发
   
   async with self.sem:
       resp = await call_api()

扩展思考：多轮对话实现

要实现支持多轮对话的 Skill（如酒店预订），建议：
1. 定义明确的对话状态机（如 等待选择日期→等待确认价格）
2. 使用上下文补全缺失参数：

if not msg.entities.get('date'):
    return Response(
        "您想预订哪天的房间？",
        expecting="date"  # 声明期待的参数类型
    )

3. 设置对话超时（通常 5 -10 分钟未响应则重置状态）

通过这个天气查询 Skill 的实例，你应该已经掌握了 Agent 技能开发的核心模式。接下来可以尝试更复杂的场景，比如需要调用多个 API 的机票比价 Skill，记得始终遵循：明确意图声明→健壮的业务逻辑→完善的上下文管理 这三个原则。