智能体skill的五种设计模式：从新手入门到生产实践

2次阅读

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

在智能体开发中，skill 层作为核心功能模块的集合，承担着任务解析、逻辑执行和结果反馈的关键职责。然而，新手开发者常面临以下痛点：

复用性差 ：功能逻辑与业务代码高度耦合，难以跨场景复用
维护成本高 ：状态管理混乱导致调试困难，变更影响范围不可控
扩展性弱 ：新增功能时需要修改核心流程，违反开闭原则
性能瓶颈 ：同步阻塞调用导致响应延迟，高并发场景下吞吐量骤降

适用场景 ：需要顺序执行多个处理阶段的技能链

from typing import Callable, Any

class Pipeline:
    def __init__(self):
        self._steps = []

    def add_step(self, func: Callable[[Any], Any]) -> None:
        self._steps.append(func)

    def execute(self, input_data: Any) -> Any:
        result = input_data
        for step in self._steps:
            result = step(result)
        return result

# 使用示例
def normalize_text(text: str) -> str:
    return text.lower().strip()

def remove_stopwords(text: str) -> str:
    stopwords = {"a", "the", "is"}
    return " ".join(w for w in text.split() if w not in stopwords)

pipeline = Pipeline()
pipeline.add_step(normalize_text)
pipeline.add_step(remove_stopwords)
print(pipeline.execute("The Quick Brown Fox"))  # 输出: quick brown fox

性能考量 ：
– 延迟：O(n) 时间复杂度，n 为步骤数
– 线程安全：若各步骤无共享状态，可并行化处理

适用场景 ：需要动态添加功能的场景

from functools import wraps
from typing import Callable

def logging_decorator(func: Callable) -> Callable:
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"Executing {func.__name__}")
        result = func(*args, **kwargs)
        print(f"Finished {func.__name__}")
        return result
    return wrapper

@logging_decorator
def process_order(order_id: str) -> bool:
    # 订单处理逻辑
    return True

优劣对比 ：
– 优点：非侵入式扩展，符合单一职责原则
– 缺点：多层装饰可能导致调用栈过深

适用场景 ：行为随状态改变而变化的场景

from abc import ABC, abstractmethod

class SkillState(ABC):
    @abstractmethod
    def handle(self) -> str:
        pass

class IdleState(SkillState):
    def handle(self) -> str:
        return "Waiting for input"

class ProcessingState(SkillState):
    def handle(self) -> str:
        return "Processing request"

class SkillContext:
    def __init__(self):
        self._state = IdleState()

    def change_state(self, state: SkillState):
        self._state = state

    def execute(self) -> str:
        return self._state.handle()

适用场景 ：事件驱动型技能协作

from typing import List, Callable

class EventPublisher:
    def __init__(self):
        self._subscribers: List[Callable[[str], None]] = []

    def subscribe(self, callback: Callable[[str], None]) -> None:
        self._subscribers.append(callback)

    def notify(self, event: str) -> None:
        for callback in self._subscribers:
            callback(event)

# 使用示例
def log_event(event: str) -> None:
    print(f"[LOG] {event}")

publisher = EventPublisher()
publisher.subscribe(log_event)
publisher.notify("Task completed")

适用场景 ：算法族可互换的场景

from abc import ABC, abstractmethod
from typing import List

class SortingStrategy(ABC):
    @abstractmethod
    def sort(self, data: List[int]) -> List[int]:
        pass

class QuickSort(SortingStrategy):
    def sort(self, data: List[int]) -> List[int]:
        return sorted(data)  # 简化实现

class BubbleSort(SortingStrategy):
    def sort(self, data: List[int]) -> List[int]:
        # 实现冒泡排序
        return data.copy()

class Sorter:
    def __init__(self, strategy: SortingStrategy):
        self._strategy = strategy

    def execute_sort(self, data: List[int]) -> List[int]:
        return self._strategy.sort(data)