1. 项目概述为什么我们需要为AI智能体设计“技能”最近在GitHub上围绕“AI Skills Agent”的项目热度居高不下这背后反映了一个清晰的趋势单纯调用一个大语言模型LLM的API已经不够了。无论是构建一个能自动处理邮件的助手还是一个能分析数据并生成报告的分析师我们都需要将这些复杂的任务拆解成一个个可复用、可组合、可管理的“技能”Skills。这就像给一个聪明的实习生LLM配备了一套标准化的工具和操作手册让他不仅能理解指令还能高效、可靠地执行具体任务。“怎么写好Agent Skills”这个问题本质上是在问如何将LLM的泛化能力工程化为解决特定领域问题的可靠组件如果设计得不好你的智能体可能会陷入“幻觉”、执行步骤混乱、难以调试和维护的泥潭。我花了大量时间研究和实践发现借鉴软件工程中经典的设计模式思想是构建高质量Agent Skills的捷径。这些模式不是死板的规则而是经过验证的最佳实践蓝图能帮你规避常见陷阱提升技能的可读性、可维护性和复用性。接下来我就结合具体场景拆解5种最核心、最实用的Agent Skills设计模式。2. 核心设计模式解析从理念到实践设计模式的核心价值在于提供了一种共享的语言和思维框架。在Agent Skills的语境下我们关注的是如何组织提示词Prompt、工具调用Tool Calling、记忆Memory以及技能之间的协作关系。下面这五种模式基本覆盖了从简单到复杂的技能构建场景。2.1 模式一链式模式这是最基础也是最常用的模式灵感来源于责任链模式。其核心思想是将一个复杂任务分解为一系列顺序执行的子任务技能前一个技能的输出作为后一个技能的输入。场景示例一个“周报生成”智能体。任务不是直接让LLM“写周报”而是分解为1. 从日程工具中提取本周会议记录技能A2. 从代码仓库解析本周提交记录技能B3. 整合A和B的输出并按照固定模板生成周报草稿技能C4. 将草稿发送给用户确认技能D。实操要点与代码示意# 伪代码示例展示链式执行逻辑 class ChainOfSkillsAgent: def run(self, task_input): context task_input for skill in [skill_a, skill_b, skill_c, skill_d]: # 每个skill都是一个独立的函数或对象负责特定子任务 context skill.execute(context) # 此处可加入错误处理如果某个skill失败可中断链条或重试 if context.get(error): break return context # 技能A提取会议记录 def skill_a_extract_meetings(context): # 调用日历API的工具 meetings call_calendar_api(context[date_range]) # 可能用LLM总结会议要点 summary llm_call(f总结以下会议记录的关键点{meetings}) return {**context, meeting_summary: summary}为什么有效链式模式强制进行了任务分解使得每个技能保持单一职责Single Responsibility极大降低了单个技能的复杂度。调试时你可以清晰地看到数据在链条中的流动定位问题发生在哪个环节。此外技能可以独立开发和测试。注意事项错误传播链条中的一环失败会导致整个任务失败。必须设计健壮的错误处理机制比如为关键技能设置重试逻辑或提供“降级”技能如当无法获取数据时生成一个提示用户手动输入的回复。上下文管理随着链条变长传递的context对象可能变得臃肿。需要定义清晰的数据契约明确每个技能输入和输出的字段避免技能间产生隐式依赖。2.2 模式二路由模式当智能体需要根据输入内容或当前状态动态选择执行哪条技能路径时路由模式就派上用场了。这类似于策略模式或状态模式。场景示例一个“智能客服路由”Agent。用户输入一句话Agent需要判断其意图是“查询订单状态”、“投诉产品质量”还是“咨询退货政策”。根据判断结果路由到对应的专用技能模块进行处理。实操要点 实现路由的核心是一个“路由决策器”Router。这个决策器本身可以是一个小型的LLM调用也可以是基于规则或分类模型的判断。class RouterAgent: def route(self, user_query): # 使用LLM进行意图识别也可用更轻量的分类模型 intent_prompt f 用户说{user_query} 请判断其意图属于以下哪一类 A. 查询订单 B. 产品投诉 C. 退货咨询 D. 其他 只输出字母。 intent llm_call(intent_prompt).strip() # 根据意图路由到对应的技能 skill_map { A: skill_query_order, B: skill_handle_complaint, C: skill_return_policy, D: skill_fallback_general } target_skill skill_map.get(intent, skill_fallback_general) return target_skill.execute(user_query)为什么有效它避免了构建一个“全能”但臃肿的单一技能而是通过分工协作让每个技能在其专业领域内做到极致。系统易于扩展新增一个意图类别只需增加一个技能并更新路由表即可。实操心得设计清晰的决策边界意图分类的类别必须互斥且覆盖全面。模糊的类别会导致路由不准。在初期可以允许LLM输出“不确定”并路由到一个“澄清问题”的技能。路由器的性能如果每次请求都经过LLM做路由判断可能会增加延迟和成本。对于高频、固定的意图可以考虑训练一个轻量级的文本分类模型或者缓存常见的路由结果。2.3 模式三组合模式这个模式用于处理具有树状或层级结构的任务。一个父技能可以调用多个子技能并综合它们的结果。这与面向对象中的组合模式类似强调“部分-整体”的层次关系。场景示例一个“行业分析报告生成”技能。这个父技能的工作是生成一份完整的报告。它会依次调用几个子技能1.数据收集技能可能进一步调用“爬取新闻技能”、“获取财报技能”2.数据分析技能3.图表生成技能4.报告合成技能。实操要点 组合模式中的技能像一棵树。根技能报告生成不直接处理具体任务而是协调子技能的工作。class CompositeReportSkill: def execute(self, topic): # 1. 协调数据收集 data_collector DataCollectionCompositeSkill() raw_data data_collector.execute(topic) # 内部可能调用多个数据源技能 # 2. 协调数据分析 analysis_result AnalysisSkill().execute(raw_data) # 3. 协调图表生成 charts ChartGenerationSkill().execute(analysis_result) # 4. 合成最终报告 final_report ReportSynthesisSkill().execute({ topic: topic, data: analysis_result, charts: charts }) return final_report # 数据收集本身也是一个组合技能 class DataCollectionCompositeSkill: def execute(self, topic): news NewsCrawlerSkill().execute(topic) financials FinancialDataSkill().execute(topic) return {news: news, financials: financials}为什么有效它提供了极大的灵活性。你可以像搭积木一样复用和替换子技能。例如更换一个更快的图表生成库只需修改对应的子技能父技能和其他部分完全不受影响。它也使得复杂技能的单元测试成为可能你可以单独测试每个叶子技能。注意事项依赖管理组合层次过深可能导致依赖关系复杂。要明确技能间的接口避免循环依赖。执行顺序与并行有些子技能可以并行执行以提升效率如同时爬取新闻和财报。在设计组合技能时需要考虑哪些步骤有先后依赖哪些可以并发。2.4 模式四装饰器模式装饰器模式允许你动态地为技能添加额外的功能而不改变其核心逻辑。这是实现横切关注点如日志记录、权限检查、重试机制、输入验证的绝佳方式。场景示例你有一个核心的“数据查询技能”。现在想为其增加1. 缓存功能相同查询直接返回缓存结果2. 速率限制防止被外部API限流3. 详细的执行日志。实操要点 通过高阶函数或装饰器类来包装核心技能函数。import functools import time from cachetools import TTLCache # 1. 缓存装饰器 def cache_skill(ttl300): # 缓存5分钟 cache TTLCache(maxsize100, ttlttl) def decorator(skill_func): functools.wraps(skill_func) def wrapper(*args, **kwargs): # 生成缓存键这里简单用参数拼接实际可能需要更复杂的序列化 key str(args) str(kwargs) if key in cache: print(f缓存命中: {key}) return cache[key] result skill_func(*args, **kwargs) cache[key] result return result return wrapper return decorator # 2. 重试装饰器 def retry_skill(max_retries3, delay1): def decorator(skill_func): functools.wraps(skill_func) def wrapper(*args, **kwargs): for i in range(max_retries): try: return skill_func(*args, **kwargs) except Exception as e: if i max_retries - 1: raise print(f技能执行失败第{i1}次重试... 错误: {e}) time.sleep(delay) return None return wrapper return decorator # 核心技能 retry_skill(max_retries2) cache_skill(ttl600) def core_data_query_skill(query_params): # 这里是实际的查询逻辑比如调用数据库或API print(f执行核心查询: {query_params}) # ... 模拟耗时操作 return {data: f结果 for {query_params}} # 使用装饰后的技能 result core_data_query_skill({user_id: 123})为什么有效它遵循了“开闭原则”——对扩展开放对修改关闭。你可以灵活地组合各种装饰器为技能叠加不同的能力而核心业务代码保持干净。这极大地提升了代码的可维护性和复用性。实操心得装饰器顺序装饰器的应用顺序很重要。通常缓存应该在重试之前因为如果缓存命中我们根本不需要执行重试逻辑。而日志记录可能需要在最外层以记录完整的调用过程。状态管理像缓存这类装饰器可能需要维护状态缓存字典需要注意线程安全或进程间共享的问题。2.5 模式五观察者模式事件驱动模式在这种模式下技能不是被主动调用而是订阅特定的事件或主题。当事件发生时所有订阅了该事件的技能会被自动触发。这非常适合构建松耦合、响应式的智能体系统。场景示例一个“项目协作”智能体。当代码仓库有新的推送Push Event时可以触发一系列技能1.自动运行CI/CD流水线的技能2.通知相关团队成员的技能3.更新项目文档状态的技能。实操要点 你需要一个简单的事件总线Event Bus或消息队列来管理事件和订阅者。class EventBus: def __init__(self): self.subscribers {} def subscribe(self, event_type, skill_callback): if event_type not in self.subscribers: self.subscribers[event_type] [] self.subscribers[event_type].append(skill_callback) def publish(self, event_type, event_data): if event_type in self.subscribers: for callback in self.subscribers[event_type]: # 异步执行避免阻塞 import threading threading.Thread(targetcallback, args(event_data,)).start() # 初始化事件总线 bus EventBus() # 定义技能事件处理器 def skill_ci_cd(event_data): print(fCI/CD技能被触发仓库{event_data[repo]}, 分支{event_data[branch]}) # 调用Jenkins/GitHub Actions API def skill_notify_team(event_data): print(f通知技能被触发作者{event_data[author]}) # 调用钉钉/飞书/Slack API # 技能订阅事件 bus.subscribe(code_push, skill_ci_cd) bus.subscribe(code_push, skill_notify_team) # 模拟事件发生 bus.publish(code_push, {repo: my-project, branch: main, author: 张三})为什么有效实现了彻底的解耦。事件发布者如GitHub的Webhook完全不知道有哪些技能会响应。新增一个技能如自动生成代码变更摘要只需让其订阅事件无需修改任何现有代码。系统的可扩展性极强。注意事项错误处理一个技能的失败不应影响其他技能的执行。每个技能需要有独立的错误处理机制。事件顺序与一致性如果技能之间有顺序依赖纯观察者模式可能不适用或者需要在事件数据或技能内部状态中体现顺序。性能考量大量事件和技能可能导致线程/进程激增需要考虑使用任务队列如Celery、RQ进行更专业的管理。3. 模式的选择与混合应用没有一种模式是银弹。在实际项目中你往往会混合使用多种模式。例如一个链式模式的某个环节可能内部是一个组合技能。链式模式的开头可能用一个路由技能来决定走哪条处理链。整个链路上的每个技能都可能被装饰器包裹添加日志和重试。当某个技能执行完成后可能会向事件总线发布一个事件触发其他后台任务观察者模式。选择模式的关键在于分析你所要解决问题的本质结构线性流程- 链式模式条件分支- 路由模式整体-部分关系- 组合模式添加辅助功能- 装饰器模式松散耦合的事件响应- 观察者模式4. 从模式到实践构建技能的工程化要点掌握了设计模式就像有了建筑设计图。但要盖好房子还需要关注工程细节。4.1 技能的统一接口与契约无论采用何种模式建议为所有技能定义一个统一的调用接口。这极大地提升了系统的可维护性和可测试性。from abc import ABC, abstractmethod from typing import Any, Dict class Skill(ABC): 技能抽象基类 abstractmethod def execute(self, input_data: Dict[str, Any], context: Dict[str, Any] None) - Dict[str, Any]: 执行技能。 :param input_data: 技能的主要输入数据。 :param context: 运行时上下文如会话ID、用户信息、历史记录。 :return: 执行结果字典。 pass property def name(self) - str: 技能名称 return self.__class__.__name__ property def description(self) - str: 技能功能描述可用于自动生成工具调用说明 return A generic skill4.2 提示词工程与工具调用的封装一个技能的核心往往是其提示词Prompt。好的技能设计会将易变的提示词部分参数化、模板化。class SummarizationSkill(Skill): def __init__(self, model_client): self.client model_client # 将提示词模板化 self.prompt_template 请对以下文本进行摘要要求 1. 抓住核心观点。 2. 字数控制在{word_limit}字以内。 3. 语言风格{tone}。 文本 {text} def execute(self, input_data, contextNone): text input_data.get(text) if not text: raise ValueError(输入数据中必须包含 text 字段) # 从输入或上下文中获取参数 word_limit input_data.get(word_limit, 200) tone input_data.get(tone, 专业) prompt self.prompt_template.format( texttext, word_limitword_limit, tonetone ) response self.client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}] ) summary response.choices[0].message.content return { original_text_length: len(text), summary: summary, word_count: len(summary) }实操心得将提示词存储在外部文件如YAML、JSON或数据库中可以在不修改代码的情况下动态调整技能行为这对于A/B测试和快速迭代至关重要。4.3 状态的持久化与记忆管理对于多轮对话或复杂任务技能可能需要访问历史信息。这就是记忆Memory的用武之地。记忆可以简单到是一个对话列表也可以复杂到是一个向量数据库用于检索相关历史片段。class SkillWithMemory(Skill): def __init__(self, core_skill, memory_store): self.core_skill core_skill self.memory_store memory_store # 可以是列表、数据库等 def execute(self, input_data, context): session_id context.get(session_id) if session_id: # 从记忆库中取出本次会话的历史 history self.memory_store.get(session_id, []) # 将历史信息作为上下文的一部分注入到本次执行中 enriched_input {**input_data, conversation_history: history} result self.core_skill.execute(enriched_input, context) # 将本次交互存入历史 new_entry {input: input_data, output: result} history.append(new_entry) self.memory_store.set(session_id, history) return result else: # 无会话ID直接执行核心技能 return self.core_skill.execute(input_data, context)5. 调试、监控与性能优化设计良好的技能模式也为调试和监控带来了便利。5.1 结构化日志与追踪在每个技能的execute方法开始和结束时记录结构化日志包含技能名、输入、输出、耗时和错误信息。这对于在复杂的链式或组合调用中定位问题不可或缺。import logging import time class LoggedSkill(Skill): def __init__(self, skill): self.skill skill self.logger logging.getLogger(self.skill.name) def execute(self, input_data, contextNone): self.logger.info(f开始执行技能 {self.skill.name}, extra{input: input_data}) start_time time.time() try: result self.skill.execute(input_data, context) duration time.time() - start_time self.logger.info(f技能 {self.skill.name} 执行成功耗时{duration:.2f}s, extra{output: result}) return result except Exception as e: duration time.time() - start_time self.logger.error(f技能 {self.skill.name} 执行失败耗时{duration:.2f}s, exc_infoTrue) raise5.2 性能考量异步、缓存与超时异步执行对于I/O密集型技能如调用外部API、查询数据库应使用异步模式如asyncio避免阻塞整个Agent。缓存策略如前所述装饰器模式非常适合实现缓存。对于计算成本高或结果变化不频繁的技能缓存能极大提升响应速度并降低成本。超时控制为每个技能设置合理的超时时间防止因某个技能挂起导致整个Agent无响应。这可以在装饰器中实现。import asyncio import functools from concurrent.futures import TimeoutError def timeout_skill(timeout_seconds): def decorator(skill_func): functools.wraps(skill_func) async def async_wrapper(*args, **kwargs): try: return await asyncio.wait_for(skill_func(*args, **kwargs), timeouttimeout_seconds) except TimeoutError: raise TimeoutError(f技能执行超时限制{timeout_seconds}秒) return async_wrapper return decorator6. 总结与个人体会回顾这五种设计模式——链式、路由、组合、装饰器、观察者它们共同描绘了一条从“脚本式”的Prompt堆砌走向“工程化”的智能体构建的路径。我个人在项目中的体会是模式的价值不在于生搬硬套而在于提供了一种高层次的抽象语言让团队能更清晰地讨论和设计系统架构。最开始我们可能只会写一个巨大的Prompt来处理所有事情。当它变得难以维护时我们本能地开始“切分”这就自然走到了链式或路由模式。当切分后的模块需要共享功能时装饰器模式就浮出水面。当任务本身具有层次性时组合模式提供了清晰的表达。而当系统需要高度解耦和扩展时事件驱动的观察者模式就成了不二之选。最后分享一个关键心得尽早建立技能的注册与发现机制。可以创建一个简单的“技能注册中心”让所有技能在启动时自行注册并声明其功能、输入输出格式。这样你的路由Agent或编排引擎就能动态地发现和调用可用技能实现真正的插件化架构。这为未来技能的动态加载、热更新和生态系统建设打下了基础。从设计模式出发最终落脚于扎实的工程实践这才是写好Agent Skills构建强大、可靠智能体应用的正道。 AI智能体技能设计：5大核心模式与工程实践指南
1. 项目概述为什么我们需要为AI智能体设计“技能”最近在GitHub上围绕“AI Skills Agent”的项目热度居高不下这背后反映了一个清晰的趋势单纯调用一个大语言模型LLM的API已经不够了。无论是构建一个能自动处理邮件的助手还是一个能分析数据并生成报告的分析师我们都需要将这些复杂的任务拆解成一个个可复用、可组合、可管理的“技能”Skills。这就像给一个聪明的实习生LLM配备了一套标准化的工具和操作手册让他不仅能理解指令还能高效、可靠地执行具体任务。“怎么写好Agent Skills”这个问题本质上是在问如何将LLM的泛化能力工程化为解决特定领域问题的可靠组件如果设计得不好你的智能体可能会陷入“幻觉”、执行步骤混乱、难以调试和维护的泥潭。我花了大量时间研究和实践发现借鉴软件工程中经典的设计模式思想是构建高质量Agent Skills的捷径。这些模式不是死板的规则而是经过验证的最佳实践蓝图能帮你规避常见陷阱提升技能的可读性、可维护性和复用性。接下来我就结合具体场景拆解5种最核心、最实用的Agent Skills设计模式。2. 核心设计模式解析从理念到实践设计模式的核心价值在于提供了一种共享的语言和思维框架。在Agent Skills的语境下我们关注的是如何组织提示词Prompt、工具调用Tool Calling、记忆Memory以及技能之间的协作关系。下面这五种模式基本覆盖了从简单到复杂的技能构建场景。2.1 模式一链式模式这是最基础也是最常用的模式灵感来源于责任链模式。其核心思想是将一个复杂任务分解为一系列顺序执行的子任务技能前一个技能的输出作为后一个技能的输入。场景示例一个“周报生成”智能体。任务不是直接让LLM“写周报”而是分解为1. 从日程工具中提取本周会议记录技能A2. 从代码仓库解析本周提交记录技能B3. 整合A和B的输出并按照固定模板生成周报草稿技能C4. 将草稿发送给用户确认技能D。实操要点与代码示意# 伪代码示例展示链式执行逻辑 class ChainOfSkillsAgent: def run(self, task_input): context task_input for skill in [skill_a, skill_b, skill_c, skill_d]: # 每个skill都是一个独立的函数或对象负责特定子任务 context skill.execute(context) # 此处可加入错误处理如果某个skill失败可中断链条或重试 if context.get(error): break return context # 技能A提取会议记录 def skill_a_extract_meetings(context): # 调用日历API的工具 meetings call_calendar_api(context[date_range]) # 可能用LLM总结会议要点 summary llm_call(f总结以下会议记录的关键点{meetings}) return {**context, meeting_summary: summary}为什么有效链式模式强制进行了任务分解使得每个技能保持单一职责Single Responsibility极大降低了单个技能的复杂度。调试时你可以清晰地看到数据在链条中的流动定位问题发生在哪个环节。此外技能可以独立开发和测试。注意事项错误传播链条中的一环失败会导致整个任务失败。必须设计健壮的错误处理机制比如为关键技能设置重试逻辑或提供“降级”技能如当无法获取数据时生成一个提示用户手动输入的回复。上下文管理随着链条变长传递的context对象可能变得臃肿。需要定义清晰的数据契约明确每个技能输入和输出的字段避免技能间产生隐式依赖。2.2 模式二路由模式当智能体需要根据输入内容或当前状态动态选择执行哪条技能路径时路由模式就派上用场了。这类似于策略模式或状态模式。场景示例一个“智能客服路由”Agent。用户输入一句话Agent需要判断其意图是“查询订单状态”、“投诉产品质量”还是“咨询退货政策”。根据判断结果路由到对应的专用技能模块进行处理。实操要点 实现路由的核心是一个“路由决策器”Router。这个决策器本身可以是一个小型的LLM调用也可以是基于规则或分类模型的判断。class RouterAgent: def route(self, user_query): # 使用LLM进行意图识别也可用更轻量的分类模型 intent_prompt f 用户说{user_query} 请判断其意图属于以下哪一类 A. 查询订单 B. 产品投诉 C. 退货咨询 D. 其他 只输出字母。 intent llm_call(intent_prompt).strip() # 根据意图路由到对应的技能 skill_map { A: skill_query_order, B: skill_handle_complaint, C: skill_return_policy, D: skill_fallback_general } target_skill skill_map.get(intent, skill_fallback_general) return target_skill.execute(user_query)为什么有效它避免了构建一个“全能”但臃肿的单一技能而是通过分工协作让每个技能在其专业领域内做到极致。系统易于扩展新增一个意图类别只需增加一个技能并更新路由表即可。实操心得设计清晰的决策边界意图分类的类别必须互斥且覆盖全面。模糊的类别会导致路由不准。在初期可以允许LLM输出“不确定”并路由到一个“澄清问题”的技能。路由器的性能如果每次请求都经过LLM做路由判断可能会增加延迟和成本。对于高频、固定的意图可以考虑训练一个轻量级的文本分类模型或者缓存常见的路由结果。2.3 模式三组合模式这个模式用于处理具有树状或层级结构的任务。一个父技能可以调用多个子技能并综合它们的结果。这与面向对象中的组合模式类似强调“部分-整体”的层次关系。场景示例一个“行业分析报告生成”技能。这个父技能的工作是生成一份完整的报告。它会依次调用几个子技能1.数据收集技能可能进一步调用“爬取新闻技能”、“获取财报技能”2.数据分析技能3.图表生成技能4.报告合成技能。实操要点 组合模式中的技能像一棵树。根技能报告生成不直接处理具体任务而是协调子技能的工作。class CompositeReportSkill: def execute(self, topic): # 1. 协调数据收集 data_collector DataCollectionCompositeSkill() raw_data data_collector.execute(topic) # 内部可能调用多个数据源技能 # 2. 协调数据分析 analysis_result AnalysisSkill().execute(raw_data) # 3. 协调图表生成 charts ChartGenerationSkill().execute(analysis_result) # 4. 合成最终报告 final_report ReportSynthesisSkill().execute({ topic: topic, data: analysis_result, charts: charts }) return final_report # 数据收集本身也是一个组合技能 class DataCollectionCompositeSkill: def execute(self, topic): news NewsCrawlerSkill().execute(topic) financials FinancialDataSkill().execute(topic) return {news: news, financials: financials}为什么有效它提供了极大的灵活性。你可以像搭积木一样复用和替换子技能。例如更换一个更快的图表生成库只需修改对应的子技能父技能和其他部分完全不受影响。它也使得复杂技能的单元测试成为可能你可以单独测试每个叶子技能。注意事项依赖管理组合层次过深可能导致依赖关系复杂。要明确技能间的接口避免循环依赖。执行顺序与并行有些子技能可以并行执行以提升效率如同时爬取新闻和财报。在设计组合技能时需要考虑哪些步骤有先后依赖哪些可以并发。2.4 模式四装饰器模式装饰器模式允许你动态地为技能添加额外的功能而不改变其核心逻辑。这是实现横切关注点如日志记录、权限检查、重试机制、输入验证的绝佳方式。场景示例你有一个核心的“数据查询技能”。现在想为其增加1. 缓存功能相同查询直接返回缓存结果2. 速率限制防止被外部API限流3. 详细的执行日志。实操要点 通过高阶函数或装饰器类来包装核心技能函数。import functools import time from cachetools import TTLCache # 1. 缓存装饰器 def cache_skill(ttl300): # 缓存5分钟 cache TTLCache(maxsize100, ttlttl) def decorator(skill_func): functools.wraps(skill_func) def wrapper(*args, **kwargs): # 生成缓存键这里简单用参数拼接实际可能需要更复杂的序列化 key str(args) str(kwargs) if key in cache: print(f缓存命中: {key}) return cache[key] result skill_func(*args, **kwargs) cache[key] result return result return wrapper return decorator # 2. 重试装饰器 def retry_skill(max_retries3, delay1): def decorator(skill_func): functools.wraps(skill_func) def wrapper(*args, **kwargs): for i in range(max_retries): try: return skill_func(*args, **kwargs) except Exception as e: if i max_retries - 1: raise print(f技能执行失败第{i1}次重试... 错误: {e}) time.sleep(delay) return None return wrapper return decorator # 核心技能 retry_skill(max_retries2) cache_skill(ttl600) def core_data_query_skill(query_params): # 这里是实际的查询逻辑比如调用数据库或API print(f执行核心查询: {query_params}) # ... 模拟耗时操作 return {data: f结果 for {query_params}} # 使用装饰后的技能 result core_data_query_skill({user_id: 123})为什么有效它遵循了“开闭原则”——对扩展开放对修改关闭。你可以灵活地组合各种装饰器为技能叠加不同的能力而核心业务代码保持干净。这极大地提升了代码的可维护性和复用性。实操心得装饰器顺序装饰器的应用顺序很重要。通常缓存应该在重试之前因为如果缓存命中我们根本不需要执行重试逻辑。而日志记录可能需要在最外层以记录完整的调用过程。状态管理像缓存这类装饰器可能需要维护状态缓存字典需要注意线程安全或进程间共享的问题。2.5 模式五观察者模式事件驱动模式在这种模式下技能不是被主动调用而是订阅特定的事件或主题。当事件发生时所有订阅了该事件的技能会被自动触发。这非常适合构建松耦合、响应式的智能体系统。场景示例一个“项目协作”智能体。当代码仓库有新的推送Push Event时可以触发一系列技能1.自动运行CI/CD流水线的技能2.通知相关团队成员的技能3.更新项目文档状态的技能。实操要点 你需要一个简单的事件总线Event Bus或消息队列来管理事件和订阅者。class EventBus: def __init__(self): self.subscribers {} def subscribe(self, event_type, skill_callback): if event_type not in self.subscribers: self.subscribers[event_type] [] self.subscribers[event_type].append(skill_callback) def publish(self, event_type, event_data): if event_type in self.subscribers: for callback in self.subscribers[event_type]: # 异步执行避免阻塞 import threading threading.Thread(targetcallback, args(event_data,)).start() # 初始化事件总线 bus EventBus() # 定义技能事件处理器 def skill_ci_cd(event_data): print(fCI/CD技能被触发仓库{event_data[repo]}, 分支{event_data[branch]}) # 调用Jenkins/GitHub Actions API def skill_notify_team(event_data): print(f通知技能被触发作者{event_data[author]}) # 调用钉钉/飞书/Slack API # 技能订阅事件 bus.subscribe(code_push, skill_ci_cd) bus.subscribe(code_push, skill_notify_team) # 模拟事件发生 bus.publish(code_push, {repo: my-project, branch: main, author: 张三})为什么有效实现了彻底的解耦。事件发布者如GitHub的Webhook完全不知道有哪些技能会响应。新增一个技能如自动生成代码变更摘要只需让其订阅事件无需修改任何现有代码。系统的可扩展性极强。注意事项错误处理一个技能的失败不应影响其他技能的执行。每个技能需要有独立的错误处理机制。事件顺序与一致性如果技能之间有顺序依赖纯观察者模式可能不适用或者需要在事件数据或技能内部状态中体现顺序。性能考量大量事件和技能可能导致线程/进程激增需要考虑使用任务队列如Celery、RQ进行更专业的管理。3. 模式的选择与混合应用没有一种模式是银弹。在实际项目中你往往会混合使用多种模式。例如一个链式模式的某个环节可能内部是一个组合技能。链式模式的开头可能用一个路由技能来决定走哪条处理链。整个链路上的每个技能都可能被装饰器包裹添加日志和重试。当某个技能执行完成后可能会向事件总线发布一个事件触发其他后台任务观察者模式。选择模式的关键在于分析你所要解决问题的本质结构线性流程- 链式模式条件分支- 路由模式整体-部分关系- 组合模式添加辅助功能- 装饰器模式松散耦合的事件响应- 观察者模式4. 从模式到实践构建技能的工程化要点掌握了设计模式就像有了建筑设计图。但要盖好房子还需要关注工程细节。4.1 技能的统一接口与契约无论采用何种模式建议为所有技能定义一个统一的调用接口。这极大地提升了系统的可维护性和可测试性。from abc import ABC, abstractmethod from typing import Any, Dict class Skill(ABC): 技能抽象基类 abstractmethod def execute(self, input_data: Dict[str, Any], context: Dict[str, Any] None) - Dict[str, Any]: 执行技能。 :param input_data: 技能的主要输入数据。 :param context: 运行时上下文如会话ID、用户信息、历史记录。 :return: 执行结果字典。 pass property def name(self) - str: 技能名称 return self.__class__.__name__ property def description(self) - str: 技能功能描述可用于自动生成工具调用说明 return A generic skill4.2 提示词工程与工具调用的封装一个技能的核心往往是其提示词Prompt。好的技能设计会将易变的提示词部分参数化、模板化。class SummarizationSkill(Skill): def __init__(self, model_client): self.client model_client # 将提示词模板化 self.prompt_template 请对以下文本进行摘要要求 1. 抓住核心观点。 2. 字数控制在{word_limit}字以内。 3. 语言风格{tone}。 文本 {text} def execute(self, input_data, contextNone): text input_data.get(text) if not text: raise ValueError(输入数据中必须包含 text 字段) # 从输入或上下文中获取参数 word_limit input_data.get(word_limit, 200) tone input_data.get(tone, 专业) prompt self.prompt_template.format( texttext, word_limitword_limit, tonetone ) response self.client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}] ) summary response.choices[0].message.content return { original_text_length: len(text), summary: summary, word_count: len(summary) }实操心得将提示词存储在外部文件如YAML、JSON或数据库中可以在不修改代码的情况下动态调整技能行为这对于A/B测试和快速迭代至关重要。4.3 状态的持久化与记忆管理对于多轮对话或复杂任务技能可能需要访问历史信息。这就是记忆Memory的用武之地。记忆可以简单到是一个对话列表也可以复杂到是一个向量数据库用于检索相关历史片段。class SkillWithMemory(Skill): def __init__(self, core_skill, memory_store): self.core_skill core_skill self.memory_store memory_store # 可以是列表、数据库等 def execute(self, input_data, context): session_id context.get(session_id) if session_id: # 从记忆库中取出本次会话的历史 history self.memory_store.get(session_id, []) # 将历史信息作为上下文的一部分注入到本次执行中 enriched_input {**input_data, conversation_history: history} result self.core_skill.execute(enriched_input, context) # 将本次交互存入历史 new_entry {input: input_data, output: result} history.append(new_entry) self.memory_store.set(session_id, history) return result else: # 无会话ID直接执行核心技能 return self.core_skill.execute(input_data, context)5. 调试、监控与性能优化设计良好的技能模式也为调试和监控带来了便利。5.1 结构化日志与追踪在每个技能的execute方法开始和结束时记录结构化日志包含技能名、输入、输出、耗时和错误信息。这对于在复杂的链式或组合调用中定位问题不可或缺。import logging import time class LoggedSkill(Skill): def __init__(self, skill): self.skill skill self.logger logging.getLogger(self.skill.name) def execute(self, input_data, contextNone): self.logger.info(f开始执行技能 {self.skill.name}, extra{input: input_data}) start_time time.time() try: result self.skill.execute(input_data, context) duration time.time() - start_time self.logger.info(f技能 {self.skill.name} 执行成功耗时{duration:.2f}s, extra{output: result}) return result except Exception as e: duration time.time() - start_time self.logger.error(f技能 {self.skill.name} 执行失败耗时{duration:.2f}s, exc_infoTrue) raise5.2 性能考量异步、缓存与超时异步执行对于I/O密集型技能如调用外部API、查询数据库应使用异步模式如asyncio避免阻塞整个Agent。缓存策略如前所述装饰器模式非常适合实现缓存。对于计算成本高或结果变化不频繁的技能缓存能极大提升响应速度并降低成本。超时控制为每个技能设置合理的超时时间防止因某个技能挂起导致整个Agent无响应。这可以在装饰器中实现。import asyncio import functools from concurrent.futures import TimeoutError def timeout_skill(timeout_seconds): def decorator(skill_func): functools.wraps(skill_func) async def async_wrapper(*args, **kwargs): try: return await asyncio.wait_for(skill_func(*args, **kwargs), timeouttimeout_seconds) except TimeoutError: raise TimeoutError(f技能执行超时限制{timeout_seconds}秒) return async_wrapper return decorator6. 总结与个人体会回顾这五种设计模式——链式、路由、组合、装饰器、观察者它们共同描绘了一条从“脚本式”的Prompt堆砌走向“工程化”的智能体构建的路径。我个人在项目中的体会是模式的价值不在于生搬硬套而在于提供了一种高层次的抽象语言让团队能更清晰地讨论和设计系统架构。最开始我们可能只会写一个巨大的Prompt来处理所有事情。当它变得难以维护时我们本能地开始“切分”这就自然走到了链式或路由模式。当切分后的模块需要共享功能时装饰器模式就浮出水面。当任务本身具有层次性时组合模式提供了清晰的表达。而当系统需要高度解耦和扩展时事件驱动的观察者模式就成了不二之选。最后分享一个关键心得尽早建立技能的注册与发现机制。可以创建一个简单的“技能注册中心”让所有技能在启动时自行注册并声明其功能、输入输出格式。这样你的路由Agent或编排引擎就能动态地发现和调用可用技能实现真正的插件化架构。这为未来技能的动态加载、热更新和生态系统建设打下了基础。从设计模式出发最终落脚于扎实的工程实践这才是写好Agent Skills构建强大、可靠智能体应用的正道。