1. 项目概述为什么“Agent Skills”是当下最值得投入的技术方向最近和几个做AI应用开发的朋友聊天大家不约而同地都在讨论一个词Agent Skills。这已经不是实验室里的概念了而是正在快速落地、产生实际商业价值的核心技术。简单来说你可以把Agent Skills理解为“智能体的技能库”。过去我们训练一个大语言模型希望它无所不能但结果往往是“样样通样样松”。而Agent Skills的思路是反过来的我们不再追求一个全能模型而是构建一个“大脑”Agent然后为它配备各种专业、可插拔的“技能”Skills。这个大脑负责理解用户意图、规划任务、调用合适的技能并整合结果。这解决了什么痛点想象一下你开发了一个客服机器人。用户问“帮我查一下上周三的订单状态然后发一份PDF报告到我的邮箱”。传统的单一模型要么无法执行查询数据库和发送邮件这两个动作要么需要极其复杂的提示词工程才能勉强完成且稳定性很差。而有了Agent Skills架构你的Agent可以理解这是两个步骤第一步调用“查询订单数据库”这个Skill第二步调用“生成PDF并发送邮件”这个Skill。每个Skill都是独立、专注、经过充分测试的模块。这种解耦带来的好处是巨大的开发效率高可以并行开发不同技能、维护成本低一个技能出问题不影响其他、可扩展性强新需求来了就加个新技能。所以无论你是AI产品经理、全栈开发者还是业务线的技术负责人深入理解并掌握Agent Skills的构建方法都意味着你拿到了打开下一代AI应用大门的钥匙。它不再是“未来可期”而是“现在就要用”。接下来我将结合自己从零搭建多个智能体项目的实战经验拆解从设计理念到代码落地的完整路径分享那些官方文档里不会写的“坑”和“技巧”。2. 核心架构设计如何规划一个高内聚、低耦合的Skill体系设计Agent Skills最忌讳的就是一开始就埋头写代码。糟糕的架构设计会让后期的技能管理变成一场噩梦。我的经验是必须先从顶层想清楚技能的边界、通信方式和生命周期管理。2.1 技能的分类与边界定义不是所有功能都适合做成一个Skill。一个良好的Skill应该符合“单一职责原则”。我通常将Skill分为三大类工具型技能这是最常见的一类执行一个具体的、原子性的操作。例如search_web: 执行网络搜索并返回摘要。query_database: 根据自然语言查询数据库。send_email: 发送邮件。calculate: 执行数学计算或单位换算。这类技能的特点是输入输出明确内部逻辑专注不依赖其他技能的状态。流程型技能它本身封装了一个小的工作流内部可能会按顺序或条件调用多个工具型技能。例如generate_report: 这个技能可能内部先调用query_database获取数据再调用data_visualization生成图表最后调用compile_document整合成报告。对外它仍然是一个统一的接口。handle_customer_complaint: 内部包含查询订单、分析问题、生成解决方案模板、通知客服经理等多个步骤。设计这类技能的关键是定义清晰的输入输出并做好内部子步骤的异常处理和状态回滚。决策型技能这类技能不直接操作外部系统而是提供分析、判断或建议。例如sentiment_analysis: 分析一段文本的情感倾向。intent_classification: 判断用户输入的意图属于哪个类别。risk_assessment: 根据提供的信息评估风险等级。它们通常是其他技能尤其是流程型技能的“前哨”为Agent的决策规划提供依据。边界定义的黄金法则如果一个功能会被多个不同的任务或工作流频繁用到且其逻辑相对独立那么它就值得被抽象成一个独立的Skill。反之如果一段逻辑只服务于某个非常特定的场景且变化可能性极低那么它可以作为某个Skill的内部实现不必独立。2.2 通信协议与数据契约Skill和Agent之间如何对话这里有两个核心协议和数据结构。协议层面目前主流有两种方式函数调用这是最自然的方式。每个Skill在Agent那里注册为一个“函数”包含函数名、描述和参数JSON Schema。当Agent决定调用某个Skill时就生成符合该Schema的参数。这种方式与OpenAI的Function Calling、Google的Tool Calling等原生支持完美契合开发体验好。我绝大部分工具型技能都采用这种方式。消息队列/事件驱动适用于异步、长时间运行或需要更高解耦的场景。Skill作为一个独立服务监听特定主题的消息。Agent发布一个任务事件相应的Skill消费并处理再将结果事件发布回来。这种方式更适合微服务架构但复杂度也更高。我通常只在处理视频渲染、大规模文档处理等耗时任务时使用。数据契约层面这是最容易出问题的地方。你必须为每个Skill定义严格的输入输出Schema。我强烈建议使用像Pydantic这样的库来定义数据模型。这不仅能做运行时验证还能自动生成清晰的文档。from pydantic import BaseModel, Field from typing import List # 为“查询天气”Skill定义输入输出模型 class WeatherQueryInput(BaseModel): location: str Field(..., description城市名例如北京、Shanghai) date: str Field(defaulttoday, description查询日期格式YYYY-MM-DD 或 today, tomorrow) class WeatherQueryOutput(BaseModel): location: str date: str temperature_high: float Field(..., description最高气温摄氏度) temperature_low: float Field(..., description最低气温摄氏度) condition: str Field(..., description天气状况如晴、多云、雨) forecast: List[str] Field(default_factorylist, description简要预报列表) # Skill的实现函数 def get_weather(query: WeatherQueryInput) - WeatherQueryOutput: # ... 实现逻辑 return WeatherQueryOutput(...)这样做的好处是Agent在规划时就能“知道”每个Skill需要什么、产出什么大大降低了调用错误。同时这些模型定义本身就是最好的API文档。2.3 技能的生命周期与依赖管理随着Skill数量增多超过20个管理就成了大问题。你需要一个“技能注册中心”。我的方案是维护一个中心化的skill_registry.py文件或者使用一个简单的YAML配置文件。skills: - name: get_weather type: tool description: 查询指定城市和日期的天气信息 module_path: skills.weather.query function_name: get_weather input_schema: skills.weather.models.WeatherQueryInput output_schema: skills.weather.models.WeatherQueryOutput required_configs: - WEATHER_API_KEY - name: send_slack_message type: tool description: 向指定的Slack频道发送消息 module_path: skills.communication.slack function_name: send_message # ... 其他配置Agent启动时会加载这个注册表动态导入所有Skill模块并将它们注册到LLM的上下文中。对于依赖管理比如某个Skill需要数据库连接池或者特定的API客户端我推荐使用依赖注入。在Skill的工厂函数或初始化方法中传入这些依赖而不是在Skill内部硬编码创建这样便于测试和替换。实操心得Skill的“冷启动”与“热加载”在开发初期我们经常需要增减或修改Skill。每次都重启Agent服务太麻烦。我的做法是在开发环境将Skill注册表设计为可动态加载。使用像watchdog这样的库监听skills/目录的文件变化当检测到__init__.py或主要模型文件变更时自动向Agent发送信号触发技能列表的重新加载。这能极大提升开发调试效率。但在生产环境务必关闭此功能通过标准的CI/CD流程来更新技能。3. 技能开发实战从零构建一个健壮、可用的Tool Skill理论说再多不如动手写一个。我们以构建一个“智能邮件摘要”Skill为例它需要读取邮箱中的未读邮件调用LLM生成摘要并将结果保存。这个Skill兼具工具型读邮件、调用LLM、写文件和一点流程型串联多个步骤的特点很有代表性。3.1 第一步明确需求与接口设计这个Skill我们命名为summarize_unread_emails。输入用户可能想摘要“过去24小时”的邮件或“来自客户A”的邮件。所以输入参数需要灵活。输出一个结构化的摘要报告包含摘要列表和可能的关键词。核心动作认证邮箱、获取邮件、解析内容、调用LLM摘要、格式化输出、持久化。首先我们定义数据契约from pydantic import BaseModel, Field from datetime import datetime from typing import List, Optional from enum import Enum class TimeRange(str, Enum): LAST_24H last_24h LAST_7_DAYS last_7_days TODAY today class EmailSummaryInput(BaseModel): mailbox: str Field(defaultINBOX, description要扫描的邮箱文件夹如 INBOX, ‘Sent’) time_range: TimeRange Field(defaultTimeRange.LAST_24H, description时间范围) sender_filter: Optional[str] Field(defaultNone, description发件人过滤如 ‘company.com’) max_emails: int Field(default10, ge1, le50, description最多处理多少封邮件) class EmailSummary(BaseModel): subject: str sender: str received_time: datetime summary: str keywords: List[str] class EmailSummaryOutput(BaseModel): total_processed: int summaries: List[EmailSummary] report_path: Optional[str] Field(None, description生成的摘要报告文件路径)3.2 第二步分模块实现与错误处理不要在一个大函数里写完所有逻辑。我们将Skill拆解为几个子函数每个负责一个环节这样易于测试和维护。import imaplib import email from email.header import decode_header import logging from typing import List from .models import EmailSummaryInput, EmailSummary, EmailSummaryOutput logger logging.getLogger(__name__) class EmailSummarizerSkill: def __init__(self, llm_client, storage_client): # 依赖注入LLM客户端和存储客户端如本地文件系统或云存储 self.llm llm_client self.storage storage_client def _connect_mailbox(self, server, user, password, mailboxINBOX): 连接邮箱返回已选择的邮箱对象 try: mail imaplib.IMAP4_SSL(server) mail.login(user, password) status, _ mail.select(mailbox) if status ! OK: raise ConnectionError(f无法选择邮箱: {mailbox}) return mail except imaplib.IMAP4.error as e: logger.error(f邮箱连接失败: {e}) raise def _fetch_unread_emails(self, mail_conn, time_range, sender_filter, max_emails): 根据条件获取未读邮件ID # 构建IMAP搜索查询字符串 query [UNSEEN] # 未读 if time_range TimeRange.LAST_24H: query.append(SINCE, (datetime.now() - timedelta(days1)).strftime(%d-%b-%Y)) # ... 其他时间范围处理 if sender_filter: query.append(FROM, sender_filter) status, message_ids mail_conn.search(None, *query) if status ! OK: return [] # 截取最多需要的邮件ID ids message_ids[0].split()[-max_emails:] if max_emails else message_ids[0].split() return ids def _parse_email_content(self, msg): 解析邮件原始内容提取主题、发件人、正文文本 # 解码主题 subject, encoding decode_header(msg[Subject])[0] if isinstance(subject, bytes): subject subject.decode(encoding if encoding else utf-8, errorsignore) # 提取发件人 sender msg.get(From) # 提取纯文本正文简化版实际需处理多部分邮件 body if msg.is_multipart(): for part in msg.walk(): if part.get_content_type() text/plain: charset part.get_content_charset() or utf-8 body part.get_payload(decodeTrue).decode(charset, errorsignore) break else: charset msg.get_content_charset() or utf-8 body msg.get_payload(decodeTrue).decode(charset, errorsignore) return subject, sender, body[:5000] # 限制正文长度避免LLM上下文过长 def _summarize_with_llm(self, subject, body): 调用LLM生成摘要和关键词 prompt f 请对以下邮件内容生成一个简洁的摘要不超过100字并提取3-5个关键词。 邮件主题{subject} 邮件正文{body} 请以JSON格式回复包含summary和keywords两个字段。 try: response self.llm.chat_complete(messages[{role: user, content: prompt}]) # 解析LLM返回的JSON import json result json.loads(response.choices[0].message.content) return result.get(summary, ), result.get(keywords, []) except Exception as e: logger.warning(fLLM摘要失败: {e}, 使用备用方案) # 备用方案简单截取 return body[:150] ..., [] def execute(self, input_data: EmailSummaryInput) - EmailSummaryOutput: Skill的主执行函数 summaries_list [] mail_conn None try: # 1. 连接邮箱密码等应从环境变量或配置中心获取此处为示例 mail_conn self._connect_mailbox( serverimap.example.com, useros.getenv(EMAIL_USER), passwordos.getenv(EMAIL_PASS), mailboxinput_data.mailbox ) # 2. 获取邮件ID email_ids self._fetch_unread_emails( mail_conn, input_data.time_range, input_data.sender_filter, input_data.max_emails ) logger.info(f找到 {len(email_ids)} 封待处理邮件。) # 3. 遍历处理每封邮件 for e_id in email_ids: try: status, msg_data mail_conn.fetch(e_id, (RFC822)) if status ! OK: continue raw_email msg_data[0][1] msg email.message_from_bytes(raw_email) subject, sender, body self._parse_email_content(msg) if not body.strip(): # 跳过空正文邮件 continue summary, keywords self._summarize_with_llm(subject, body) email_summary EmailSummary( subjectsubject, sendersender, received_timedatetime.now(), # 实际应从邮件头解析 summarysummary, keywordskeywords ) summaries_list.append(email_summary) except Exception as e: logger.error(f处理邮件ID {e_id} 时出错: {e}, exc_infoTrue) # 单封邮件处理失败不应导致整个Skill失败记录后继续 continue # 4. 生成报告文件 report_path None if summaries_list: report_content \n\n.join([f主题{s.subject}\n发件人{s.sender}\n摘要{s.summary}\n关键词{, .join(s.keywords)} for s in summaries_list]) report_filename femail_summary_{datetime.now().strftime(%Y%m%d_%H%M%S)}.txt report_path self.storage.save(report_content, report_filename) # 5. 返回结果 return EmailSummaryOutput( total_processedlen(email_ids), summariessummaries_list, report_pathreport_path ) except Exception as e: logger.critical(fEmailSummarizerSkill 执行失败: {e}, exc_infoTrue) raise RuntimeError(f邮件摘要技能执行失败: {e}) from e finally: if mail_conn: try: mail_conn.close() mail_conn.logout() except: pass3.3 第三步配置、注册与测试将Skill注册到中心注册表。同时为其编写单元测试和集成测试。单元测试针对_parse_email_content,_summarize_with_llm等内部函数集成测试则模拟邮箱连接和LLM调用验证整个execute流程。避坑指南Skill的幂等性与超时控制幂等性Agent可能会因为网络抖动或自身逻辑重试某个Skill。如果你的Skill不是幂等的比如“发送邮件”发两次就会出问题。对于写操作类Skill一定要设计幂等键。例如发送邮件时附带一个唯一ID在服务端检查该ID是否已执行过。超时控制永远不要相信外部服务。在Skill的execute方法中必须为任何网络IO数据库查询、API调用设置合理的超时。可以使用asyncio.wait_for或threading.Timer。超时后应抛出明确异常让Agent有机会尝试其他方案或向用户报错。资源清理就像上面的示例中在finally块里关闭邮箱连接一样任何Skill在结束时都必须确保释放它占用的资源网络连接、文件句柄、临时文件等。否则随着调用次数增加你的Agent服务会慢慢被拖垮。4. Agent大脑的训练如何让Agent学会精准调用Skills有了好用的Skills下一个关键就是让Agent这个“大脑”知道在什么情况下该用哪个Skill以及如何组合它们。这主要依靠提示词工程和规划能力的训练。4.1 技能描述的“艺术”Agent是通过你提供的技能描述来理解每个Skill能做什么的。一段糟糕的描述会导致误调用。写描述时要站在Agent的角度思考清晰明确用简单句说明这个技能是干什么的。避免模糊词汇。差“处理数据”。处理什么数据怎么处理好“根据用户提供的自然语言问题查询产品数据库返回匹配的产品名称、ID和库存状态。”场景举例在描述后附上1-2个典型的使用例子。这能极大提升LLM对意图匹配的准确性。例如“例如当用户问‘红色连衣裙还有货吗’或‘帮我找一下iPhone 15的库存’时可以调用此技能。”输入输出强调明确指出关键输入参数和输出是什么。“输入需要包含一个‘query’字段即用户的自然语言问题。输出是一个产品列表每个产品包含name, id, stock字段。”一个完整的技能描述模板如下技能名称query_product_inventory 描述根据用户的自然语言查询在产品数据库中搜索相关产品并返回其详细信息与库存状态。当用户询问商品是否有货、查找特定商品时使用。 输入参数 - query (string): 用户的查询语句例如“红色的尺码M的T恤衫”。 输出 - products (array): 匹配的产品列表每个产品是一个对象包含 name, product_id, color, size, stock_count 字段。 示例调用场景 1. 用户“你们还有黑色的笔记本电脑吗” - 调用此技能query“黑色的笔记本电脑”。 2. 用户“我想看看运动鞋要耐克的。” - 调用此技能query“耐克运动鞋”。4.2 思维链与规划提示词设计对于简单任务Agent可能直接调用一个Skill。但对于复杂任务如“查天气然后推荐穿衣”Agent需要规划。我们需要在系统提示词中引导Agent进行“思考”。我的系统提示词通常包含以下几个部分角色定义明确告诉AI它的角色是什么。能力清单以结构化方式列出所有可用的Skills及其描述。这是最重要的部分。操作指令规定它必须如何思考和行为。“请逐步思考用户的请求。”“首先分析用户的目标并将其分解为子任务。”“然后浏览你的技能列表为每个子任务选择最合适的一个技能。”“在决定调用技能前请先确认你已完全理解该技能的输入要求并在脑海中构思好参数。”“一次只调用一个技能。等待该技能返回结果后再基于结果决定下一步行动。”输出格式要求它以特定格式如JSON输出它的“思考过程”和“调用决定”方便程序解析。一个简化的示例你是一个高效的AI助手拥有以下技能 1. get_weather: [描述...] 2. recommend_clothing: [描述...] 3. ... 请遵循以下步骤响应用户 1. 思考分析用户请求的最终目标是什么需要哪些步骤 2. 规划将目标分解为顺序执行的子任务。 3. 匹配为每个子任务从我提供的技能列表中挑选最合适的一个。 4. 执行生成一个JSON数组按顺序列出每个要调用的技能名称和参数。格式[{skill: 技能名, input: {参数}}] 用户请求{用户输入} 请开始你的思考。通过这种方式我们“强迫”LLM进行显式推理并将推理过程结构化大大提高了复杂任务执行的准确性和可解释性。4.3 基于人类反馈的微调提示词不是一劳永逸的。在Agent上线初期必然会出现“该调用时不调用”或“调用错了”的情况。我们需要收集这些bad cases用于微调。数据收集记录每一次用户交互。包括用户原始输入、Agent的思考过程如果有、Agent调用的技能及参数、技能返回的结果、最终给用户的回复。最重要的是要有一个反馈机制让用户或标注员标记这次交互是否成功。微调策略监督微调将成功的交互用户输入 - 正确的技能调用序列作为训练数据对基础LLM进行微调让它更倾向于做出正确的规划。这能显著提升技能选择的准确性。奖励模型对于更复杂的场景可以训练一个奖励模型来评价Agent的整个规划序列的好坏然后使用强化学习如PPO来优化Agent的策略。这属于进阶玩法成本较高但对于性能提升的上限也更高。实操心得从“ReAct”模式到“Plan-and-Execute”早期我们常用ReActReasoning and Acting模式让Agent在“思考一句话”和“执行一个动作”之间交替进行。这在简单场景下有效但对于多步骤复杂任务容易陷入循环或迷失。现在我更倾向于“Plan-and-Execute”模式。在第一步用一个“规划器”LLM可以是同一个模型但给予更强的规划提示词生成一个完整的、分步骤的计划。然后另一个“执行器”LLM或同一个模型切换上下文严格按计划一步步调用技能。这种解耦让逻辑更清晰也更容易调试。你可以把“规划器”本身也看作一个特殊的Skill。5. 高级话题技能编排、流式响应与生产环境部署当单个Agent和多个Skills能稳定工作后我们会面临更高级的需求如何让多个技能协同完成更宏大的任务如何让用户体验更流畅如何保证线上服务的稳定5.1 复杂工作流与技能编排对于“制定一份市场推广计划”这样的任务它需要1) 搜索最新行业趋势2) 分析竞争对手动态3) 生成创意文案4) 估算预算。这涉及到多个技能的顺序、并行甚至条件分支执行。此时单纯的Agent大脑LLM来做全流程的实时规划可能会显得笨重且容易出错。我们需要引入工作流引擎的概念。我的做法是将复杂任务模板化将常见的复杂任务预先定义成工作流模板。例如“市场推广计划生成”就是一个模板。使用DAG有向无环图定义流程每个节点是一个Skill或一个决策点由LLM判断。边定义了执行顺序和依赖关系。引入专用编排器开发一个轻量的“工作流编排器”服务。它的职责是加载DAG模板根据用户输入初始化参数然后按图索骥依次触发各个Skill的执行并管理它们之间的数据传递。工具上可以直接使用像Prefect或Airflow这样的成熟工作流调度框架也可以根据复杂度自己实现一个简单的状态机。这样做的好处是可维护性工作流逻辑清晰可见修改方便。可观测性每个节点的状态成功、失败、执行中都可以被监控。复用性同一个Skill可以被多个不同的工作流复用。5.2 流式响应与用户体验优化LLM生成文本可以流式输出让用户感觉响应很快。那么当Agent在执行一个包含多个步骤、可能耗时较长的工作流时我们如何给用户及时的反馈我的方案是分级流式响应规划流当用户提出复杂请求后Agent首先流式输出它的“思考过程”和“执行计划”。例如“好的您需要制定市场推广计划。我将分四步进行1. 搜索行业趋势... 2. 分析竞品... 3. ... 4. ... 现在开始执行第一步。”执行状态流每开始执行一个Skill就推送一条通知“正在搜索行业趋势...”Skill执行成功推送“行业趋势搜索完成找到X条关键信息。”执行失败推送“搜索遇到问题正在尝试备用方案...”。最终结果流所有步骤完成后流式输出最终的报告内容。这需要前后端配合使用WebSocket或Server-Sent Events (SSE) 来建立双向通信通道。对于用户来说他们能清晰地知道任务进展到了哪一步而不是面对一个长时间的空转界面体验提升巨大。5.3 生产环境部署与监控告警将Agent Skills系统部署上线远不止是启动一个服务那么简单。部署架构建议采用微服务架构。将Agent核心LLM交互、规划逻辑作为一个服务每个或每组相关的Skills作为独立的服务。它们之间通过轻量级RPC如gRPC或消息队列如RabbitMQ, Kafka通信。这样便于独立扩缩容。例如search_webSkill可能调用频繁可以单独多部署几个实例。配置中心所有Skills的API密钥、数据库连接串、服务端点等配置必须从环境变量或配置中心如Consul, etcd读取绝不能硬编码在代码中。日志与追踪必须实现分布式追踪。为每一个用户会话生成一个唯一的trace_id这个ID需要穿透Agent服务和所有被调用的Skill服务。这样当出现问题时你可以在日志系统中通过trace_id一键拉出整个调用链的所有日志快速定位问题根因。使用像Jaeger或OpenTelemetry这样的工具。监控与告警需要监控几个关键指标技能调用成功率每个Skill的成功/失败次数。失败率突然升高要立即告警。技能调用延迟P50, P95, P99延迟。延迟变长可能意味着依赖的第三方API变慢或自身资源不足。Agent规划延迟从收到用户请求到生成第一个Token的时间。这反映了LLM本身的响应速度。Token消耗每天、每用户的Token使用量用于成本核算。业务指标如“订单查询”Skill的调用量直接关联业务。熔断、降级与重试对于依赖外部API的Skill如搜索、支付必须实现熔断器如使用tenacity库。当连续失败次数达到阈值熔断器打开短时间内直接拒绝请求避免拖垮系统。同时要有降级方案比如搜索Skill挂了是否可以返回缓存的历史数据或者给用户一个友好的提示对于暂时性错误如网络超时要设计合理的重试策略如指数退避。生产环境血泪教训技能版本管理这是最容易忽略的一点。当你更新了一个Skill的逻辑比如修改了数据库查询语句如何确保线上正在运行的所有Agent实例都能平滑地切换到新版本粗暴地重启所有服务会导致服务中断。我的做法是为每个Skill定义版本号如v1.2.0并在Skill注册中心记录。Agent在调用Skill时可以指定版本号或者默认调用稳定版。通过蓝绿部署或金丝雀发布的方式先将新Skill版本部署到少量实例让一部分Agent流量导入测试确认无误后再全量切换。同时确保Skill的接口向后兼容至少在一段时间内新旧版本接口要同时支持。