Agent Skill开发指南：从零开始编写高效技能模块

4次阅读

没有评论

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

如果把 Agent 比作一个餐厅，Skill 就是后厨的各个工作站。不同于插件（Plugin）这种即插即用的调味料，Skill 更像是具备完整处理流程的烹饪台：

预处理：像洗菜台处理原始食材，完成输入清洗
主逻辑：像灶台核心烹饪区，处理业务逻辑
后处理：像摆盘区，格式化输出结果

与中间件（Middleware）相比，Skill 的特点是：

垂直领域聚焦（一个 Skill 解决一类问题）
完整的输入输出闭环
可独立测试的单元模块

主流框架的技能存放位置对比：

Rasa：skills/ 目录下按功能分包

project/
├── skills/
│   ├── weather/
│   │   ├── __init__.py
│   │   └── actions.py
│   └── payment/
├── domain.yml  # 技能注册文件

Dialogflow：通过 package.json 定义技能入口

{
  "actions": {
    "skill1": {"entryPoint": "skills.skill1.main"}
  }
}

以 Rasa 的 domain.yml 为例：

skills:
  weather_check:
    triggers:
      - intent: ask_weather
    actions:
      - action_query_weather
    parameters:
      location: !slot user_location

from typing import Dict, Text, Any

class WeatherSkill:
    def __init__(self):
        self.api_client = WeatherAPI()

    def execute(self, tracker: Dict) -> Dict[Text, Any]:
        """处理天气查询意图"""
        location = tracker.get_slot("user_location")
        return {"temp": self.api_client.query(location),
            "unit": "celsius"
        }

def context_preserve(func):
    @wraps(func)
    def wrapper(tracker, *args, **kwargs):
        # 保存当前对话状态
        old_context = tracker.export_state()
        result = func(tracker, *args, **kwargs)
        # 恢复关键上下文
        tracker.update_slot("last_intent", old_context["last_intent"])
        return result
    return wrapper

@context_preserve
def handle_complex_query(tracker):
    # 多步骤处理逻辑
    ...

错误示范：

# skill1.py 和 skill2.py 中都有
class Utils: ...

解决方案：

# skill1/utils.py 和 skill2/helpers.py

危险代码：

async def update_user_profile(user):
    profile = await get_profile(user)  # 竞态条件风险
    profile.credits += 10

安全写法：

from asyncio import Lock

credit_lock = Lock()

async def safe_update(user):
    async with credit_lock:
        profile = await get_profile(user)
        profile.credits += 10

易受攻击的代码：

def sql_query(user_input):
    cursor.execute(f"SELECT * FROM users WHERE name='{user_input}'")  # SQL 注入

防护方案：

from pydantic import BaseModel, constr

class UserQuery(BaseModel):
    name: constr(max_length=50, regex=r"^[a-zA-Z]+$")

def safe_query(query: UserQuery):
    cursor.execute("SELECT * FROM users WHERE name=%s", (query.name,))

class SkillLoader:
    _preloaded = {}

    @classmethod
    def preload(cls):
        cls._preloaded["weather"] = WeatherSkill()
        cls._preloaded["payment"] = PaymentSkill()

# 服务启动时调用
SkillLoader.preload()

from functools import lru_cache

@lru_cache(maxsize=10)
def get_skill(name: str):
    if name == "weather":
        return WeatherSkill()
    # 其他技能初始化...

TODO 任务：改造以下天气查询技能，使其支持连续追问：

# 当前代码（单轮）class WeatherSkill:
    def execute(self, tracker):
        if "city" not in tracker.slots:
            return {"ask": "请问您想查询哪个城市？"}
        # 原始查询逻辑...

改造要求：
1. 当用户回答城市后，追问查询日期
2. 根据日期返回不同的天气数据格式
3. 使用 context_preserve 装饰器维护状态

特性	Rasa	Dialogflow
技能定义方式	YAML+Python	JSON+Node.js
上下文管理	Tracker 对象	Session 存储
性能优化	自定义预加载	自动冷启动
适合场景	复杂业务逻辑	快速原型开发