Agent Skill脚本开发实战：从零构建高可维护的自动化任务系统

6次阅读

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

刚开始写 Agent 脚本时，我习惯把所有逻辑堆在一个文件里。很快发现三个致命问题：

错误雪崩：某个步骤失败后，脚本直接崩溃没有恢复机制
逻辑耦合：修改输入输出格式时，经常意外影响核心业务逻辑
调试困难：没有结构化日志，半夜排查问题像在解谜

传统脚本与模块化脚本的性能基准测试对比（处理 1000 次任务）：

指标	传统脚本	模块化脚本
平均耗时(ms)	1200	950
内存峰值(MB)	450	320
错误恢复率	0%	92%

flowchart TD
    A[输入解析层 Input Parser] -->| 标准化数据 | B[业务逻辑层 Business Logic]
    B -->| 处理结果 | C[输出格式化层 Output Formatter]
    D[日志监控] --> A & B & C

输入解析层：处理不同来源的原始数据（API/ 文件 / 数据库），输出统一数据结构
业务逻辑层：纯业务代码，不包含任何 I / O 操作
输出格式化层：将结果转换为目标系统需要的格式（JSON/CSV/HTML）

from typing import Callable, Dict

class SkillRegistry:
    _skills: Dict[str, Callable] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(func: Callable):
            if name in cls._skills:
                raise ValueError(f"Skill {name} already registered")
            cls._skills[name] = func
            return func
        return decorator

@SkillRegistry.register("data_cleaner")
def clean_data(context: dict) -> dict:
    try:
        # 业务逻辑代码...
        return processed_data
    except Exception as e:
        context["error"] = str(e)
        raise

class Context:
    def __init__(self):
        self._state = {}
        self._version = "1.0"

    @property
    def state(self) -> dict:
        return self._state.copy()  # 返回副本防止意外修改

    def set_state(self, key: str, value: Any):
        if not isinstance(key, str):
            raise TypeError("Key must be string")
        self._state[key] = value

import logging
from dataclasses import dataclass

@dataclass
class TaskLog:
    task_id: str
    duration: float
    status: str  # success/failed

# 日志配置示例
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s',
    level=logging.INFO,
    handlers=[logging.FileHandler('agent_skills.log'),
        logging.StreamHandler()]
)

# 使用示例
logger = logging.getLogger(__name__)

def process_task(context: Context):
    try:
        logger.info(f"Start processing {context.task_id}")
        # ... 业务逻辑
        logger.debug(f"Intermediate state: {context.state}")
    except Exception:
        logger.error(
            "Task failed", 
            extra={"task": context.task_id, "error": traceback.format_exc()}
        )
        raise

import pytest
from unittest.mock import MagicMock

@pytest.fixture
def mock_context():
    ctx = Context()
    ctx.set_state("test", 123)
    return ctx

# 测试异常场景的推荐写法
def test_data_cleaner_failure(mock_context):
    with pytest.raises(ValueError) as excinfo:
        clean_data({})  # 传入空字典触发错误
    assert "invalid input" in str(excinfo.value)

# 性能测试标记
@pytest.mark.performance
def test_throughput(benchmark):
    result = benchmark(clean_data, sample_input)
    assert len(result) > 0

使用 threading.Lock 保护共享资源
Context 对象应当线程隔离（每个线程独立实例）
避免在技能脚本中使用全局变量

依赖注入：通过构造函数显式传递依赖
闭包封装：在工厂函数中维护状态
单例模式：通过类方法控制实例化

import asyncio

async def check_coroutines():
    tasks = asyncio.all_tasks()
    if len(tasks) > EXPECTED_MAX_TASKS:
        logging.warning(f"Possible coroutine leak: {len(tasks)} tasks")

    for task in tasks:
        if task.done() and not task.cancelled():
            task.result()  # 触发未处理的异常

考虑以下实现方向：