从零构建智能体技能：解锁AI自动化工作流的核心实践

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

在实际的 AI 应用开发中，我们常常会遇到一个困境：模型能力强大，但如何让它稳定、可靠地执行我们设定的复杂任务？尤其是在处理需要多步骤、条件判断、外部工具调用的自动化流程时，简单的提示词往往力不从心。这时，一个结构化的“技能”或“工具”调用框架就显得至关重要。无论是 Codex、Claude 的 Code Skills，还是 Dify、Coze 等智能体开发平台，其核心进阶玩法都围绕着如何定义、编排和复用这些“Skills”来构建强大的自动化工作流。

如果你已经安装好了相关平台或工具，却感觉它只是个聊天界面，无法实现真正的自动化，那问题很可能出在没有掌握“Skills”的运用。本文将带你深入理解智能体技能（Skills）的本质，从零开始构建一个可复用的自动化工作流。我们将聚焦于通用概念和设计模式，无论你使用哪个具体平台，这些核心思路都能帮助你吃透智能体技能，告别“装好不会用”的尴尬。

1. 理解智能体技能：从“聊天”到“执行”的跨越

在深入实操之前，我们必须先厘清几个核心概念。这能帮助你理解为什么需要 Skills，而不是仅仅依赖模型的原始文本生成能力。

1.1 什么是智能体技能？

通俗地讲，智能体技能是赋予大语言模型“动手能力”的扩展模块。模型本身擅长理解和生成文本，但它无法直接操作数据库、发送邮件、查询天气或执行一段代码。技能就是连接模型“思考”与外部世界“行动”的桥梁。它通常由三部分组成：

1. 技能描述：用自然语言告诉模型这个技能是做什么的，何时使用它。
2. 输入参数：定义执行该技能需要哪些信息，例如城市名称、日期、查询关键词等。
3. 执行逻辑：一段实际的代码（如 Python、JavaScript）或一个可调用的 API 接口，用于完成具体任务。

当用户提出“帮我查一下北京明天的天气，然后发邮件提醒我带伞”这样的请求时，一个具备get_weather和send_email技能的智能体，会先调用天气查询技能获取数据，再组织内容调用邮件发送技能，从而完成整个工作流。

1.2 Skills、Tools、Actions 与 Harness：概念辨析

在不同的平台和语境下，类似的概念可能有不同名称，但内核一致：

• Skills / Tools / Actions：在大多数上下文中可以互换使用，都指代智能体可以调用的具体功能单元。例如 Claude 的 “Code Skills”， OpenAI 的 “Function Calling”（工具调用）。
• Harness：这个词在某些智能体框架中，指的是一种更高级的封装或“套件”，它可能集成了多个相关的 Skills，并提供统一的管理、上下文保持或资源分配。你可以把它理解为一个“技能包”或“工具箱”。例如，一个“数据分析 Harness”可能包含了数据读取、清洗、可视化和报告生成等多个技能。

理解这些术语有助于你阅读不同平台的文档，其核心思想都是将复杂能力模块化，并通过标准化接口暴露给大语言模型调用。

1.3 为什么 Skills 是构建自动化工作流的关键？

没有 Skills 的智能体，其能力被禁锢在单次对话的文本生成内。而引入了 Skills，意味着：

• 确定性操作：发送邮件、写入数据库等操作的结果是确定且可验证的，弥补了大模型输出可能存在的“幻觉”。
• 连接外部系统：可以与企业内部的 CRM、ERP，或公网的各类 API 服务连接，打破信息孤岛。
• 实现复杂逻辑：通过编排多个 Skills 的调用顺序和条件分支，可以实现如“如果A则执行B，否则执行C”的复杂业务流程。
• 能力复用：一个定义好的query_database技能，可以被客服、报表、风控等多个不同的智能体复用，提升开发效率。

2. 环境准备与核心工具选择

在开始构建技能之前，你需要一个可以运行和测试技能的环境。由于输入材料中提到了多个平台和工具，我们将从通用性角度出发，搭建一个最小化的本地开发测试环境。

2.1 基础环境配置

我们将使用 Python 作为技能开发的主要语言，因为它有丰富的库支持和广泛的社区认可。同时，我们会模拟一个类似 n8n 或 Dify 的本地工作流编排场景。

1. 安装 Python 和包管理工具确保你的系统已安装 Python 3.8 或更高版本。推荐使用venv创建虚拟环境以隔离依赖。

# 检查Python版本 python3 --version # 创建并激活虚拟环境（以项目目录 `ai_agent_skills` 为例） mkdir ai_agent_skills && cd ai_agent_skills python3 -m venv venv # 在 macOS/Linux 上激活 source venv/bin/activate # 在 Windows 上激活 # venv\Scripts\activate

2. 安装核心依赖我们将安装openai库（用于模拟与大模型的交互）和fastapi、uvicorn（用于将技能暴露为 API，这是大多数智能体平台调用技能的通用方式）。

pip install openai fastapi uvicorn pydantic requests

2.2 模拟智能体调度框架

为了清晰地演示技能的定义、注册和调用流程，我们不直接依赖某个特定商业平台，而是自己编写一个极简的模拟框架。这能让你透彻理解底层机制。

在你的项目根目录下，创建以下文件结构：

ai_agent_skills/ ├── skills/ # 存放所有技能模块 │ ├── __init__.py │ └── weather.py # 示例：天气查询技能 ├── agent_core.py # 模拟智能体核心，负责调度技能 ├── main.py # FastAPI 应用入口 └── requirements.txt

requirements.txt内容如下：

openai>=1.0.0 fastapi>=0.104.0 uvicorn[standard]>=0.24.0 pydantic>=2.0.0 requests>=2.31.0

3. 从零定义你的第一个智能体技能

让我们以实现一个“天气查询”技能为例，完整走通技能创建、描述、参数定义和执行逻辑的流程。

3.1 设计技能接口：描述与参数

在skills/weather.py中，我们首先定义这个技能。一个好的技能定义需要清晰说明其功能、输入和输出。

# skills/weather.py import requests from pydantic import BaseModel, Field from typing import Optional # 1. 定义技能的输入参数模型 class WeatherQueryInput(BaseModel): """天气查询技能的输入参数""" city: str = Field(description="要查询天气的城市名称，例如：北京、Shanghai") date: Optional[str] = Field(default="today", description="查询日期，格式为 'today' 或 'YYYY-MM-DD'，默认为今天") # 2. 定义技能的输出模型 class WeatherQueryOutput(BaseModel): """天气查询技能的输出结果""" city: str date: str weather: str temperature: str humidity: Optional[str] = None wind: Optional[str] = None # 3. 技能执行函数 def get_weather(query: WeatherQueryInput) -> WeatherQueryOutput: """ 根据城市和日期查询天气信息。 这是一个模拟函数，实际项目中应接入真实的天气API（如和风天气、OpenWeatherMap）。 """ # 模拟API调用和数据处理 # 注意：此处为演示，直接返回模拟数据。真实场景需处理网络请求、错误和API密钥。 mock_data = { "北京": {"today": {"weather": "晴", "temp": "22°C/12°C", "humidity": "45%", "wind": "北风3级"}}, "上海": {"today": {"weather": "多云", "temp": "25°C/18°C", "humidity": "65%", "wind": "东南风2级"}}, } city_data = mock_data.get(query.city, {}) date_key = query.date if query.date in city_data else "today" weather_info = city_data.get(date_key, {"weather": "未知", "temp": "N/A", "humidity": "N/A", "wind": "N/A"}) return WeatherQueryOutput( city=query.city, date=query.date, weather=weather_info["weather"], temperature=weather_info["temp"], humidity=weather_info.get("humidity"), wind=weather_info.get("wind") ) # 4. 技能的元数据，用于告知智能体此技能的存在和用法 skill_metadata = { "name": "get_weather", "description": "查询指定城市在指定日期的天气情况，包括天气状况、温度、湿度和风力。", "input_model": WeatherQueryInput, "function": get_weather }

关键点解释：

• BaseModel：使用 Pydantic 模型来严格定义输入输出的数据结构，这能自动进行数据验证和类型转换。
• Field(description=...)：这里的描述至关重要！大语言模型（LLM）正是依靠这些描述来理解何时以及如何使用该技能。
• 模拟数据：在学习和测试阶段，使用模拟数据可以避免依赖外部 API 密钥和网络稳定性，快速验证流程。生产环境必须替换为真实的 API 调用，并加入完善的错误处理（如网络超时、API 限流、无效城市名等）。

3.2 构建技能注册与管理中心

智能体需要知道它有哪些技能可用。我们在agent_core.py中创建一个简单的技能注册中心。

# agent_core.py from typing import Dict, Any, Callable from pydantic import BaseModel class Skill: """技能封装类""" def __init__(self, name: str, description: str, input_model: BaseModel, function: Callable): self.name = name self.description = description self.input_model = input_model self.function = function def execute(self, input_data: Dict[str, Any]): """执行技能，并验证输入参数""" # 使用Pydantic模型验证输入 validated_input = self.input_model(**input_data) # 调用实际的功能函数 result = self.function(validated_input) return result class SkillRegistry: """技能注册表（单例模式）""" _instance = None _skills: Dict[str, Skill] = {} def __new__(cls): if cls._instance is None: cls._instance = super(SkillRegistry, cls).__new__(cls) return cls._instance @classmethod def register_skill(cls, name: str, description: str, input_model, function): """注册一个技能""" cls._skills[name] = Skill(name, description, input_model, function) print(f"[Skill Registered] {name}: {description}") @classmethod def get_skill(cls, name: str) -> Skill: """根据名称获取技能""" return cls._skills.get(name) @classmethod def list_skills(cls) -> Dict[str, str]: """列出所有可用技能及其描述""" return {name: skill.description for name, skill in cls._skills.items()} # 提供一个便捷的装饰器来注册技能 def skill(name: str, description: str): """技能注册装饰器""" def decorator(func): # 这里假设被装饰函数的第一个参数是Pydantic模型实例 # 在实际框架中，可能需要更复杂的方式来获取input_model # 为简化，我们要求技能函数必须有一个类型注解为BaseModel的参数 from inspect import signature sig = signature(func) params = list(sig.parameters.values()) if not params: raise ValueError(f"Skill function '{func.__name__}' must have at least one parameter for input.") input_model = params[0].annotation SkillRegistry.register_skill(name, description, input_model, func) return func return decorator

这个注册中心管理了所有技能，并提供了统一的执行接口execute。@skill装饰器让技能的声明式注册变得非常简洁。

3.3 将技能暴露为 API 端点

大多数智能体平台通过 HTTP API 调用技能。我们使用 FastAPI 快速创建一个服务。在main.py中：

# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Dict, Any import skills.weather # 导入技能模块以触发注册 from agent_core import SkillRegistry app = FastAPI(title="AI Agent Skills API") # 自动导入 skills 目录下所有模块，实现自动注册 import os import importlib skills_dir = "skills" for filename in os.listdir(skills_dir): if filename.endswith(".py") and filename != "__init__.py": module_name = f"skills.{filename[:-3]}" importlib.import_module(module_name) class SkillExecuteRequest(BaseModel): skill_name: str input: Dict[str, Any] @app.get("/") async def root(): return {"message": "AI Agent Skills API is running"} @app.get("/skills") async def list_skills(): """列出所有已注册的技能""" return SkillRegistry.list_skills() @app.post("/execute") async def execute_skill(request: SkillExecuteRequest): """执行指定技能""" skill = SkillRegistry.get_skill(request.skill_name) if not skill: raise HTTPException(status_code=404, detail=f"Skill '{request.skill_name}' not found.") try: result = skill.execute(request.input) # 将Pydantic模型输出转换为字典 return result.dict() except Exception as e: raise HTTPException(status_code=500, detail=f"Skill execution failed: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

现在，我们的技能服务已经就绪。它提供了两个关键接口：

1. GET /skills：查看所有可用技能。
2. POST /execute：通过传入技能名和输入参数来执行某个技能。

3.4 运行与验证你的第一个技能

1. 启动技能 API 服务：

   python main.py

   看到Uvicorn running on http://0.0.0.0:8000即表示启动成功。

2. 验证技能列表： 打开浏览器或使用curl访问http://localhost:8000/skills，你应该看到：

   {"get_weather": "查询指定城市在指定日期的天气情况，包括天气状况、温度、湿度和风力。"}

3. 测试技能执行： 使用curl或 Postman 等工具发送 POST 请求：

   curl -X POST "http://localhost:8000/execute" \ -H "Content-Type: application/json" \ -d '{ "skill_name": "get_weather", "input": { "city": "北京", "date": "today" } }'

   预期返回结果：

   { "city": "北京", "date": "today", "weather": "晴", "temperature": "22°C/12°C", "humidity": "45%", "wind": "北风3级" }

至此，你已经成功定义、注册并通过 API 暴露了一个完整的智能体技能。这是所有自动化工作流最基础的砖石。

4. 编排多个技能，构建自动化工作流

单一技能价值有限，真正的威力在于将多个技能串联起来，形成自动化工作流。例如，“查询天气 -> 生成出行建议 -> 发送邮件提醒”。

4.1 设计一个连贯的工作流：天气邮件提醒

我们新增两个技能：

1. generate_advice：根据天气信息生成出行建议。
2. send_email（模拟）：发送邮件。

首先，在skills目录下创建advice.py和email.py。

# skills/advice.py from pydantic import BaseModel, Field from agent_core import skill class AdviceInput(BaseModel): """生成建议的输入参数""" weather_info: dict = Field(description="包含天气详情的字典，通常来自get_weather技能的输出") user_preference: str = Field(default="通用", description="用户偏好，如‘怕冷’、‘户外活动’等") class AdviceOutput(BaseModel): """建议输出""" advice: str @skill(name="generate_advice", description="根据天气信息和用户偏好，生成具体的出行或生活建议。") def generate_advice(query: AdviceInput) -> AdviceOutput: weather = query.weather_info.get('weather', '') temp = query.weather_info.get('temperature', '') preference = query.user_preference base_advice = f"天气{weather}，温度{temp}。" if "雨" in weather: advice_text = base_advice + "建议携带雨具。" elif "晴" in weather and int(temp.split('°')[0]) > 28: advice_text = base_advice + "紫外线较强，请注意防晒。" else: advice_text = base_advice + "天气适宜，可正常安排出行。" if "怕冷" in preference: advice_text += " 您比较怕冷，建议比常人多添一件外套。" elif "户外" in preference: advice_text += " 适合进行户外活动。" return AdviceOutput(advice=advice_text)

# skills/email.py from pydantic import BaseModel, Field, EmailStr from agent_core import skill class EmailInput(BaseModel): """发送邮件的输入参数""" to_email: EmailStr = Field(description="收件人邮箱地址") subject: str = Field(description="邮件主题") content: str = Field(description="邮件正文内容") class EmailOutput(BaseModel): """邮件发送结果""" status: str message_id: str = None @skill(name="send_email", description="向指定的邮箱地址发送一封邮件。此为模拟功能。") def send_email(query: EmailInput) -> EmailOutput: # 模拟邮件发送逻辑 # 真实场景应集成SMTP服务（如smtplib）或邮件发送API（如SendGrid, AWS SES） print(f"[模拟邮件发送] 发往: {query.to_email}") print(f" 主题: {query.subject}") print(f" 内容: {query.content}") print("-" * 40) # 假设发送成功 return EmailOutput(status="success", message_id="mock_msg_123456")

4.2 实现工作流编排器

现在，我们需要一个“大脑”来协调这些技能。在项目根目录创建workflow_orchestrator.py。这个编排器将模拟大语言模型的决策过程：理解用户意图，规划技能调用顺序，并传递数据。

# workflow_orchestrator.py import json from agent_core import SkillRegistry from typing import Dict, Any, List class WorkflowOrchestrator: """一个简单的工作流编排器，模拟LLM的规划和执行逻辑""" def __init__(self): self.skill_registry = SkillRegistry def plan_and_execute(self, user_request: str) -> List[Dict[str, Any]]: """ 根据用户请求，规划并执行一系列技能。 这是一个高度简化的版本，真实LLM会基于技能描述进行动态规划。 """ execution_log = [] # 1. 技能规划（此处为硬编码逻辑，真实场景由LLM根据技能描述动态生成） if "天气" in user_request and ("邮件" in user_request or "提醒" in user_request): plan = [ {"skill": "get_weather", "input": {"city": "北京", "date": "today"}}, {"skill": "generate_advice", "input": {"user_preference": "通用"}}, {"skill": "send_email", "input": {"to_email": "user@example.com", "subject": "天气提醒", "content": ""}} ] else: # 简单回退：如果请求无法识别，尝试直接调用一个通用技能或返回错误 return [{"error": f"无法理解或处理请求: {user_request}"}] # 2. 按顺序执行规划好的技能 previous_output = None for step in plan: skill_name = step["skill"] skill_input = step["input"].copy() # 复制一份输入，避免修改原计划 # 关键：将上一个技能的输出，作为下一个技能的输入的一部分 if previous_output and skill_name == "generate_advice": skill_input["weather_info"] = previous_output if previous_output and skill_name == "send_email": # 假设previous_output现在是advice_output if isinstance(previous_output, dict) and 'advice' in previous_output: skill_input["content"] = f"您好，这是为您生成的天气建议：\n\n{previous_output['advice']}" print(f"[Orchestrator] 执行技能: {skill_name}, 输入: {skill_input}") skill_obj = self.skill_registry.get_skill(skill_name) if not skill_obj: execution_log.append({"step": skill_name, "status": "error", "detail": "Skill not found"}) break try: result = skill_obj.execute(skill_input) result_dict = result.dict() if hasattr(result, 'dict') else result execution_log.append({"step": skill_name, "status": "success", "output": result_dict}) previous_output = result_dict except Exception as e: execution_log.append({"step": skill_name, "status": "error", "detail": str(e)}) break # 一个步骤失败，终止工作流 return execution_log if __name__ == "__main__": # 导入所有技能模块以完成注册 import skills.weather import skills.advice import skills.email orchestrator = WorkflowOrchestrator() user_request = "查一下北京今天的天气，然后给我发个邮件提醒" print(f"用户请求: {user_request}") log = orchestrator.plan_and_execute(user_request) print("\n工作流执行日志:") print(json.dumps(log, indent=2, ensure_ascii=False))

运行这个编排器：

python workflow_orchestrator.py

你将看到控制台输出完整的技能调用链日志，以及模拟的邮件发送信息。这演示了如何将多个独立技能串联成一个自动化流程。

4.3 工作流编排的关键模式

在实际的智能体平台（如 n8n, Dify, Coze）中，编排通常通过可视化界面或更高级的 DSL（领域特定语言）完成，但其核心模式不变：

1. 顺序执行：技能 A -> 技能 B -> 技能 C。
2. 条件分支：如果技能 A 的输出满足条件 X，则执行技能 B，否则执行技能 C。
3. 循环迭代：对列表中的每一项，执行某个技能。
4. 并行执行：同时执行多个独立技能，然后聚合结果。
5. 错误处理：某个技能执行失败时，是重试、跳过还是终止整个工作流。

理解这些模式，你就能设计出复杂的业务自动化流程，如自动生成日报、监控告警并通知、客户咨询自动分类处理等。

5. 进阶：技能开发与集成的核心要点

掌握了基础技能和工作流后，要将其用于实际项目，还需要关注以下几个关键方面。

5.1 技能设计的通用原则

5.2 连接真实外部服务

我们的示例使用了模拟数据。在生产环境中，你需要替换为真实的 API 调用。以get_weather技能为例，接入真实天气 API：

# skills/real_weather.py import requests from pydantic import BaseModel, Field from typing import Optional import os class RealWeatherInput(BaseModel): city: str = Field(description="城市名称，例如：北京") # 可能还需要城市ID、经纬度等，取决于API class RealWeatherOutput(BaseModel): city: str weather: str temp: str feels_like: str humidity: str wind: str update_time: str def get_real_weather(query: RealWeatherInput) -> RealWeatherOutput: api_key = os.getenv("WEATHER_API_KEY") # 从环境变量读取密钥 if not api_key: raise ValueError("WEATHER_API_KEY 环境变量未设置") # 示例：使用和风天气API（需要注册获取key） url = f"https://devapi.qweather.com/v7/weather/now?location={query.city}&key={api_key}" try: response = requests.get(url, timeout=10) response.raise_for_status() # 检查HTTP错误 data = response.json() if data['code'] != '200': raise Exception(f"Weather API error: {data['message']}") now = data['now'] return RealWeatherOutput( city=query.city, weather=now['text'], temp=now['temp'], feels_like=now['feelsLike'], humidity=now['humidity'], wind=f"{now['windDir']} {now['windScale']}级", update_time=data['updateTime'] ) except requests.exceptions.Timeout: raise Exception("天气API请求超时") except requests.exceptions.RequestException as e: raise Exception(f"网络请求失败: {e}") except KeyError as e: raise Exception(f"解析天气API响应失败，字段缺失: {e}")

关键点：

• 密钥管理：永远不要将 API 密钥硬编码在代码中。使用环境变量或专业的密钥管理服务。
• 超时设置：为网络请求设置合理的超时时间，避免技能调用长时间阻塞。
• 错误处理：全面处理网络异常、API 返回错误、数据解析错误等。
• 重试机制：对于暂时性网络故障，可以考虑加入重试逻辑。

5.3 在主流智能体平台中使用技能

理解了技能的本质后，将其迁移到具体平台就非常直观。以下是通用思路：

1. Dify / Coze / 扣子：

   • 在平台的“工具”或“技能”配置页面，选择“自定义 API”或“HTTP 请求”。
   • 填入你本地或部署好的技能 API 地址（如http://your-server:8000/execute）。
   • 根据平台要求，描述技能、定义输入参数（通常通过 JSON Schema）。
   • 平台会在后台将你的技能描述注入给 LLM，并在需要时调用你的 API。

2. Claude Code Skills / OpenAI Function Calling：

   • 你需要按照其规定的格式（通常是 JSON Schema）来描述你的函数（技能）。
   • 在对话中，当 LLM 认为需要调用你的函数时，它会暂停文本生成，返回一个特殊的函数调用请求。
   • 你的程序需要捕获这个请求，执行对应的本地函数或调用你的 API，然后将结果以特定格式返回给 LLM，LLM 再基于结果继续生成回复。

3. n8n：

   • n8n 本身就是一个强大的工作流自动化工具。你可以将每个技能实现为一个独立的“HTTP Request”节点或自定义代码节点。
   • 然后使用 n8n 的图形化界面来编排这些节点的执行顺序、条件分支和数据流转。

6. 常见问题排查与最佳实践

6.1 技能调用失败排查清单

当你的技能无法正常工作时，可以按照以下顺序排查：

6.2 生产环境最佳实践

1. 技能服务化与部署：

   • 将技能 API 部署为独立的微服务，使用 Docker 容器化。
   • 使用 API 网关进行统一的路由、认证、限流和监控。
   • 为服务配置健康检查端点。

2. 安全性：

   • 认证与授权：技能 API 必须实施认证（如 API Key, JWT），确保只有可信的智能体平台可以调用。
   • 输入验证：除了 Pydantic 模型验证，还要对输入进行业务逻辑层面的校验，防止注入攻击。
   • 输出过滤：从技能返回给 LLM 的数据，要过滤掉敏感信息（如内部系统错误详情、用户隐私数据）。

3. 可观测性：

   • 结构化日志：记录每一次技能调用的请求、响应、耗时和状态。使用如 JSON 格式，便于接入 ELK 或 Loki。
   • 指标监控：监控技能调用的 QPS、成功率、延迟（P50, P95, P99）。
   • 分布式追踪：为每个用户请求生成唯一的 Trace ID，贯穿所有技能调用，方便排查跨服务问题。

4. 版本管理：

   • 技能的输入输出模型可能发生变化。为技能 API 设计版本号（如/v1/execute），避免不兼容的更新影响线上智能体。

掌握智能体技能的核心在于理解其“模块化”和“接口化”的思想。它不是魔法，而是一种严谨的软件工程实践。从定义一个清晰的技能描述和参数开始，到实现可靠的功能逻辑，再到通过 API 暴露并融入自动化工作流，每一步都需要扎实的开发和运维能力。当你能够熟练地设计、开发和编排技能时，你就真正解锁了智能体从“聊天伙伴”到“业务助手”乃至“自动化员工”的进阶能力。接下来，你可以尝试将更多的企业内外部系统（如数据库、CRM、知识库、审批流）封装成技能，构建出真正解决实际业务痛点的智能体应用。

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度