qwen_agent/plans/schedule-job.md

feat: Schedule Job 定时任务系统

概述

为每个用户的每个 bot 提供定时任务能力，支持 cron 周期任务和一次性定时任务。到期后系统自动调用 Agent 执行任务，结果写入日志并通过 PostAgent Hook 推送通知。

背景

当前状态

系统已有的后台任务机制：

• Huey 任务队列 (task_queue/): SQLite 后端，用于文件处理
• AsyncIO 后台任务: asyncio.create_task() 用于非阻塞操作
• Checkpoint 清理调度器 (agent/db_pool_manager.py): asyncio loop 方式

缺失能力：

• 无法为用户设置周期性或定时触发的 Agent 任务
• 无 cron 式调度器
• 无用户级任务管理接口

设计目标

• 用户通过 AI 对话即可创建/管理定时任务
• 任务数据以 YAML 文件存储在用户专属目录，便于查看和备份
• 全局调度器以 asyncio 后台任务运行，无需额外进程
• 复用现有 create_agent_and_generate_response() 执行任务

架构设计

┌─────────────────┐     ┌──────────────────────┐     ┌────────────────────┐
│  AI Agent 对话   │     │   全局异步调度器       │     │  Agent 执行引擎     │
│  (Shell脚本工具)  │────▶│  (asyncio loop)      │────▶│  create_agent_and_  │
│  增删改查 tasks   │     │  每60s扫描tasks.yaml  │     │  generate_response  │
└─────────────────┘     └──────────────────────┘     └────────────────────┘
        │                        │                           │
        ▼                        ▼                           ▼
  tasks.yaml                tasks.yaml                  task_logs/
  (用户任务数据)            (更新执行时间)               (执行结果日志)

数据模型

tasks.yaml

存储路径: projects/robot/{bot_id}/users/{user_id}/tasks.yaml

tasks:
  - id: "task_20260330143000_abc123"    # 自动生成
    name: "每日新闻摘要"                  # 任务名称
    type: "cron"                         # cron | once
    schedule: "0 9 * * *"               # cron 表达式（type=cron 时）
    scheduled_at: null                   # ISO 8601 UTC（type=once 时）
    timezone: "Asia/Tokyo"               # 用户时区，用于 cron 解析
    message: "请帮我总结今天的科技新闻"     # 发送给 agent 的消息
    status: "active"                     # active | paused | done | expired
    created_at: "2026-03-30T05:30:00Z"
    last_executed_at: null               # 上次执行时间（UTC）
    next_run_at: "2026-03-31T00:00:00Z"  # 下次执行时间（UTC，调度器用此判断）
    execution_count: 0                   # 已执行次数

关键设计决策：

• 所有时间统一 UTC 存储，cron 表达式结合 timezone 字段在本地时间计算后转 UTC
• next_run_at 预计算，调度器只需简单比较时间戳即可判断是否到期
• 一次性任务执行后 status 自动变为 done

execution.log

存储路径: projects/robot/{bot_id}/users/{user_id}/task_logs/execution.log

- task_id: "task_20260330143000_abc123"
  task_name: "每日新闻摘要"
  executed_at: "2026-03-31T00:00:15Z"
  status: "success"       # success | error
  response: "今天的科技新闻摘要：..."
  duration_ms: 12500

保留最近 100 条日志，自动清理旧记录。

新增文件

1. skills/schedule-job/scripts/schedule_manager.py

Shell 命令行工具（argparse CLI），AI 通过 shell 调用来管理用户的定时任务。

环境变量输入（通过 plugin hook 机制传入）：

• BOT_ID: 当前 bot ID
• USER_IDENTIFIER: 当前用户标识

支持的命令：

命令	说明	示例
list	列出任务	list --format brief
add	添加任务	add --name "每日新闻" --type cron --schedule "0 9 * * *" --timezone "Asia/Tokyo" --message "..."
edit	编辑任务	edit <task_id> --schedule "0 10 * * 1-5"
delete	删除任务	delete <task_id>
toggle	暂停/恢复	toggle <task_id>
logs	查看日志	logs --task-id <id> --limit 10

核心逻辑：

• 一次性任务 --scheduled-at 接受 ISO 8601 格式（带时区偏移），内部转 UTC
• cron 任务通过 croniter + timezone 计算 next_run_at
• 自动创建 users/{user_id}/ 目录

2. skills/schedule-job/SKILL.md

Skill 描述文件，注入 AI prompt，告诉 AI：

• 如何使用 schedule_manager.py CLI
• cron 表达式语法说明
• 时区映射规则（语言 → 时区）
• 使用示例

3. skills/schedule-job/.claude-plugin/plugin.json

{
  "hooks": {
    "PrePrompt": {
      "command": "python scripts/schedule_manager.py list --format brief"
    }
  }
}

通过 PrePrompt hook，AI 每次对话时自动看到用户当前的任务列表。

4. services/schedule_executor.py

全局异步调度器，核心类 ScheduleExecutor。

运行机制：参考 db_pool_manager.py 的 _cleanup_loop() 模式

启动 → asyncio.create_task(_scan_loop)
       ↓
    每 SCAN_INTERVAL 秒
       ↓
    遍历 projects/robot/*/users/*/tasks.yaml
       ↓
    找到 status=active && next_run_at <= now 的任务
       ↓
    asyncio.create_task(_execute_task) → 受 Semaphore 并发控制
       ↓
    构建 AgentConfig → create_agent_and_generate_response(stream=False)
       ↓
    写入 execution.log → 更新 tasks.yaml (next_run_at / status)

并发与防重复：

• _executing_tasks: set 记录正在执行的任务 ID，防止同一任务被重复触发
• asyncio.Semaphore(MAX_CONCURRENT) 限制最大并发数（默认 5）

任务执行后更新：

• cron 任务：使用 croniter 计算下次 UTC 时间写入 next_run_at
• once 任务：将 status 设为 done，清空 next_run_at
• 失败时也更新 next_run_at，避免无限重试

Agent 调用构建：

bot_config = await fetch_bot_config(bot_id)
project_dir = create_project_directory(dataset_ids, bot_id, skills)
config = AgentConfig(
    bot_id=bot_id,
    user_identifier=user_id,
    session_id=f"schedule_{task_id}",  # 专用 session
    stream=False,
    tool_response=False,
    messages=[{"role": "user", "content": task["message"]}],
    ...  # 其余参数从 bot_config 获取
)
result = await create_agent_and_generate_response(config)

修改文件

pyproject.toml

新增依赖：

"croniter (>=2.0.0,<4.0.0)",
"pyyaml (>=6.0,<7.0)",

utils/settings.py

新增环境变量：

SCHEDULE_ENABLED = os.getenv("SCHEDULE_ENABLED", "true") == "true"
SCHEDULE_SCAN_INTERVAL = int(os.getenv("SCHEDULE_SCAN_INTERVAL", "60"))   # 扫描间隔（秒）
SCHEDULE_MAX_CONCURRENT = int(os.getenv("SCHEDULE_MAX_CONCURRENT", "5"))  # 最大并发数

fastapi_app.py

在 lifespan 中集成调度器生命周期：

• startup: schedule_executor.start() — 在 checkpoint 清理调度器之后启动
• shutdown: await schedule_executor.stop() — 在 Mem0 关闭之前停止

复用的现有组件

组件	路径	用途
create_agent_and_generate_response()	routes/chat.py:246	执行 agent 调用
fetch_bot_config()	utils/fastapi_utils.py	获取 bot 配置
create_project_directory()	utils/fastapi_utils.py	创建项目目录
AgentConfig	agent/agent_config.py	构建执行配置
_cleanup_loop() 模式	agent/db_pool_manager.py:240	asyncio 后台循环参考
execute_hooks('PostAgent')	agent/plugin_hook_loader.py	执行结果通知

验证方式

# 1. 启动服务
poetry run uvicorn fastapi_app:app --host 0.0.0.0 --port 8001

# 2. CLI 创建测试任务
BOT_ID=<bot_id> USER_IDENTIFIER=test_user \
  poetry run python skills/schedule-job/scripts/schedule_manager.py add \
  --name "测试任务" --type once \
  --scheduled-at "$(date -u -v+2M '+%Y-%m-%dT%H:%M:%SZ')" \
  --message "你好，这是一个定时任务测试"

# 3. 检查 tasks.yaml
cat projects/robot/<bot_id>/users/test_user/tasks.yaml

# 4. 等待调度器执行（最多 60 秒），观察应用日志

# 5. 检查执行结果
cat projects/robot/<bot_id>/users/test_user/task_logs/execution.log

# 6. 通过 API 对话测试 skill
curl -X POST http://localhost:8001/api/v2/chat/completions \
  -H 'authorization: Bearer <token>' \
  -H 'content-type: application/json' \
  -d '{"messages":[{"role":"user","content":"帮我创建一个每天早上9点的定时任务"}],"stream":false,"bot_id":"<bot_id>","user_identifier":"test_user"}'

后续扩展方向

• 任务失败重试: 可增加 max_retries 和 retry_delay 字段
• 任务执行超时: 为单个任务设置超时限制
• 通知渠道: 集成飞书/邮件等 PostAgent Hook 推送执行结果
• Web 管理界面: 提供 REST API 查询/管理定时任务
• 任务模板: 预设常用任务模板（每日新闻、天气播报等）