ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Agent Skill实战:从概念到代码,构建可复用的大模型技能包

Agent Skill实战:从概念到代码,构建可复用的大模型技能包 Agent Skill 是最近 AI 大模型应用开发里出现频率很高的一个词。很多人第一次听到时会以为它是一个新框架或者新协议实际上它更像是一套“给 Agent 用的可复用技能包”。这篇文章不绕弯子直接回答三个问题Agent Skill 到底是什么它和 Agent、MCP 有什么区别以及怎么在一台普通开发机上从零写一个 Skill 并跑通代码实战。文章会从概念拆解开始然后给出一套本地开发 Agent Skill 的环境准备思路接着用 Python 完成一个“从定义 Skill 到注册进 Agent再到批量任务调用”的完整示例。最后补上资源占用观察、常见问题排查和工程化建议。如果你正在做 AI 大模型应用开发或者准备把自己的工具链接入 Agent 生态这篇文章可以直接收藏。1. 核心能力速览能力项说明项目类型Agent Skill 概念讲解 代码实战开发教程核心概念Agent SkillAgent 技能包、Agent、MCPModel Context Protocol适用人群AI 大模型应用开发者、Agent 框架使用者、提示词工程进阶者开发语言Python 3.9推荐硬件普通开发机即可纯逻辑 Skill 不需要 GPU显存占用训练/推理型 Skill 需按实际模型评估逻辑型 Skill 约为 0启动方式命令行运行 / Python 脚本调用支持 API可封装为 REST API示例使用 FastAPI支持批量任务支持本文提供单线程与线程池两种批量调用示例主要功能Skill 定义、参数声明、Agent 注册、调用执行、批量处理、服务封装适合场景本地 Agent 应用开发、工具链集成、MCP 对比学习、AI 应用原型验证2. Agent Skill 到底是什么2.1 从“模型会”到“Agent 会”大模型本身具备通用能力但通用不等于专精。比如一个模型能理解邮件内容但不一定知道“按优先级把邮件分成紧急、普通、推广三类并提取发件人和截止时间”这套完整流程。你可以通过提示词让模型临时做这件事但每次都要重复描述规则且效果不稳定。Agent Skill 要解决的问题就是把这套“流程 规则 输入输出约定”封装成一个可复用的技能单元。Agent 在运行过程中可以直接加载这个 Skill不需要修改底层大模型的权重也不需要每次重新写提示词。换句话说Skill 是模型和具体业务场景之间的一层“能力适配层”。从实现上看一个 Skill 通常包含技能名称和描述告诉 Agent 这个技能是做什么的。输入参数声明定义调用时需要的字段。执行逻辑可以是提示词模板、代码逻辑或者两者的组合。输出格式约定定义返回给 Agent 的结构化结果。2.2 Skill 和 Agent 的区别Agent 是一个能够感知环境、做出决策并执行动作的智能体。它负责“思考下一步做什么”。Skill 是 Agent 可以调用的能力单元负责“具体怎么做”。两者关系可以理解为Agent 是大脑加调度器Skill 是工具箱里的工具。Agent 根据用户请求选择调用哪个 SkillSkill 执行完成后把结果返回给 AgentAgent 再决定下一步动作。一个 Agent 可以挂载多个 Skill同一个 Skill 也可以被不同 Agent 复用。2.3 Skill 和 MCP 有什么区别MCPModel Context Protocol是另一套常见方案很多人会把 Skill 和 MCP 搞混。简单区分MCP 是一种标准化通信协议解决的是“Agent 如何连接外部工具和数据源”的问题。它采用客户端-服务端模型通过标准化接口让 Agent 动态发现并调用外部工具。Agent Skill 更接近“技能包”解决的是“Agent 如何按照特定流程完成任务”的问题。Skill 不一定需要外部服务它可以只是一段提示词模板加一段本地逻辑。实际项目中两者经常配合使用MCP 负责连接外部系统Skill 负责编排任务逻辑。比如一个邮件处理 Agent通过 MCP 连接邮箱服务获取邮件通过 Skill 完成邮件分类和优先级判断。理解这个区别对设计 Agent 应用架构很有帮助。3. 适用场景与使用边界3.1 适合什么场景需要让 Agent 稳定复现某种业务规则比如内容审核、信息抽取、格式转换。需要在多个 Agent 之间复用同一套能力避免重复写提示词。需要将领域知识封装成可维护的模块业务规则变化时只改 Skill不影响 Agent 主体。需要把本地能力开放给 Agent 生态可以通过 API 方式把 Skill 暴露出去。3.2 不适合什么场景模型本身已经很强且不需要固定流程的简单问答直接调用模型即可不需要引入 Skill 层。需要高频繁实时更新的外部数据优先考虑 MCP 工具连接而不是把数据逻辑写死在 Skill 里。超低延迟场景Skill 封装带来的额外调度开销需要评估。3.3 使用边界与合规提醒Agent Skill 开发涉及模型调用和数据流转实际使用时要确认输入数据的合法来源不能把未经授权的个人信息、版权内容上传到第三方模型服务。如果 Skill 涉及人脸识别、声音处理、自动化决策等内容必须遵守相关法律法规并取得必要授权。本地开发时建议关闭外网权限用脱敏数据做测试。4. 环境准备与前置条件4.1 硬件与操作系统Agent Skill 本身不对硬件提出特殊要求。它是逻辑层概念跑在 Python 进程里即可。但如果 Skill 内部需要调用本地大模型推理或向量检索则需要评估 GPU 资源。纯提示词型 Skill 完全可以在没有 GPU 的开发机上运行。推荐环境操作系统Windows 10/11、Ubuntu 20.04、macOS 12内存8GB 以上磁盘预留 5GB 以上用于 Python 环境、依赖库和模型缓存如果涉及本地模型4.2 Python 与依赖建议使用 Python 3.9 以上版本。创建独立虚拟环境避免污染全局环境。python -m venv venv_agent_skill # Windows venv_agent_skill\Scripts\activate # Linux/macOS source venv_agent_skill/bin/activate4.3 需要安装的核心库pip install openai pydantic fastapi uvicorn python-dotenv如果你使用其他模型服务商把openai换成对应的 SDK 即可。示例中会用到大模型 API 做演示但核心代码逻辑不绑定具体服务商。4.4 模型服务配置虽然本文的 Skill 示例可以在没有模型的情况下先跑通流程但为了验证真实效果建议准备一个可以调用的模型 API。创建.env文件MODEL_API_BASEhttps://your-model-provider.com/v1 MODEL_API_KEYyour_api_key_here MODEL_NAMEyour_model_name没有 API Key 的时候可以用模拟函数代替模型调用完整流程不受影响。5. Agent Skill 代码实战开发这一章从零开始写一个可以实际运行的 Agent Skill 示例。选择的任务是“邮件优先级分类与信息抽取”因为它逻辑清晰、业务价值高、不依赖外部服务。5.1 定义 Skill 数据结构使用 Pydantic 定义输入和输出结构这是承接 LLM 返回结果的良好实践。from pydantic import BaseModel, Field from typing import Optional class EmailInput(BaseModel): subject: str Field(..., description邮件主题) body: str Field(..., description邮件正文) sender: Optional[str] Field(None, description发件人地址) class EmailAnalysisOutput(BaseModel): priority: str Field(..., description优先级: urgent/normal/low) category: str Field(..., description分类: work/promo/social/other) deadline: Optional[str] Field(None, description截止时间, 如果存在) summary: str Field(..., description一句话摘要)这段代码同时起到了一个 JSON Schema 的作用后续可以用于校验模型返回内容。5.2 实现 Skill 执行逻辑Skill 的核心是一个可调用对象。它接收输入参数调用模型返回结构化结果。这里给出两种实现方式。方式一使用代码逻辑直接处理无模型依赖可离线运行import re import json class EmailPrioritySkill: 邮件优先级分类 Skill name email_priority_skill description 分析邮件内容, 判断优先级、分类并生成摘要 def execute(self, email_input: EmailInput) - dict: text f{email_input.subject}\n{email_input.body}.lower() # 优先级判断规则 urgent_keywords [urgent, asap, 紧急, 尽快, deadline, 截止] promo_keywords [discount, 促销, 优惠, sale, 打折] if any(k in text for k in urgent_keywords): priority urgent elif any(k in text for k in promo_keywords): priority low else: priority normal # 分类 if discount in text or 优惠 in text or 促销 in text: category promo elif meeting in text or 会议 in text or report in text: category work else: category other # 提取截止时间(简化规则) deadline_match re.search(r(\d{4}-\d{2}-\d{2}), email_input.body) deadline deadline_match.group(1) if deadline_match else None summary f邮件主题: {email_input.subject[:30]} output EmailAnalysisOutput( prioritypriority, categorycategory, deadlinedeadline, summarysummary ) return output.model_dump()方式二调用 LLM 推理实现更灵活适合复杂规则from openai import OpenAI from dotenv import load_dotenv import os load_dotenv() class LLMEmailSkill: 基于 LLM 推理的邮件分析 Skill name llm_email_skill description 使用大模型判断邮件优先级并生成结构化结果 def __init__(self): self.client OpenAI( base_urlos.getenv(MODEL_API_BASE), api_keyos.getenv(MODEL_API_KEY) ) self.model os.getenv(MODEL_NAME) def execute(self, email_input: EmailInput) - dict: prompt f 你是一个邮件助理。请分析以下邮件并输出 JSON。 必须严格按照这个格式输出不要输出额外内容: {{ priority: urgent | normal | low, category: work | promo | social | other, deadline: 截止日期, 没有就为空字符串, summary: 一句话中文摘要 }} 邮件主题: {email_input.subject} 发件人: {email_input.sender or unknown} 邮件正文: {email_input.body} response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是一个严谨的邮件处理助手只输出 JSON。}, {role: user, content: prompt} ], temperature0.1, response_format{type: json_object} # 如果服务商支持 ) result json.loads(response.choices[0].message.content) # 校验并转换为标准输出 output EmailAnalysisOutput( priorityresult.get(priority, normal), categoryresult.get(category, other), deadlineresult.get(deadline) or None, summaryresult.get(summary, ) ) return output.model_dump()5.3 将 Skill 注册到 Agent真正的 Agent 框架会有自己的注册机制。这里实现一个轻量级 Agent 来演示注册与调度逻辑。from typing import Dict, Any, Callable class SimpleAgent: 极简 Agent 容器, 用于演示 Skill 注册和调用 def __init__(self): self.skills: Dict[str, Callable] {} def register_skill(self, skill) - None: 注册技能 self.skills[skill.name] skill print(f[Agent] Skill 已注册: {skill.name}) def run(self, skill_name: str, **kwargs) - Any: 调用技能 if skill_name not in self.skills: raise ValueError(fUnknown skill: {skill_name}) skill self.skills[skill_name] print(f[Agent] 调用 Skill: {skill_name}) return skill.execute(**kwargs) # 实例化 Agent 并注册 Skill agent SimpleAgent() email_skill EmailPrioritySkill() # 先用规则版, 无需 API agent.register_skill(email_skill)5.4 测试调用email EmailInput( subject【紧急】服务器故障处理, body请立即处理, 客户系统已无法访问, 预计截止时间2026-03-01, 需要尽快恢复。, senderopsexample.com ) result agent.run(email_priority_skill, email_inputemail) print(json.dumps(result, ensure_asciiFalse, indent2))预期输出{ priority: urgent, category: other, deadline: 2026-03-01, summary: 邮件主题: 【紧急】服务器故障处理 }5.5 使用 LLM 版本验证如果配置好了模型 API可以切到 LLM 版本再跑一次llm_skill LLMEmailSkill() agent.register_skill(llm_skill) result2 agent.run(llm_email_skill, email_inputemail) print(json.dumps(result2, ensure_asciiFalse, indent2))LLM 版本的输出会更灵活摘要更完整优先级判断更接近人类理解。但代价是每次调用消耗 token且响应时间更长。实际项目中可以先跑规则版稳定交付再逐步引入模型推理能力。6. 接口 API 与批量任务Skill 单独运行只是第一步。真实项目中 Skill 通常要嵌入服务端流程或者支持批量文件处理。这一章演示如何把 Skill 封装成 REST API并实现批量任务。6.1 用 FastAPI 封装 Skill 服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(titleAgent Skill 服务, version1.0.0) # 复用前面的 EmailInput/EmailPrioritySkill email_skill EmailPrioritySkill() app.post(/v1/skills/email_priority) async def run_email_skill(email: EmailInput): 调用邮件优先级 Skill try: result email_skill.execute(email) return {code: 0, data: result} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/v1/skills) async def list_skills(): 列出可用 Skill 列表 return { skills: [ { name: email_priority_skill, description: 邮件优先级分类与信息抽取 } ] } if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动服务python api_server.py用 curl 测试curl -X POST http://127.0.0.1:8000/v1/skills/email_priority \ -H Content-Type: application/json \ -d { subject: 合同审批, body: 请今天下班前完成审批, 截止时间2026-03-02, sender: hrexample.com }启动后可以看到服务监听在 8000 端口所有 Skill 能力都会作为 HTTP 接口暴露出来方便接入到现有的 Web 服务或自动化流程中。6.2 批量任务实现批量处理的核心是遍历输入列表对每个条目执行 Skill并聚合结果。下面给出单线程和线程池两种写法后者的吞吐量更高。import json import concurrent.futures from typing import List def load_emails_from_file(file_path: str) - List[EmailInput]: 从 JSON 文件加载批量邮件数据 with open(file_path, r, encodingutf-8) as f: data json.load(f) return [EmailInput(**item) for item in data] def process_single_email(email: EmailInput) - dict: 处理单个邮件 skill EmailPrioritySkill() try: result skill.execute(email) return {success: True, input: email.subject, result: result} except Exception as e: return {success: False, input: email.subject, error: str(e)} def batch_process_serial(emails: List[EmailInput]) - List[dict]: 串行批量处理 return [process_single_email(e) for e in emails] def batch_process_parallel(emails: List[EmailInput], max_workers: int 4) - List[dict]: 线程池并发处理 with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: return list(executor.map(process_single_email, emails)) if __name__ __main__: import time # 构造测试数据 test_emails [ EmailInput(subject服务器告警, body磁盘空间不足, 紧急处理, senderopsexample.com), EmailInput(subject周报提交, body请周五前提交本周周报, sendermanagerexample.com), EmailInput(subject产品促销, body全场8折优惠, 限时抢购, sendermarketingexample.com), EmailInput(subject客户会议, body下周一下午3点会议室A开会, senderclientexample.com), EmailInput(subject招聘面试, body候选人已到, 请安排面试官, senderhrexample.com), ] # 串行 start time.time() serial_results batch_process_serial(test_emails) serial_time time.time() - start # 并发 start time.time() parallel_results batch_process_parallel(test_emails, max_workers4) parallel_time time.time() - start print(f串行耗时: {serial_time:.2f}s, 并发耗时: {parallel_time:.2f}s) for r in parallel_results: print(json.dumps(r, ensure_asciiFalse))批量任务设计建议输入和输出都使用结构化 JSON便于日志审计和任务失败重放。单条失败不能拖垮整个任务上面示例已做 try/except 隔离。大批量任务建议引入消息队列如 Redis Queue、Celery断点续跑。6.3 批量任务失败重试策略对于并发批处理重试时需要区分暂时性错误网络超时、API 限流和永久性错误数据格式错误。注意如果 Skill 的execute方法在单条调用时报错任务会返回success: false这种情况不应该盲目重试常见做法是实现带最大重试次数的包装函数只重试那些标记为暂时性错误的记录。import time from functools import wraps def retry(max_retries: int 3, delay: float 1.0): 简单重试装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt max_retries - 1: raise time.sleep(delay * (attempt 1)) return None return wrapper return decorator # 使用重试机制 retry(max_retries3, delay0.5) def process_single_email_with_retry(email: EmailInput) - dict: skill LLMEmailSkill() # LLM 版本更容易出现临时错误 result skill.execute(email) return {success: True, input: email.subject, result: result}7. 资源占用与性能观察Agent Skill 的资源占用取决于内部实现。纯规则型 Skill 几乎不消耗额外资源LLM 推理型 Skill 的消耗主要集中在模型推理阶段实际占用会随模型大小和输入长度变化需要在本机实测确认。7.1 显存与内存运行纯逻辑型 Skill内存占用通常可以控制在 200MB 以下不占用显存。运行 LLM 推理型 Skill需要评估模型推理的显存占用。如果使用本地大模型建议单独开一个模型服务Skill 只作为客户端调用避免在同一个进程里既跑模型又跑业务逻辑。7.2 性能瓶颈观察以下三个指标需要关注单次 Skill 执行延迟主要取决于模型响应速度、网络延迟和提示词长度。并发吞吐量线程池能提高 IO 密集型任务的吞吐上限。token 消耗成本LLM Skill 的每个 token 都是成本需要记录和监控。7.3 降低资源占用的方法减少提示词长度只传必要字段。优先使用规则型逻辑处理简单场景LLM 只处理复杂场景。使用流式输出避免长时间占用连接资源。批量任务采用固定大小线程池避免无限创建线程。模型网关层加缓存相同输入直接返回缓存结果。7.4 端口冲突与进程管理如果运行 FastAPI 服务时遇到端口被占用可以检查端口占用进程并释放或者修改端口号。服务进程建议用 nohup 或 systemd 托底避免终端退出导致服务中断。# 查看端口占用 # Windows netstat -ano | findstr :8000 # Linux/macOS lsof -i :8000 # 指定新的端口启动 python api_server.py --port 80018. 常见问题与排查方法问题现象可能原因排查方式解决方案Skill 注册失败名称重复或未实例化检查注册代码和名称确保 skill.name 唯一模型 API 调用超时网络问题或模型规格设置不当查看服务商后台日志增加超时时间或改用流式调用LLM 返回 JSON 格式错误模型未遵循输出格式打印原始输出强化 prompt 约束, 开启 response_format批量任务运行一段时间后变慢线程池资源耗尽或限流观察 CPU 和 API 调用频率降低并发数, 加入退避策略本地服务启动后无法访问端口被占用或防火墙拦截检查端口和监听地址监听 0.0.0.0 并确认防火墙放行依赖安装失败Python 版本或 pip 源问题检查 pip 版本和网络使用国内镜像源或升级 PythonCSV/JSON 大数据量处理慢单线程处理观察 CPU 利用率改用并发处理或分批处理拿不到预期字段输出解析逻辑与真实返回不匹配打印 JSON Schema 和原始响应增加字段校验和容错处理提示词里包含敏感数据数据经第三方模型流转检查 AP I调用日志脱敏后再上传, 或使用本地模型8.1 模型输出不稳定怎么处理LLM 推理型 Skill 最常见的问题是“这次结果和上次不一样”。这不是 bug而是模型属性。工程上有几种抗波动手段降低 temperature让模型更保守。明确输出 JSON 格式并做 schema 校验。增加约束性示例让模型按示例格式输出。对关键字段做规则后处理例如日期格式统一。9. 最佳实践与使用建议9.1 先小参数验证再上量第一次跑 Skill 时先用一两条数据验证确认输出结构稳定后再跑完整批量任务。不要一上来就全量执行。9.2 保持一套最小可运行配置把 Skill 的代码、测试数据、输出样例整理在固定目录下方便后续复用。9.3 模型文件、输入素材、输出结果分目录管理agent-skill-demo/ ├── skills/ # Skill 定义 │ ├── email_skill.py │ └── __init__.py ├── data/ │ ├── inputs/ # 输入数据 │ └── outputs/ # 输出结果 ├── config/ │ └── .env ├── tests/ │ └── test_email_skill.py └── api_server.py9.4 批量任务务必加日志和失败重试每条任务记录输入文件、调用时间、返回码、耗时、失败原因。失败任务自动重试最多 3 次超过重试次数后进入死信队列。9.5 接口服务要限制访问范围Skill API 如果暴露在公网必须加鉴权。最基础的做法是 API Key 校验。from fastapi import Header, HTTPException API_KEY your-secret-api-key app.post(/v1/skills/email_priority) async def run_email_skill( email: EmailInput, x_api_key: str Header(defaultNone) ): if x_api_key ! API_KEY: raise HTTPException(status_code401, detailInvalid API Key) result email_skill.execute(email) return {code: 0, data: result}9.6 数据与版权合规边界在开发 AI 应用时要特别注意数据来源的合法性和用户隐私保护。凡是涉及个人数据、商业机密或版权素材的内容在接入第三方模型前应进行脱敏处理并确认已获得必要的授权。在真实环境中不要把未脱敏的客户邮件、用户画像等敏感数据直接传给外部模型服务。如果确实需要处理敏感数据优先考虑本地部署模型或使用私有化服务。9.7 Skill 的版本管理Skill 会随着业务规则变化而调整。建议使用语义化版本号管理1.0.0首次发布。1.1.0新增字段或能力。1.0.1修复 bug接口不变。每次变更后跑一次回归测试确保历史调用方不受影响。10. 总结与下一步Agent Skill 给了开发者一条让大模型能力“业务化”的清晰路径。它不依赖复杂的框架通过定义输入、抽象逻辑、注册调用就能完成能力封装。对比直接写提示词Skill 更可控、更可复用对比 MCPSkill 更轻量、更贴近模型自身的行为编排。两者不是替代关系而是互补关系。这篇文章里用“邮件优先级分类”这个例子走完了从定义 Skill 到注册 Agent、再到 API 封装和批量任务的全链路。建议你自己动手验证的重点有三个第一先跑通纯规则版 Skill确认流程闭环第二切到 LLM 版对比输出稳定性第三把 Skill 封装成 API 并用 curl 或 Postman 测试。最容易踩的坑是模型输出格式不稳定。写 LLM 版 Skill 时一定要加 JSON Schema 校验raw 输出不能直接下游使用。下一步可以探索的方向把 Skill 接入 MCP Server实现 Agent 通过标准协议动态发现和调用 Skill将 Skill 组合成复杂工作流或者用向量检索的方式根据用户请求自动选择最合适的 Skill。理论上只要模型能理解输入输出约定Agent Skill 就能覆盖大量重复性业务逻辑。这也是大模型应用开发中值得投入时间掌握的一个方向。
返回列表