Dify支持Skill开发实战：从零构建你的第一个AI技能

1次阅读

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

Dify 作为新一代 AI 开发平台，通过 Skill 机制将 AI 能力模块化。其核心价值体现在：

低代码集成：提供可视化编排工具，减少底层 API 调用复杂度
技能市场生态：支持开发者发布 / 共享 Skill，形成能力闭环
统一接口规范：标准化输入输出格式，降低对接成本
动态加载机制：支持运行时热更新技能，无需停机部署

注册 Dify 开发者账号（需企业邮箱认证）
安装 Python 3.8+ 和 Node.js 16+ 环境
准备 Dify CLI 工具包：

pip install dify-client --upgrade
dify init --profile developer

# 创建 Skill 脚手架
from dify.skills import SkillBuilder

weather_skill = SkillBuilder(
    name="weather_query",
    version="1.0.0",
    description="城市天气查询技能"
)

# skill.yaml
inputs:
  - name: "city"
    type: "string"
    required: true
    description: "查询城市名称"

outputs:
  - name: "weather"
    type: "object"
    properties:
      temp: "float"
      condition: "string"

@weather_skill.handler
async def query_weather(params):
    # 对接第三方天气 API
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.weather.com/v3?city={params['city']}"
        )
        return {"temp": resp.json()["current_temp"],
            "condition": resp.json()["weather_desc"]
        }

dify skill test --file weather_skill.py

打包技能组件：dify skill build --output dist/
上传至 Dify 中心仓库：dify skill publish --version 1.0.0
平台审核通过后即可在技能市场启用

import httpx
from dify.skills import SkillBuilder, SkillMetadata

meta = SkillMetadata(
    name="weather_pro",
    category="utility",
    timeout=5000  # 毫秒级超时设置
)

weather = SkillBuilder(meta)

@weather.input_validator
def validate_city(city: str):
    if not city.isalpha():
        raise ValueError("城市名只能包含字母")
    return city.lower()

@weather.handler
async def handle_query(params: dict):
    cache_key = f"weather_{params['city']}"
    if cached := await redis.get(cache_key):
        return json.loads(cached)

    async with httpx.AsyncClient(
        timeout=10.0,
        headers={"Authorization": f"Bearer {API_KEY}"}
    ) as client:
        resp = await client.get(
            "https://api.weatherapi.com/v1/current.json",
            params={"q": params["city"]}
        )
        result = {"temp": resp.json()["current"]["temp_c"],
            "condition": resp.json()["current"]["condition"]["text"]
        }
        await redis.setex(cache_key, 3600, json.dumps(result))
        return result

flowchart TD
    A[用户请求] --> B{Dify 路由}
    B -->| 匹配技能 | C[Weather Skill]
    C --> D[输入校验]
    D --> E[缓存检查]
    E -->| 命中 | F[返回缓存]
    E -->| 未命中 | G[调用天气 API]
    G --> H[结果格式化]
    H --> I[缓存写入]
    I --> J[返回响应]