Agent Skill 写报告：从零构建自动化报告生成系统

8次阅读

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

最近接手了一个需要每天生成业务报告的项目，手动操作 Excel 和 PPT 的日子让我痛不欲生。更可怕的是，当数据源出现变动时，所有图表都得推倒重来。这种重复劳动不仅效率低下，还容易出错——上周就因漏更新某个字段，导致整份报告数据失真。这让我下定决心用自动化方案彻底解决这个问题。

传统脚本虽然能完成基础自动化，但存在明显缺陷：

脚本通常是线性执行，一个步骤失败整个流程就崩溃
状态管理混乱，很难从中间步骤恢复
功能修改需要重写大量代码

Agent Skill 架构则通过模块化设计解决了这些问题：

故障隔离 ：每个技能独立运行，单个模块失败不影响整体
状态持久化 ：自动保存执行上下文，支持断点续跑
灵活编排 ：通过状态机动态调整执行流程

每个技能需要明确定义输入输出规范，例如我们的「数据清洗」技能：

class DataCleaningSkill:
    input_schema = {
        'raw_data': pd.DataFrame,
        'date_range': (str, str)  # (start_date, end_date)
    }

    output_schema = {
        'cleaned_data': pd.DataFrame,
        'missing_stats': dict
    }

    def execute(self, context):
        # 实现细节在下文展开
        pass

以数据采集模块为例，展示带异常处理的完整实现：

import pandas as pd
from datetime import datetime

class DataFetcher:
    def __init__(self, api_endpoint):
        self.endpoint = api_endpoint

    def fetch_data(self, start_date, end_date):
        try:
            # 模拟 API 请求
            params = {'start': start_date.strftime('%Y-%m-%d'),
                'end': end_date.strftime('%Y-%m-%d')
            }

            # 这里替换为实际请求代码
            raw_data = pd.read_csv('sample_data.csv')

            # 时区处理关键点！if 'timestamp' in raw_data.columns:
                raw_data['timestamp'] = pd.to_datetime(raw_data['timestamp'], 
                    utc=True
                ).dt.tz_convert('Asia/Shanghai')

            return {'status': 'success', 'data': raw_data}

        except Exception as e:
            return {
                'status': 'error',
                'error': str(e),
                'retryable': True  # 标记是否可重试
            }

使用 transitions 库实现流程编排：

from transitions import Machine

class ReportAgent:
    states = ['idle', 'fetching', 'cleaning', 'analyzing', 'generating', 'done']

    def __init__(self):
        self.machine = Machine(
            model=self,
            states=self.states,
            initial='idle'
        )

        # 定义状态转移规则
        self.machine.add_transition(
            trigger='start',
            source='idle',
            dest='fetching',
            before='_prepare_params'
        )

        # 添加更多转移规则...

    def _prepare_params(self):
        # 准备执行参数
        self.report_date = datetime.now().strftime('%Y-%m-%d')

使用 asyncio 提升 I / O 密集型任务效率：

import asyncio

async def async_fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

async def batch_fetch(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [async_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

所有时间戳存储时强制带时区信息
业务逻辑中使用 UTC 时间计算
最终展示时转换为目标时区

import fcntl
import os

class FileLock:
    def __init__(self, filepath):
        self.filepath = filepath
        self.fd = None

    def __enter__(self):
        self.fd = open(self.filepath, 'a+')
        fcntl.flock(self.fd, fcntl.LOCK_EX)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        fcntl.flock(self.fd, fcntl.LOCK_UN)
        self.fd.close()

优化前后关键指标对比（测试数据集：10 万条记录）：