那时难决
0%

定时任务的脚本化改造与AI Agent的CodeMode

Posted on In

我最近发现我在Workbuddy配置的“晨间提醒”定时任务执行时长有点久，经脚本化改造后得到了显著的改进。

改造前状态

“晨间提醒”定时任务的大致工作内容如下：

  • 执行everyday timeline sync命令同步操作后，通过everyday cli工具获取24小时内的邮件、当天的日历事件和待办任务。
  • 通过腾讯新闻 Skill获取热点新闻和当地天气信息。
  • 通过github cli工具获取前一天的Github动态。
  • 整理上述信息后，生成一份晨间提醒报告，发送到我的个人微信。
  • 总体上该这是一个结构相对稳定、信息源相互独立的任务。

AI Agent执行“晨间提醒”定时任务的效率瓶颈主要集中以下两个方面：

  1. 获取工具用法：AI Agent在每次执行任务时，往往都会重新通过everyday --help和gh --help命令获取命令行工具的用法说明。
    类似地，AI Agent也会重新加载所需的Agent Skill并理解其意图和使用方式，做了很多重复的工作。

  2. 串行执行命令：虽然各个信息源的获取是相互独立的，但AI Agent还是会按照顺序一个接一个地执行这些命令，导致整体执行时间较长。某些Agent支持运行并行子Agent执行多个子任务，但由语言模型进行任务的分派和结果汇总的效率也不高，并且会消耗大量的Token。

脚本化改造

考虑到我配置的“晨间提醒”定时任务的工作内容是相对稳定的，并且其涉及到的信息本质都可以执行命令行工具来获取，我让Workbuddy编写了一个Python脚本用于并行执行这些命令行工具获取原始信息，然后将结果汇总后再进行处理和生成报告，从而有效提升了执行效率并节省了大量Token。

脚本的设计方案与大致实现如下：

  1. 双阶段执行策略

    • Phase 0 (串行同步)：优先执行 everyday timeline sync 和 mail sync，确保本地缓存是最新的。这是唯一不能并行的部分，且一旦失败会阻断后续流程。
    • Phase 1 (并行抓取)：利用 ThreadPoolExecutor 并发执行邮件查询、日历、待办、RSS、天气、新闻和 GitHub 动态等 I/O 密集型任务，极大缩短总耗时。

  2. 健壮的容错与兜底机制

    • 核心/非核心分级：mail、cal、todo 标记为 core=True，若失败则脚本报错退出（Exit 1）；其他资讯类任务失败仅记录日志，不影响报告生成。
    • 智能回退 (Fallback)：针对 Windows 环境下 PowerShell 调用可能出现的特定错误，脚本内置了自动切换至 .exe 直连的逻辑（should_fallback），确保环境兼容性。

  3. 数据与展示分离

    • 每个数据源都严格遵循 cmd（定义命令）+ fmt（解析 JSON/文本为 Markdown）的模式。
    • 将复杂的过滤逻辑（如“过去24h邮件”、“未来7天日程”）下沉到脚本层处理，而非依赖 LLM，既节省了 Token，又避免了上下文污染。

  4. 标准化输出

    • 最终生成一份结构化的 Markdown 文件（简报素材_YYYY-MM-DD.md），作为“纯净的原材料”供给下游的 AI Agent 或其他分发渠道（如微信推送），实现了**“脚本采编，AI 润色”**的最佳分工。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#!/usr/bin/env -S uv run --quiet --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["typer", "tzdata"]
# ///

from __future__ import annotations

import json
import shutil
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Callable
from zoneinfo import ZoneInfo

import typer

def _resolve_bin(name: str, fallback: str) -> str:
    return shutil.which(name) or fallback

# ── 子进程执行 ────────────────────────────────────────────────────────
@dataclass
class CmdResult:
    cmd: list[str]
    returncode: int
    stdout: str
    stderr: str
    @property
    def ok(self) -> bool: ...
    @property
    def merged(self) -> str: ...

def run_cmd(cmd: list[str], timeout: int = 60, cwd: Path | None = None) -> CmdResult:
    """执行命令；UTF-8 解码（规避 PS5.1 GBK 坑）。"""

# ── 任务定义 ──────────────────────────────────────────────────────────
@dataclass
class Ctx:
    today: date
    gh_date: str | None
    now_local: datetime = field(default_factory=lambda: datetime.now(TZ))
    errors: dict[str, str] = field(default_factory=dict)

@dataclass
class TaskDef:
    key: str
    desc: str
    section: str
    core: bool
    cmd: Callable[[Ctx], list[str]]
    fmt: Callable[[Ctx, CmdResult], str]
    timeout: int = 60
    fallback_cmd: Callable[[Ctx], list[str]] | None = None
    should_fallback: Callable[[CmdResult], bool] | None = None

    def run(self, ctx: Ctx) -> str:
        """执行 → 可选回退 → 失败写占位符，返回 Markdown 正文。"""

# ── 各数据源：cmd/fmt 成对实现 ────────────────────────────────────────
def _mail_cmd(ctx: Ctx) -> list[str]: ...          # everyday mail list --json --limit 30
def _mail_fmt(ctx: Ctx, res: CmdResult) -> str: ...  # 过去 24h 过滤，倒序取 20

def _cal_cmd(ctx: Ctx) -> list[str]: ...            # everyday cal list --all --json
def _cal_fmt(ctx: Ctx, res: CmdResult) -> str: ...  # 今日+7 天，过期剔除

def _todo_cmd(ctx: Ctx) -> list[str]: ...           # everyday todo list --json
def _todo_fmt(ctx: Ctx, res: CmdResult) -> str: ... # 逾期/今日到期标记

def _rss_cmd(ctx: Ctx) -> list[str]: ...            # everyday rss fetch aihot --limit 10 --json
def _rss_fmt(ctx: Ctx, res: CmdResult) -> str: ...  # 容错提取标题 top10

def _weather_cmd(ctx: Ctx) -> list[str]: ...        # pwsh7 run-cli.ps1 weather --adcode 330105
def _morning_cmd(ctx: Ctx) -> list[str]: ...        # pwsh7 run-cli.ps1 morning
def _morning_fallback_cmd(ctx: Ctx) -> list[str]: ...  # 4009 → 直跑 .exe
def _morning_should_fallback(res: CmdResult) -> bool: ...  # "4009" in stdout+stderr
def _verbatim_fmt(ctx: Ctx, res: CmdResult) -> str: ...  # 原样输出

def _github_cmd(ctx: Ctx) -> list[str]: ...         # uv run github_daily_fetch.py [--date]
def _github_fmt(ctx: Ctx, res: CmdResult) -> str: ...  # 脚本 stdout 原样

# ── 任务注册表 ────────────────────────────────────────────────────────
TASKS: dict[str, TaskDef] = {
    t.key: t for t in [
        TaskDef("mail", "过去 24h 邮件", "📬 新邮件（过去 24h）", True, _mail_cmd, _mail_fmt),
        TaskDef("cal", "今日+7 天日程", "📅 日程（今日 + 未来 7 天）", True, _cal_cmd, _cal_fmt, timeout=90),
        TaskDef("todo", "待办标记", "✅ 待办", True, _todo_cmd, _todo_fmt),
        TaskDef("rss", "AI Hot RSS", "🤖 AI Hot 精选（最新 10 条）", False, _rss_cmd, _rss_fmt),
        TaskDef("weather", "杭州拱墅天气", "🌤 天气（CLI 原样）", False, _weather_cmd, _verbatim_fmt),
        TaskDef("morning", "新闻早报", "📰 新闻早报（CLI 原样）", False,
                _morning_cmd, _verbatim_fmt, timeout=90,
                fallback_cmd=_morning_fallback_cmd, should_fallback=_morning_should_fallback),
        TaskDef("github", "GitHub 昨日动态", "🐙 GitHub 昨日动态（脚本直出）",
                False, _github_cmd, _verbatim_fmt, timeout=180),
    ]
}
ALL_KEYS = list(TASKS)
CORE_KEYS = {k for k, t in TASKS.items() if t.core}

# ── 编排 ──────────────────────────────────────────────────────────────
SYNC_STEPS: list[tuple[str, list[str]]] = [
    ("timeline sync", [EVERYDAY, "timeline", "sync"]),
    ("mail sync", [EVERYDAY, "mail", "list", "--sync", "--json", "--limit", "1"]),
]

def run_sync_phase(ctx: Ctx) -> bool:
    """Phase 0：串行全量同步（写缓存，必须最先）。失败 → 核心错误。"""

def run_query_phase(ctx: Ctx, keys: list[str]) -> dict[str, str]:
    """Phase 1：ThreadPoolExecutor 并行抓取，返回 {key: markdown 正文}。"""

def build_markdown(ctx: Ctx, keys: list[str], results: dict[str, str], sync_ok: bool) -> str:
    """汇总：标题 + 同步失败警示 + 异常段 + 各板块。"""

def main(
    tasks: str = typer.Option("all", "--tasks"),
    output: str = typer.Option("", "--output"),
    no_sync: bool = typer.Option(False, "--no-sync"),
    brief_date: str = typer.Option("", "--date"),
) -> None:
    # 1. 参数校验：--tasks 子集 / 未知任务退出码 2
    # 2. Phase 0：串行同步（--no-sync 跳过）
    # 3. Phase 1：并行抓取
    # 4. build_markdown → 写入 output/简报素材_YYYY-MM-DD.md
    # 5. 核心失败（mail/cal/todo 或同步）→ 退出码 1，否则 0

if __name__ == "__main__":
    typer.run(main)

AI Agent的CodeMode模式

我对“晨间提醒”任务的脚本化改造，一定程度上和CodeMode的理念相近。

Code Mode 是 AI Agent 的一种工具调用模式：不让模型逐个调用大量 Tool，而是提供一个代码执行环境，让模型生成代码，通过代码编排多个工具调用、数据处理和流程控制。优点是减少上下文、降低 token 消耗、支持循环/并行/条件逻辑，适合复杂任务。

  • Cloudflare Agents：Code Mode 用一个小型 API/SDK 描述替代数千个工具定义，Agent 生成 JavaScript 调用 Cloudflare API，解决 MCP 大规模工具暴露问题。(Cloudflare Docs)
  • Pydantic AI：提供原生 CodeMode capability，通过沙箱 Python（Monty runtime）让 Agent 编写代码调用 tools，并支持 asyncio.gather 并行执行。(Pydantic)
  • Claude Code：Claude Code Workflows是一种“用 TS 代码控制流程、用 LLM 执行节点”的确定性编排模式，通过代码调度并发与管道，在隔离上下文的同时避免长流程中的 Context 爆满问题。

结语

这次“晨间提醒”的脚本化改造，本质上是从 “让模型现场规划” 回归 “让代码负责确定性”。
它印证了在 AI Agent 的应用中，并非所有复杂任务都需要交给模型去实时编排。
当任务的“流程”是确定的，就该把它固化成代码；只有当“目标”是确定的而“路径”不确定时，才需要交给 Agent 去探索和规划。
这种“确定性下沉”的思路，将是未来优化 Agent 效率和成本的关键方向。

扫码加入技术交流群🖱️
QR code