1. 项目概述为什么ADK开发者需要关注Skill设计模式如果你正在使用或探索ADK进行Agent开发大概率已经体验过从零开始构建一个能稳定运行、逻辑清晰的Agent Skill技能的挑战。ADK即Agent Development Kit为开发者提供了构建智能体Agent的基础框架和工具集。但框架本身只是骨架真正让Agent具备实用价值、行为可控且易于维护的是运行在其上的一个个Skill。Skill可以理解为Agent的“能力单元”或“行为模块”它定义了Agent如何感知、决策和执行特定任务。在实际开发中我发现很多开发者尤其是刚接触ADK的同行容易陷入两个极端要么过度设计将一个简单的查询Skill拆解得过于复杂引入了不必要的抽象层导致后续维护和调试异常困难要么设计不足将所有逻辑都塞进一个庞大的、面条式的代码块里任何需求变更都牵一发而动全身测试更是无从下手。这两种情况都会严重拖慢开发进度降低代码质量。这正是设计模式的价值所在。设计模式并非银弹也不是必须严格遵守的教条而是一套经过实战检验的、针对特定问题的可复用解决方案模板。对于ADK Skill开发而言掌握几种核心的设计模式意味着你能用更结构化的方式组织代码让Skill的意图处理、上下文管理、外部服务调用、错误处理和流程编排变得清晰、健壮且可扩展。这不仅能提升你个人的开发效率更能让你构建出的Agent在复杂多变的真实场景中保持稳定和可靠。接下来我将结合自身在多个ADK项目中的实践经验为你拆解五种我认为最实用、最能解决ADK Skill开发痛点的设计模式并附上具体的实现思路和避坑指南。2. 核心设计模式深度解析与应用场景在ADK的语境下Skill设计模式需要特别关注几个核心维度事件驱动性如何响应来自用户或系统的触发、状态管理如何在多轮对话或异步任务中保持上下文、外部集成如何安全、高效地调用API或服务以及流程编排如何组织复杂的、多步骤的任务流。下面这五种模式正是围绕这些核心挑战展开的。2.1 模式一意图处理器模式这是ADK Skill开发中最基础、最常用的模式其核心思想是将不同的用户意图分发到对应的、单一职责的处理函数中。一个典型的Skill会处理多种用户请求例如一个“天气查询Skill”可能需要处理“查询当前天气”、“查询未来三天预报”、“查询空气质量”等不同意图。2.1.1 模式结构与实现要点传统的、缺乏设计的代码可能会用一个庞大的if-else或switch-case块来处理所有意图。意图处理器模式则主张建立一个“意图-处理器”的映射注册表。每个处理器都是一个独立的函数或类只负责处理一种意图。# 示例一个简单的意图处理器注册与分发框架 class IntentHandler: def __init__(self): self._handlers {} def register(self, intent_name: str, handler_func): 注册意图处理器 self._handlers[intent_name] handler_func async def handle(self, agent_context, user_input): 根据识别的意图分发请求 # 假设从agent_context或通过NLU解析出意图 intent agent_context.recognized_intent handler self._handlers.get(intent) if not handler: # 默认或回退处理器 return await self._handle_unknown(agent_context) return await handler(agent_context, user_input) # 定义具体的处理器 async def handle_weather_current(agent_context, user_input): location agent_context.get_slot(“location”) # 调用天气API weather_data await fetch_weather_api(location, “current”) # 组织Agent回复 return format_weather_response(weather_data) async def handle_weather_forecast(agent_context, user_input): # ... 处理预报逻辑 # 在Skill初始化时注册 handler IntentHandler() handler.register(“query_current_weather”, handle_weather_current) handler.register(“query_weather_forecast”, handle_weather_forecast)2.1.2 实操心得与避坑指南保持处理器纯洁性每个处理器应只做一件事——处理特定意图的业务逻辑。避免在处理器内进行复杂的意图识别或上下文修补这些工作应前置。依赖注入处理器如果需要调用外部服务如数据库、API客户端最好通过参数传入或从统一的依赖容器中获取而不是在内部硬编码创建。这极大提升了代码的可测试性。统一错误处理在分发器handle方法层面实现统一的异常捕获和日志记录而不是在每个处理器里重复写try-catch。这能确保所有意图处理都有一致的错误反馈机制。常见陷阱不要因为意图不多就放弃使用此模式。即使只有两个意图清晰的分离也能为未来的扩展第三个、第四个意图铺平道路。另一个陷阱是处理器之间共享可变状态这会导致难以追踪的Bug应通过上下文对象传递必要数据。2.2 模式二对话状态模式Agent的核心特征之一是支持多轮对话这就需要Skill能够记住对话的上下文。对话状态模式通过一个显式的、结构化的上下文对象来封装和管理单次对话会话中的所有状态信息。2.2.1 状态对象的设计与生命周期状态对象不应是一个全局变量或散落的多个变量而应该是一个与特定对话会话Session绑定的数据容器。它通常包括用户标识区分不同用户。会话ID标识唯一的对话流程。槽位从用户话语中提取的关键信息实体如{“location”: “北京”, “date”: “明天”}。历史消息近几轮的对话记录用于理解上下文。自定义业务状态如当前处于查询流程的哪一步“等待确认城市”、“等待选择日期”。class ConversationContext: def __init__(self, user_id: str, session_id: str): self.user_id user_id self.session_id session_id self.slots {} self.dialog_stack [] # 对话栈用于管理子对话 self._private_state {} # 业务自定义状态 def set_slot(self, key, value): self.slots[key] value def get_slot(self, key, defaultNone): return self.slots.get(key, default) def set_state(self, key, value): self._private_state[key] value def get_state(self, key, defaultNone): return self._private_state.get(key, default) # 在意图处理器中使用 async def handle_book_flight(agent_context, user_input): ctx agent_context.conversation_context if not ctx.get_slot(“departure_city”): # 进入“询问出发城市”的子状态 ctx.set_state(“awaiting_departure”, True) return “请问您从哪里出发” elif ctx.get_state(“awaiting_departure”): # 填充槽位并进入下一步 ctx.set_slot(“departure_city”, extract_city(user_input)) ctx.set_state(“awaiting_departure”, False) # ... 继续询问到达城市2.2.2 状态持久化与恢复策略对于Web或移动端Agent会话可能中断用户关闭页面/App。状态模式必须结合持久化策略。存储后端选择对于简单场景内存缓存如Redis足够对于需要长期记忆或重要业务应使用数据库。关键是将ConversationContext序列化如JSON后存储。序列化要点只存储必要数据避免存储无法序列化的对象如数据库连接、HTTP客户端。通常只存储slots和_private_state字典。恢复时机每次新的用户请求到来时根据user_id和session_id从存储中加载上下文反序列化后挂载到agent_context上。处理完请求后再将更新后的上下文保存回去。避坑指南状态对象容易变得臃肿。定期清理过时或无用的状态例如对话完成后清空相关槽位。另外要特别注意并发问题如果同一会话可能被近乎同时的多个请求处理虽然不常见需要考虑乐观锁等机制防止状态覆盖。2.3 模式三服务门面模式一个实用的Skill几乎必然需要与外部系统交互如调用REST API、访问数据库、发送邮件等。服务门面模式为这些外部依赖提供一个统一的、简化的内部接口将复杂的集成逻辑封装在Skill核心业务逻辑之外。2.3.1 门面层的设计与封装价值直接在意图处理器里写requests.get()或数据库查询语句是灾难性的。这会导致代码重复相同的API调用散落在各处。难以测试业务逻辑与网络I/O、数据库耦合无法进行单元测试。难以维护当外部接口变更时你需要修改所有调用它的地方。服务门面模式通过创建专门的客户端类来解决这些问题。# 外部天气API的原始调用反面教材散落在业务代码中 # async def handle_weather_current(...): # url f“https://api.weather.com/v3/...location{location}” # async with aiohttp.ClientSession() as session: # async with session.get(url, headers{...}) as resp: # data await resp.json() # # 解析data... # 服务门面模式 class WeatherServiceClient: def __init__(self, api_key: str, base_url: str, http_client): self.api_key api_key self.base_url base_url self._client http_client # 依赖注入的HTTP客户端 async def get_current_weather(self, location: str) - dict: 获取当前天气返回结构化的数据 endpoint f“{self.base_url}/current” params {“location”: location, “key”: self.api_key} try: data await self._client.get_json(endpoint, paramsparams) # 在这里进行统一的数据清洗、错误码转换、格式化 return self._normalize_weather_data(data) except ClientError as e: # 统一处理网络或API错误可能转换为业务异常 raise ServiceUnavailableError(f“Weather service error: {e}”) from e def _normalize_weather_data(self, raw_data: dict) - dict: # 将不同API的异构数据格式转换为Skill内部统一格式 return { “temp”: raw_data[“main”][“temp”], “description”: raw_data[“weather”][0][“description”], “humidity”: raw_data[“main”][“humidity”] } # 在Skill初始化时创建并注入 weather_client WeatherServiceClient(api_key“YOUR_KEY”, base_url“...”, http_clientaiohttp.ClientSession()) # 将其作为依赖提供给意图处理器可通过构造函数或上下文2.3.2 高级技巧熔断、重试与缓存在门面层你可以集中实现增强鲁棒性的策略重试逻辑对于暂时的网络故障可以自动重试。使用指数退避算法并设置最大重试次数。熔断器当外部服务连续失败时快速失败直接返回降级结果如缓存的老数据或默认信息避免积压请求拖垮整个Agent。一段时间后再尝试恢复。缓存对于不常变的数据如城市信息在门面层增加缓存减少不必要的对外调用提升响应速度。实操提示将这些策略重试、熔断的实现也抽象出来作为可插拔的组件装饰在HTTP客户端上这样不同的服务门面可以灵活配置不同的策略。2.4 模式四责任链模式当Skill需要处理一个可能由多个步骤或条件检查构成的流程时责任链模式非常有用。它将请求的发送者与接收者解耦使多个对象都有机会处理该请求并将这些对象连接成一条链请求沿着链传递直到被处理为止。2.4.1 在Skill流程编排中的应用想象一个“订单查询”Skill用户输入一个订单号后Skill需要1) 验证订单号格式2) 检查用户是否有权查看此订单3) 从数据库获取订单详情4) 从物流系统获取物流状态5) 组装最终回复。如果将这些步骤全部写在一个函数里函数会非常长且难以修改。责任链模式可以将每个步骤抽象为一个“处理器”并链接起来。from abc import ABC, abstractmethod class OrderHandler(ABC): 责任链中的处理器基类 def __init__(self): self._next_handler None def set_next(self, handler): self._next_handler handler return handler # 支持链式调用 async def handle(self, agent_context, order_id): # 当前处理器处理逻辑 processed await self._process(agent_context, order_id) # 如果处理了或者无论处理与否都传递给下一个 if self._next_handler: return await self._next_handler.handle(agent_context, order_id) return processed abstractmethod async def _process(self, agent_context, order_id): pass class FormatValidationHandler(OrderHandler): async def _process(self, agent_context, order_id): if not re.match(r“^ORD\d{10}$”, order_id): agent_context.set_response(“订单号格式不正确。”) return True # 已处理链可以终止取决于设计 return False # 未处理继续传递 class AuthorizationHandler(OrderHandler): async def _process(self, agent_context, order_id): user agent_context.user if not order_service.can_user_view_order(user, order_id): agent_context.set_response(“您无权查看此订单。”) return True return False class DataFetchHandler(OrderHandler): async def _process(self, agent_context, order_id): order_detail await order_service.fetch_details(order_id) agent_context.set_state(“order_detail”, order_detail) return False # 继续传递让下一个处理器添加物流信息 # 构建责任链 chain FormatValidationHandler() chain.set_next(AuthorizationHandler()).set_next(DataFetchHandler()).set_next(LogisticsHandler()) # 在意图处理器中调用链 async def handle_order_query(agent_context, user_input): order_id agent_context.get_slot(“order_number”) await chain.handle(agent_context, order_id) # 最终回复可能在链的某个环节被设置或在最后统一组装2.4.2 灵活性与控制力责任链模式提供了极大的灵活性动态编排你可以根据运行时条件动态地构建不同的处理链。例如对于VIP用户跳过某些验证步骤。单一职责每个处理器只关心自己的那部分逻辑代码更清晰。易于测试每个处理器可以独立进行单元测试。注意事项要明确链的终止条件。是每个处理器都必须执行还是某个处理器处理后就可以终止上面的示例展示了“处理即可能终止”的模式。你需要根据业务逻辑仔细设计_process方法的返回值语义。另外要小心循环引用或过长的链影响性能。2.5 模式五策略模式当Skill需要根据不同的条件、用户偏好或系统状态在多种算法或策略中选择一种来执行同一类操作时策略模式是理想选择。它定义了算法家族并使其可以相互替换让算法的变化独立于使用它的客户端。2.5.1 实现动态行为选择一个典型的例子是回复生成策略。同一个查询结果针对不同的渠道语音助手、文字聊天、邮件或不同的用户级别新手、专家可能需要不同详细程度或格式的回复。from abc import ABC, abstractmethod class ResponseStrategy(ABC): 回复策略接口 abstractmethod async def generate(self, data: dict, context: ConversationContext) - str: pass class SimpleTextStrategy(ResponseStrategy): 简单文本回复用于即时通讯 async def generate(self, data, context): return f“当前温度{data[‘temp’]}度天气{data[‘description’]}。” class DetailedHtmlStrategy(ResponseStrategy): 详细HTML回复用于邮件或网页 async def generate(self, data, context): return f“”” h3天气报告/h3 p温度{data[‘temp’]}°C/p p状况{data[‘description’]}/p p湿度{data[‘humidity’]}%/p “”” class VoiceOptimizedStrategy(ResponseStrategy): 语音优化回复用于智能音箱 async def generate(self, data, context): # 生成更口语化、适合朗读的文本 return f“现在室外温度是{data[‘temp’]}摄氏度{data[‘description’]}。” class ResponseContext: 策略上下文负责选择和使用策略 def __init__(self, strategy: ResponseStrategy None): self._strategy strategy def set_strategy(self, strategy: ResponseStrategy): self._strategy strategy async def execute_strategy(self, data, context): if not self._strategy: # 默认策略 self._strategy SimpleTextStrategy() return await self._strategy.generate(data, context) # 在Skill中使用 async def handle_weather_response(agent_context, weather_data): response_context ResponseContext() # 根据渠道动态选择策略 channel agent_context.get_channel() if channel “email”: response_context.set_strategy(DetailedHtmlStrategy()) elif channel “voice”: response_context.set_strategy(VoiceOptimizedStrategy()) # else 默认使用 SimpleTextStrategy reply await response_context.execute_strategy(weather_data, agent_context.conversation_context) agent_context.set_response(reply)2.5.2 策略模式的扩展与配置化策略模式的优势在于其可扩展性。当需要新增一种回复格式如Markdown时你只需新增一个实现了ResponseStrategy接口的类并在上下文中添加相应的选择逻辑即可无需修改任何已有的策略类或主要的业务逻辑。更进一步你可以将策略的选择逻辑配置化。例如在Skill的配置文件中定义渠道与策略类的映射关系这样在增加新渠道时只需要更新配置文件而无需修改代码。# config.yaml response_strategies: web_chat: “skill.strategies.SimpleTextStrategy” email: “skill.strategies.DetailedHtmlStrategy” voice: “skill.strategies.VoiceOptimizedStrategy” slack: “skill.strategies.SlackMarkdownStrategy”2.5.3 避坑指南避免过度使用策略模式。如果策略之间只有细微差别比如只是字符串模板不同那么使用简单的配置或模板引擎可能更合适。策略模式适用于算法或行为逻辑有本质不同的情况。另外确保所有策略类的接口方法签名是完全一致的否则替换会出问题。3. 模式组合与实战架构设计在实际的ADK Skill项目中这些模式很少孤立使用而是根据复杂度组合应用形成清晰的架构。下面以一个中等复杂度的“智能旅行助手Skill”为例勾勒其核心架构。3.1 架构分层示意接入层/触发器由ADK框架处理接收用户输入初始化AgentContext和ConversationContext对话状态模式。意图路由层根据NLU解析的意图通过意图处理器模式的分发器将请求路由到对应的主意图处理器如handle_flight_booking,handle_hotel_search。核心业务流程层这是业务逻辑的核心。例如handle_flight_booking处理器内部可能会启动一个责任链SlotFillingHandler检查并补全出发地、目的地、时间等槽位。BudgetCheckHandler根据用户历史或设定检查预算。VendorSelectionHandler根据策略策略模式选择机票供应商如价格优先、时间优先、航司偏好。服务调用层责任链中的处理器或策略类通过服务门面模式创建的各种客户端FlightAPIClient,HotelAPIClient,PaymentServiceClient与外部系统交互。这些客户端内部封装了重试、熔断逻辑。响应组装层获取到外部数据后根据用户渠道使用策略模式选择合适的ResponseStrategy来生成最终回复。状态持久化层在整个流程的最后将更新后的ConversationContext序列化并保存到存储中完成本次交互。3.2 代码组织建议建议按功能模块而非技术分层来组织代码目录这样更符合Skill的“能力单元”特性。my_travel_skill/ ├── __init__.py ├── intents/ # 意图处理器目录 │ ├── __init__.py │ ├── flight.py # 包含 handle_flight_booking 等 │ ├── hotel.py │ └── weather.py ├── services/ # 服务门面目录 │ ├── __init__.py │ ├── flight_client.py │ ├── hotel_client.py │ └── payment_client.py ├── chains/ # 责任链处理器目录 │ ├── __init__.py │ ├── booking_chain.py │ └── search_chain.py ├── strategies/ # 策略目录 │ ├── __init__.py │ ├── response_strategies.py │ └── vendor_strategies.py ├── models/ # 数据模型上下文、状态对象 │ ├── __init__.py │ └── context.py └── config.py # 配置如策略映射这种结构让每个模式的作用域清晰可见便于团队协作和代码维护。4. 常见问题、调试技巧与性能考量即使采用了良好的设计模式在开发过程中仍会遇到各种问题。以下是一些常见陷阱和解决思路。4.1 上下文状态丢失或混乱问题用户在多轮对话中信息突然“失忆”或者A用户看到了B用户的信息。排查首先检查ConversationContext的session_id生成和传递逻辑是否正确。确保每次请求都能正确关联到之前的会话。检查状态持久化逻辑。是否在每次修改上下文后都成功保存序列化/反序列化过程是否有数据丢失特别是自定义对象检查是否有全局变量或类变量被误用来存储会话状态。这是导致状态混乱的常见原因。技巧为ConversationContext添加一个版本号或最后修改时间戳。在保存时进行乐观锁检查可以有效防止并发导致的状态覆盖。4.2 外部服务调用超时或失败导致Skill卡死问题调用某个第三方API没有响应整个Skill线程被阻塞后续请求无法处理。解决必须设置超时在所有外部HTTP/数据库调用中显式设置超时参数。使用异步编程确保你的ADK Skill框架和你的代码是异步的如使用asyncio。这样当一个请求在等待IO时其他请求可以被处理。实施熔断和降级在服务门面中集成熔断器如pybreaker。当失败率达到阈值直接快速返回一个友好的降级信息如“服务暂时不可用请稍后再试”而不是一直等待超时。实操命令使用curl或postman模拟慢速网络测试你的Skill的超时和降级响应是否正常工作。4.3 责任链或流程逻辑变得难以追踪问题责任链太长或者流程中条件分支太多出现Bug时很难定位是哪个环节出了问题。解决结构化日志在每个处理器的入口和出口打上带有唯一请求ID和处理器名称的日志。使用JSON格式的日志便于后续用ELK等工具分析。链路追踪在复杂的分布式Agent系统中可以考虑集成OpenTelemetry等链路追踪工具可视化请求在各个环节的流转和耗时。设计复审如果链过长考虑是否可以将一些步骤合并或者拆分成多个更小的、可复用的子链。4.4 性能瓶颈分析对于高频使用的Skill性能至关重要。** profiling**使用Python的cProfile模块或py-spy等工具找出代码中的热点函数。瓶颈往往出现在复杂的字符串处理、低效的循环、频繁的数据库查询或序列化/反序列化操作。缓存应用数据缓存对不常变的外部数据如城市列表、产品目录使用内存缓存如functools.lru_cache或 Redis。上下文缓存对于活跃会话的上下文可以缓存在内存中而不是每次请求都读写数据库但要注意缓存失效和同步策略。连接池确保你的数据库客户端、HTTP客户端使用了连接池避免频繁创建和销毁连接的开销。掌握这五种设计模式并理解它们如何组合运用能让你在ADK Skill开发中从“能实现功能”进阶到“能设计出健壮、可维护、可扩展的优秀技能”。模式是工具最终目的是为了写出更清晰的代码更从容地应对需求变化。在实际项目中不必追求完美的模式应用而是从最痛点入手逐步重构让代码结构向更好的方向演化。 ADK Skill开发五大设计模式：构建健壮可维护的智能体技能
1. 项目概述为什么ADK开发者需要关注Skill设计模式如果你正在使用或探索ADK进行Agent开发大概率已经体验过从零开始构建一个能稳定运行、逻辑清晰的Agent Skill技能的挑战。ADK即Agent Development Kit为开发者提供了构建智能体Agent的基础框架和工具集。但框架本身只是骨架真正让Agent具备实用价值、行为可控且易于维护的是运行在其上的一个个Skill。Skill可以理解为Agent的“能力单元”或“行为模块”它定义了Agent如何感知、决策和执行特定任务。在实际开发中我发现很多开发者尤其是刚接触ADK的同行容易陷入两个极端要么过度设计将一个简单的查询Skill拆解得过于复杂引入了不必要的抽象层导致后续维护和调试异常困难要么设计不足将所有逻辑都塞进一个庞大的、面条式的代码块里任何需求变更都牵一发而动全身测试更是无从下手。这两种情况都会严重拖慢开发进度降低代码质量。这正是设计模式的价值所在。设计模式并非银弹也不是必须严格遵守的教条而是一套经过实战检验的、针对特定问题的可复用解决方案模板。对于ADK Skill开发而言掌握几种核心的设计模式意味着你能用更结构化的方式组织代码让Skill的意图处理、上下文管理、外部服务调用、错误处理和流程编排变得清晰、健壮且可扩展。这不仅能提升你个人的开发效率更能让你构建出的Agent在复杂多变的真实场景中保持稳定和可靠。接下来我将结合自身在多个ADK项目中的实践经验为你拆解五种我认为最实用、最能解决ADK Skill开发痛点的设计模式并附上具体的实现思路和避坑指南。2. 核心设计模式深度解析与应用场景在ADK的语境下Skill设计模式需要特别关注几个核心维度事件驱动性如何响应来自用户或系统的触发、状态管理如何在多轮对话或异步任务中保持上下文、外部集成如何安全、高效地调用API或服务以及流程编排如何组织复杂的、多步骤的任务流。下面这五种模式正是围绕这些核心挑战展开的。2.1 模式一意图处理器模式这是ADK Skill开发中最基础、最常用的模式其核心思想是将不同的用户意图分发到对应的、单一职责的处理函数中。一个典型的Skill会处理多种用户请求例如一个“天气查询Skill”可能需要处理“查询当前天气”、“查询未来三天预报”、“查询空气质量”等不同意图。2.1.1 模式结构与实现要点传统的、缺乏设计的代码可能会用一个庞大的if-else或switch-case块来处理所有意图。意图处理器模式则主张建立一个“意图-处理器”的映射注册表。每个处理器都是一个独立的函数或类只负责处理一种意图。# 示例一个简单的意图处理器注册与分发框架 class IntentHandler: def __init__(self): self._handlers {} def register(self, intent_name: str, handler_func): 注册意图处理器 self._handlers[intent_name] handler_func async def handle(self, agent_context, user_input): 根据识别的意图分发请求 # 假设从agent_context或通过NLU解析出意图 intent agent_context.recognized_intent handler self._handlers.get(intent) if not handler: # 默认或回退处理器 return await self._handle_unknown(agent_context) return await handler(agent_context, user_input) # 定义具体的处理器 async def handle_weather_current(agent_context, user_input): location agent_context.get_slot(“location”) # 调用天气API weather_data await fetch_weather_api(location, “current”) # 组织Agent回复 return format_weather_response(weather_data) async def handle_weather_forecast(agent_context, user_input): # ... 处理预报逻辑 # 在Skill初始化时注册 handler IntentHandler() handler.register(“query_current_weather”, handle_weather_current) handler.register(“query_weather_forecast”, handle_weather_forecast)2.1.2 实操心得与避坑指南保持处理器纯洁性每个处理器应只做一件事——处理特定意图的业务逻辑。避免在处理器内进行复杂的意图识别或上下文修补这些工作应前置。依赖注入处理器如果需要调用外部服务如数据库、API客户端最好通过参数传入或从统一的依赖容器中获取而不是在内部硬编码创建。这极大提升了代码的可测试性。统一错误处理在分发器handle方法层面实现统一的异常捕获和日志记录而不是在每个处理器里重复写try-catch。这能确保所有意图处理都有一致的错误反馈机制。常见陷阱不要因为意图不多就放弃使用此模式。即使只有两个意图清晰的分离也能为未来的扩展第三个、第四个意图铺平道路。另一个陷阱是处理器之间共享可变状态这会导致难以追踪的Bug应通过上下文对象传递必要数据。2.2 模式二对话状态模式Agent的核心特征之一是支持多轮对话这就需要Skill能够记住对话的上下文。对话状态模式通过一个显式的、结构化的上下文对象来封装和管理单次对话会话中的所有状态信息。2.2.1 状态对象的设计与生命周期状态对象不应是一个全局变量或散落的多个变量而应该是一个与特定对话会话Session绑定的数据容器。它通常包括用户标识区分不同用户。会话ID标识唯一的对话流程。槽位从用户话语中提取的关键信息实体如{“location”: “北京”, “date”: “明天”}。历史消息近几轮的对话记录用于理解上下文。自定义业务状态如当前处于查询流程的哪一步“等待确认城市”、“等待选择日期”。class ConversationContext: def __init__(self, user_id: str, session_id: str): self.user_id user_id self.session_id session_id self.slots {} self.dialog_stack [] # 对话栈用于管理子对话 self._private_state {} # 业务自定义状态 def set_slot(self, key, value): self.slots[key] value def get_slot(self, key, defaultNone): return self.slots.get(key, default) def set_state(self, key, value): self._private_state[key] value def get_state(self, key, defaultNone): return self._private_state.get(key, default) # 在意图处理器中使用 async def handle_book_flight(agent_context, user_input): ctx agent_context.conversation_context if not ctx.get_slot(“departure_city”): # 进入“询问出发城市”的子状态 ctx.set_state(“awaiting_departure”, True) return “请问您从哪里出发” elif ctx.get_state(“awaiting_departure”): # 填充槽位并进入下一步 ctx.set_slot(“departure_city”, extract_city(user_input)) ctx.set_state(“awaiting_departure”, False) # ... 继续询问到达城市2.2.2 状态持久化与恢复策略对于Web或移动端Agent会话可能中断用户关闭页面/App。状态模式必须结合持久化策略。存储后端选择对于简单场景内存缓存如Redis足够对于需要长期记忆或重要业务应使用数据库。关键是将ConversationContext序列化如JSON后存储。序列化要点只存储必要数据避免存储无法序列化的对象如数据库连接、HTTP客户端。通常只存储slots和_private_state字典。恢复时机每次新的用户请求到来时根据user_id和session_id从存储中加载上下文反序列化后挂载到agent_context上。处理完请求后再将更新后的上下文保存回去。避坑指南状态对象容易变得臃肿。定期清理过时或无用的状态例如对话完成后清空相关槽位。另外要特别注意并发问题如果同一会话可能被近乎同时的多个请求处理虽然不常见需要考虑乐观锁等机制防止状态覆盖。2.3 模式三服务门面模式一个实用的Skill几乎必然需要与外部系统交互如调用REST API、访问数据库、发送邮件等。服务门面模式为这些外部依赖提供一个统一的、简化的内部接口将复杂的集成逻辑封装在Skill核心业务逻辑之外。2.3.1 门面层的设计与封装价值直接在意图处理器里写requests.get()或数据库查询语句是灾难性的。这会导致代码重复相同的API调用散落在各处。难以测试业务逻辑与网络I/O、数据库耦合无法进行单元测试。难以维护当外部接口变更时你需要修改所有调用它的地方。服务门面模式通过创建专门的客户端类来解决这些问题。# 外部天气API的原始调用反面教材散落在业务代码中 # async def handle_weather_current(...): # url f“https://api.weather.com/v3/...location{location}” # async with aiohttp.ClientSession() as session: # async with session.get(url, headers{...}) as resp: # data await resp.json() # # 解析data... # 服务门面模式 class WeatherServiceClient: def __init__(self, api_key: str, base_url: str, http_client): self.api_key api_key self.base_url base_url self._client http_client # 依赖注入的HTTP客户端 async def get_current_weather(self, location: str) - dict: 获取当前天气返回结构化的数据 endpoint f“{self.base_url}/current” params {“location”: location, “key”: self.api_key} try: data await self._client.get_json(endpoint, paramsparams) # 在这里进行统一的数据清洗、错误码转换、格式化 return self._normalize_weather_data(data) except ClientError as e: # 统一处理网络或API错误可能转换为业务异常 raise ServiceUnavailableError(f“Weather service error: {e}”) from e def _normalize_weather_data(self, raw_data: dict) - dict: # 将不同API的异构数据格式转换为Skill内部统一格式 return { “temp”: raw_data[“main”][“temp”], “description”: raw_data[“weather”][0][“description”], “humidity”: raw_data[“main”][“humidity”] } # 在Skill初始化时创建并注入 weather_client WeatherServiceClient(api_key“YOUR_KEY”, base_url“...”, http_clientaiohttp.ClientSession()) # 将其作为依赖提供给意图处理器可通过构造函数或上下文2.3.2 高级技巧熔断、重试与缓存在门面层你可以集中实现增强鲁棒性的策略重试逻辑对于暂时的网络故障可以自动重试。使用指数退避算法并设置最大重试次数。熔断器当外部服务连续失败时快速失败直接返回降级结果如缓存的老数据或默认信息避免积压请求拖垮整个Agent。一段时间后再尝试恢复。缓存对于不常变的数据如城市信息在门面层增加缓存减少不必要的对外调用提升响应速度。实操提示将这些策略重试、熔断的实现也抽象出来作为可插拔的组件装饰在HTTP客户端上这样不同的服务门面可以灵活配置不同的策略。2.4 模式四责任链模式当Skill需要处理一个可能由多个步骤或条件检查构成的流程时责任链模式非常有用。它将请求的发送者与接收者解耦使多个对象都有机会处理该请求并将这些对象连接成一条链请求沿着链传递直到被处理为止。2.4.1 在Skill流程编排中的应用想象一个“订单查询”Skill用户输入一个订单号后Skill需要1) 验证订单号格式2) 检查用户是否有权查看此订单3) 从数据库获取订单详情4) 从物流系统获取物流状态5) 组装最终回复。如果将这些步骤全部写在一个函数里函数会非常长且难以修改。责任链模式可以将每个步骤抽象为一个“处理器”并链接起来。from abc import ABC, abstractmethod class OrderHandler(ABC): 责任链中的处理器基类 def __init__(self): self._next_handler None def set_next(self, handler): self._next_handler handler return handler # 支持链式调用 async def handle(self, agent_context, order_id): # 当前处理器处理逻辑 processed await self._process(agent_context, order_id) # 如果处理了或者无论处理与否都传递给下一个 if self._next_handler: return await self._next_handler.handle(agent_context, order_id) return processed abstractmethod async def _process(self, agent_context, order_id): pass class FormatValidationHandler(OrderHandler): async def _process(self, agent_context, order_id): if not re.match(r“^ORD\d{10}$”, order_id): agent_context.set_response(“订单号格式不正确。”) return True # 已处理链可以终止取决于设计 return False # 未处理继续传递 class AuthorizationHandler(OrderHandler): async def _process(self, agent_context, order_id): user agent_context.user if not order_service.can_user_view_order(user, order_id): agent_context.set_response(“您无权查看此订单。”) return True return False class DataFetchHandler(OrderHandler): async def _process(self, agent_context, order_id): order_detail await order_service.fetch_details(order_id) agent_context.set_state(“order_detail”, order_detail) return False # 继续传递让下一个处理器添加物流信息 # 构建责任链 chain FormatValidationHandler() chain.set_next(AuthorizationHandler()).set_next(DataFetchHandler()).set_next(LogisticsHandler()) # 在意图处理器中调用链 async def handle_order_query(agent_context, user_input): order_id agent_context.get_slot(“order_number”) await chain.handle(agent_context, order_id) # 最终回复可能在链的某个环节被设置或在最后统一组装2.4.2 灵活性与控制力责任链模式提供了极大的灵活性动态编排你可以根据运行时条件动态地构建不同的处理链。例如对于VIP用户跳过某些验证步骤。单一职责每个处理器只关心自己的那部分逻辑代码更清晰。易于测试每个处理器可以独立进行单元测试。注意事项要明确链的终止条件。是每个处理器都必须执行还是某个处理器处理后就可以终止上面的示例展示了“处理即可能终止”的模式。你需要根据业务逻辑仔细设计_process方法的返回值语义。另外要小心循环引用或过长的链影响性能。2.5 模式五策略模式当Skill需要根据不同的条件、用户偏好或系统状态在多种算法或策略中选择一种来执行同一类操作时策略模式是理想选择。它定义了算法家族并使其可以相互替换让算法的变化独立于使用它的客户端。2.5.1 实现动态行为选择一个典型的例子是回复生成策略。同一个查询结果针对不同的渠道语音助手、文字聊天、邮件或不同的用户级别新手、专家可能需要不同详细程度或格式的回复。from abc import ABC, abstractmethod class ResponseStrategy(ABC): 回复策略接口 abstractmethod async def generate(self, data: dict, context: ConversationContext) - str: pass class SimpleTextStrategy(ResponseStrategy): 简单文本回复用于即时通讯 async def generate(self, data, context): return f“当前温度{data[‘temp’]}度天气{data[‘description’]}。” class DetailedHtmlStrategy(ResponseStrategy): 详细HTML回复用于邮件或网页 async def generate(self, data, context): return f“”” h3天气报告/h3 p温度{data[‘temp’]}°C/p p状况{data[‘description’]}/p p湿度{data[‘humidity’]}%/p “”” class VoiceOptimizedStrategy(ResponseStrategy): 语音优化回复用于智能音箱 async def generate(self, data, context): # 生成更口语化、适合朗读的文本 return f“现在室外温度是{data[‘temp’]}摄氏度{data[‘description’]}。” class ResponseContext: 策略上下文负责选择和使用策略 def __init__(self, strategy: ResponseStrategy None): self._strategy strategy def set_strategy(self, strategy: ResponseStrategy): self._strategy strategy async def execute_strategy(self, data, context): if not self._strategy: # 默认策略 self._strategy SimpleTextStrategy() return await self._strategy.generate(data, context) # 在Skill中使用 async def handle_weather_response(agent_context, weather_data): response_context ResponseContext() # 根据渠道动态选择策略 channel agent_context.get_channel() if channel “email”: response_context.set_strategy(DetailedHtmlStrategy()) elif channel “voice”: response_context.set_strategy(VoiceOptimizedStrategy()) # else 默认使用 SimpleTextStrategy reply await response_context.execute_strategy(weather_data, agent_context.conversation_context) agent_context.set_response(reply)2.5.2 策略模式的扩展与配置化策略模式的优势在于其可扩展性。当需要新增一种回复格式如Markdown时你只需新增一个实现了ResponseStrategy接口的类并在上下文中添加相应的选择逻辑即可无需修改任何已有的策略类或主要的业务逻辑。更进一步你可以将策略的选择逻辑配置化。例如在Skill的配置文件中定义渠道与策略类的映射关系这样在增加新渠道时只需要更新配置文件而无需修改代码。# config.yaml response_strategies: web_chat: “skill.strategies.SimpleTextStrategy” email: “skill.strategies.DetailedHtmlStrategy” voice: “skill.strategies.VoiceOptimizedStrategy” slack: “skill.strategies.SlackMarkdownStrategy”2.5.3 避坑指南避免过度使用策略模式。如果策略之间只有细微差别比如只是字符串模板不同那么使用简单的配置或模板引擎可能更合适。策略模式适用于算法或行为逻辑有本质不同的情况。另外确保所有策略类的接口方法签名是完全一致的否则替换会出问题。3. 模式组合与实战架构设计在实际的ADK Skill项目中这些模式很少孤立使用而是根据复杂度组合应用形成清晰的架构。下面以一个中等复杂度的“智能旅行助手Skill”为例勾勒其核心架构。3.1 架构分层示意接入层/触发器由ADK框架处理接收用户输入初始化AgentContext和ConversationContext对话状态模式。意图路由层根据NLU解析的意图通过意图处理器模式的分发器将请求路由到对应的主意图处理器如handle_flight_booking,handle_hotel_search。核心业务流程层这是业务逻辑的核心。例如handle_flight_booking处理器内部可能会启动一个责任链SlotFillingHandler检查并补全出发地、目的地、时间等槽位。BudgetCheckHandler根据用户历史或设定检查预算。VendorSelectionHandler根据策略策略模式选择机票供应商如价格优先、时间优先、航司偏好。服务调用层责任链中的处理器或策略类通过服务门面模式创建的各种客户端FlightAPIClient,HotelAPIClient,PaymentServiceClient与外部系统交互。这些客户端内部封装了重试、熔断逻辑。响应组装层获取到外部数据后根据用户渠道使用策略模式选择合适的ResponseStrategy来生成最终回复。状态持久化层在整个流程的最后将更新后的ConversationContext序列化并保存到存储中完成本次交互。3.2 代码组织建议建议按功能模块而非技术分层来组织代码目录这样更符合Skill的“能力单元”特性。my_travel_skill/ ├── __init__.py ├── intents/ # 意图处理器目录 │ ├── __init__.py │ ├── flight.py # 包含 handle_flight_booking 等 │ ├── hotel.py │ └── weather.py ├── services/ # 服务门面目录 │ ├── __init__.py │ ├── flight_client.py │ ├── hotel_client.py │ └── payment_client.py ├── chains/ # 责任链处理器目录 │ ├── __init__.py │ ├── booking_chain.py │ └── search_chain.py ├── strategies/ # 策略目录 │ ├── __init__.py │ ├── response_strategies.py │ └── vendor_strategies.py ├── models/ # 数据模型上下文、状态对象 │ ├── __init__.py │ └── context.py └── config.py # 配置如策略映射这种结构让每个模式的作用域清晰可见便于团队协作和代码维护。4. 常见问题、调试技巧与性能考量即使采用了良好的设计模式在开发过程中仍会遇到各种问题。以下是一些常见陷阱和解决思路。4.1 上下文状态丢失或混乱问题用户在多轮对话中信息突然“失忆”或者A用户看到了B用户的信息。排查首先检查ConversationContext的session_id生成和传递逻辑是否正确。确保每次请求都能正确关联到之前的会话。检查状态持久化逻辑。是否在每次修改上下文后都成功保存序列化/反序列化过程是否有数据丢失特别是自定义对象检查是否有全局变量或类变量被误用来存储会话状态。这是导致状态混乱的常见原因。技巧为ConversationContext添加一个版本号或最后修改时间戳。在保存时进行乐观锁检查可以有效防止并发导致的状态覆盖。4.2 外部服务调用超时或失败导致Skill卡死问题调用某个第三方API没有响应整个Skill线程被阻塞后续请求无法处理。解决必须设置超时在所有外部HTTP/数据库调用中显式设置超时参数。使用异步编程确保你的ADK Skill框架和你的代码是异步的如使用asyncio。这样当一个请求在等待IO时其他请求可以被处理。实施熔断和降级在服务门面中集成熔断器如pybreaker。当失败率达到阈值直接快速返回一个友好的降级信息如“服务暂时不可用请稍后再试”而不是一直等待超时。实操命令使用curl或postman模拟慢速网络测试你的Skill的超时和降级响应是否正常工作。4.3 责任链或流程逻辑变得难以追踪问题责任链太长或者流程中条件分支太多出现Bug时很难定位是哪个环节出了问题。解决结构化日志在每个处理器的入口和出口打上带有唯一请求ID和处理器名称的日志。使用JSON格式的日志便于后续用ELK等工具分析。链路追踪在复杂的分布式Agent系统中可以考虑集成OpenTelemetry等链路追踪工具可视化请求在各个环节的流转和耗时。设计复审如果链过长考虑是否可以将一些步骤合并或者拆分成多个更小的、可复用的子链。4.4 性能瓶颈分析对于高频使用的Skill性能至关重要。** profiling**使用Python的cProfile模块或py-spy等工具找出代码中的热点函数。瓶颈往往出现在复杂的字符串处理、低效的循环、频繁的数据库查询或序列化/反序列化操作。缓存应用数据缓存对不常变的外部数据如城市列表、产品目录使用内存缓存如functools.lru_cache或 Redis。上下文缓存对于活跃会话的上下文可以缓存在内存中而不是每次请求都读写数据库但要注意缓存失效和同步策略。连接池确保你的数据库客户端、HTTP客户端使用了连接池避免频繁创建和销毁连接的开销。掌握这五种设计模式并理解它们如何组合运用能让你在ADK Skill开发中从“能实现功能”进阶到“能设计出健壮、可维护、可扩展的优秀技能”。模式是工具最终目的是为了写出更清晰的代码更从容地应对需求变化。在实际项目中不必追求完美的模式应用而是从最痛点入手逐步重构让代码结构向更好的方向演化。