5种Agent技能设计模式解析：ADK开发者必知的核心实践

6次阅读

共计 2346 个字符，预计需要花费 6 分钟才能阅读完成。

在开发 Agent 技能时，我们常会遇到以下典型问题：

代码臃肿 ：单个技能类随着功能增加变得庞大难维护
跨技能耦合 ：技能间直接调用导致修改牵一发而动全身
状态混乱 ：对话流程中的状态管理缺乏清晰边界
扩展困难 ：新增功能需要修改已有代码，违反开闭原则

这些问题会显著降低开发效率，特别是在需要快速迭代和组合多种技能的对话系统中。

模式 (Pattern)	适用场景	性能开销	典型应用案例
观察者模式 (Observer)	事件驱动场景，如用户意图变化通知	低	用户输入触发多技能响应
策略模式 (Strategy)	算法可替换场景，如不同 NLU 解析器	低	对话策略动态切换
状态模式 (State)	状态转移明确场景，如多轮对话流程	中	订单查询状态管理
装饰器模式 (Decorator)	功能叠加场景，如技能能力扩展	中高	日志记录 + 缓存装饰
责任链模式 (Chain of Responsibility)	处理流程分派场景，如意图路由	中	用户请求管道处理

from abc import ABC, abstractmethod
from typing import List

class SkillObserver(ABC):
    @abstractmethod
    def update(self, event_data: dict):
        pass

class UserIntentNotifier:
    def __init__(self):
        self._observers: List[SkillObserver] = []

    def attach(self, observer: SkillObserver):
        self._observers.append(observer)

    def notify(self, event_data: dict):
        for observer in self._observers:
            try:
                observer.update(event_data)
            except Exception as e:
                print(f"Observer notification failed: {str(e)}")

class WeatherSkill(SkillObserver):
    def update(self, event_data: dict):
        if event_data.get('intent') == 'weather_query':
            self.handle_weather_request(event_data)

    def handle_weather_request(self, data: dict):
        print("Handling weather request...")

应用案例 ：当用户输入 ” 今天天气如何 ” 时，通知所有注册的技能，只有 WeatherSkill 会响应处理。

from typing import Protocol

class NLUStrategy(Protocol):
    def parse(self, text: str) -> dict:
        pass

class RegexNLU:
    def parse(self, text: str) -> dict:
        # 实现正则匹配逻辑
        return {'intent': 'regex_matched'}

class MLNLU:
    def parse(self, text: str) -> dict:
        # 实现模型预测逻辑
        return {'intent': 'model_predicted'}

class DialogManager:
    def __init__(self, strategy: NLUStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: NLUStrategy):
        self._strategy = strategy

    def process_input(self, text: str):
        try:
            return self._strategy.parse(text)
        except Exception as e:
            print(f"NLU processing failed: {str(e)}")
            return {'intent': 'fallback'}

应用案例 ：根据运行时条件切换使用规则引擎或机器学习模型进行自然语言理解。

在状态模式中，Context 对象通常维护当前状态引用。在多线程环境中需要特别注意：

from threading import Lock

class ConversationContext:
    def __init__(self):
        self._state = None
        self._lock = Lock()

    def change_state(self, new_state):
        with self._lock:
            self._state = new_state

    def handle_input(self, input_data):
        with self._lock:
            return self._state.handle(input_data)

装饰器嵌套过深会导致调用栈膨胀，建议：