AI Agent任务持久化、后台执行与定时唤醒实战指南

1. 项目概述：为什么你的Agent一觉不醒？

最近在折腾AI Agent项目，发现一个挺普遍的问题：很多开发者辛辛苦苦搭好了一个Agent，让它去处理一个长任务，比如爬取数据、生成周报或者监控系统状态。结果呢？程序一关，Agent的记忆就清零了；或者想让它在凌晨自动运行，却不知道如何设置；更头疼的是，Agent在后台跑着跑着，界面就卡死了，用户还以为程序崩溃了。这背后的核心，就是Agent的“持续工作能力”问题。

一个真正能投入生产的Agent，绝不能是“一次性”的。它需要像一位不知疲倦的虚拟员工，能够记住未完成的任务（任务持久化），在用户看不见的地方默默干活（后台执行），并且能在指定的时间点自动醒来工作（定时唤醒）。这三点，构成了Agent自动化与可靠性的基石。无论是做一个自动回复邮件的助手，还是一个定时巡检服务器状态的监控Agent，都离不开这套机制。

网上相关的讨论很多，从“Cron表达式怎么写”到“Winform后台刷新控件卡死”，再到各种Agent框架（如Hermes Agent）的对比，都指向了实际开发中的痛点。本文将从一个全栈开发者的视角，抛开理论，直接切入实战，手把手拆解如何为你的Agent赋予“持续工作”的灵魂。我们会从最底层的原理讲起，一直讲到不同技术栈下的实现方案与避坑指南。

2. 任务持久化：让Agent拥有“记忆”

任务持久化，简单说就是让Agent能够记住它的工作状态。想象一下，你让Agent整理一份100页的文档摘要，它刚处理到第50页，你的电脑重启了。如果没有持久化，Agent重启后要么从头开始，要么直接忘记了这个任务。这显然是不可接受的。

2.1 持久化的核心：状态与上下文

Agent的任务状态通常包括：

1. 任务元数据：任务ID、创建时间、任务类型（如“数据清洗”、“报告生成”）、优先级、状态（待处理、进行中、已完成、失败）。
2. 执行上下文：当前处理到了哪个步骤、已经获取或生成了哪些中间数据、遇到了哪些异常或需要人工干预的点。
3. Agent自身状态：在复杂Agent中，可能还包括其短期记忆、工具调用历史、对话轮次等。

实现持久化，本质上是将这些状态序列化后，存储到一个可靠的外部存储中，并在Agent恢复时能够反序列化加载。

2.2 存储方案选型与实战

选择哪种存储，取决于你的应用场景、数据量和复杂度。

方案一：关系型数据库（如MySQL, PostgreSQL）这是最通用、结构最清晰的方案。你可以设计几张表：

• tasks表：存储任务元数据。
• task_steps表：存储任务步骤的详细日志和状态。
• agent_context表：以JSON或序列化二进制格式存储Agent的完整上下文。

-- 示例表结构 CREATE TABLE tasks ( id VARCHAR(64) PRIMARY KEY, type VARCHAR(50) NOT NULL, status ENUM('pending', 'running', 'paused', 'completed', 'failed') DEFAULT 'pending', progress INT DEFAULT 0, -- 进度百分比 input_data TEXT, -- 任务输入参数（JSON格式） output_data TEXT, -- 任务输出结果（JSON格式） created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, checkpoint TEXT -- 存储序列化的上下文快照 );

为什么选它？结构化查询强大，便于做任务管理后台、统计报表。事务支持能保证状态更新的原子性。适合任务类型固定、需要复杂查询和管理的场景。

方案二：文档数据库（如MongoDB）由于Agent状态通常是半结构化的JSON数据，用文档数据库存储非常自然。

// 一个MongoDB文档示例 { “_id”: “task_001”, “type”: “document_summarization”, “status”: “running”, “context”: { “document_id”: “doc_123”, “processed_pages”: [1, 2, 3, ..., 50], “current_summary”: “这里是前50页的摘要...”, “next_action”: “fetch_page_51” }, “created_at”: ISODate(“2023-10-27T08:00:00Z”), “updated_at”: ISODate(“2023-10-27T08:30:00Z”) }

为什么选它？模式灵活，扩展性强，读写速度快。特别适合Agent上下文复杂、频繁变更的场景。与Node.js等生态结合紧密。

方案三：键值存储/缓存（如Redis）对于需要极高读写速度、状态相对简单的Agent任务，Redis是绝佳选择。你可以用Hash结构存储任务详情，用Sorted Set来管理任务队列和优先级。

# 存储任务上下文 HSET task:task_001 status “running” progress “50” context “{JSON字符串}” # 将任务放入延迟队列，实现简单的定时唤醒（后面会详述） ZADD delayed_tasks <执行时间戳> “task_001”

为什么选它？性能极致，支持丰富的数据结构。适合做高速任务队列、会话缓存或分布式锁。但需要注意Redis的持久化策略（RDB/AOF），防止内存数据丢失。

方案四：本地文件系统对于单机、轻量级的Agent，或者作为临时检查点，写入本地文件（如JSON、Pickle）是最快的方式。

import json import pickle def save_checkpoint(task_id, agent_state): # 使用JSON（可读性好） with open(f“checkpoints/{task_id}.json”, ‘w’) as f: json.dump(agent_state, f) # 或使用Pickle（可保存Python对象，但注意安全） with open(f“checkpoints/{task_id}.pkl”, ‘wb’) as f: pickle.dump(agent_state, f)

注意：文件系统方案在分布式部署或程序崩溃时可靠性较低，通常只用于开发调试或作为数据库之外的补充备份。

实操心得：

• 混合存储策略：在实际项目中，我经常采用混合策略。用MySQL管理任务元数据和生命周期，用Redis作为高速缓存存储活跃任务的上下文，用MongoDB归档完整的任务执行日志。这样各取所长。
• 序列化陷阱：如果用Pickle或Java序列化，要警惕类定义变更导致的兼容性问题。JSON虽然安全，但无法直接存储复杂的Python对象（如函数、类实例）。一个折中方案是定义好状态对象的to_dict()和from_dict()方法。
• 定时快照 vs. 事件驱动保存：不要每一步操作都保存状态，这会导致IO瓶颈。可以采用“定时快照”（例如每处理10个单元保存一次）加“关键事件保存”（如步骤完成、调用外部API前后）的策略。

3. 后台执行：让Agent在幕后稳定运行

后台执行解决了“界面不卡死”和“程序退出后任务继续”两大问题。这在开发桌面应用（Winform）、Web后台任务或常驻服务时至关重要。

3.1 理解执行模型：同步、异步与多线程/进程

• 同步阻塞：你的代码一行行执行，遇到一个耗时操作（如下载文件、调用大模型API），整个程序就卡住等待，界面自然“未响应”。
• 异步非阻塞：程序发起一个耗时操作后，不会傻等，而是继续执行后面的代码。等那个耗时操作完成了，再回来处理结果。这是现代后台任务的首选模型。
• 多线程/多进程：真正意义上同时执行多个任务。线程轻量，共享内存；进程重量，内存独立，更安全。

对于Agent来说，其核心“大脑”（LLM调用、逻辑推理）通常是IO密集型（等待网络响应），而非CPU密集型，因此异步编程（Asynchronous Programming）是最高效的范式。

3.2 各平台后台执行实战

场景一：Web应用（如Python FastAPI/Django, Node.js）Web服务器本身是无状态的，Agent任务必须作为后台作业运行。

• Celery + Redis/RabbitMQ（Python经典组合）：

  # tasks.py from celery import Celery app = Celery(‘agent_tasks’, broker=‘redis://localhost:6379/0’) @app.task(bind=True) def long_running_agent_task(self, task_input): # 这里是你的Agent核心逻辑 for i in range(100): # 模拟工作 self.update_state(state=‘PROGRESS’, meta={‘current’: i, ‘total’: 100}) # 处理逻辑... return {‘result’: ‘success’} # 在API接口中触发后台任务 from .tasks import long_running_agent_task task = long_running_agent_task.delay(user_input) return {“task_id”: task.id}

  Celery Worker会在后台独立进程运行，与Web服务解耦。通过task.id可以查询状态或结果。

• Node.js使用Bull或Agenda：

  const Queue = require(‘bull’); const agentQueue = new Queue(‘agent’); agentQueue.process(‘summarize’, async (job) => { // 你的Agent逻辑 for (let i = 0; i < 100; i++) { await job.progress(i); // ...处理逻辑 } return { summary: ‘完成’ }; }); // 在路由中入队 app.post(‘/task’, async (req, res) => { const job = await agentQueue.add(‘summarize’, { url: req.body.url }); res.json({ jobId: job.id }); });

场景二：桌面应用（如Winform、WPF）这是“刷新控件导致卡死”问题的重灾区。关键在于将耗时的Agent逻辑与UI渲染线程分离。

• .NET的 async/await 与 BackgroundWorker：

  private async void btnStartAgent_Click(object sender, EventArgs e) { btnStartAgent.Enabled = false; lblStatus.Text = “Agent运行中...”; // 错误做法：直接在UI线程调用耗时方法，会导致界面冻结 // var result = RunLongTimeAgent(); // 这会卡死UI // 正确做法：使用Task.Run在后台线程池执行 var result = await Task.Run(() => RunLongTimeAgent()); // 此后的代码会在任务完成后，自动回到UI线程执行 lblStatus.Text = “任务完成！”; txtResult.Text = result; btnStartAgent.Enabled = true; } private string RunLongTimeAgent() { // 模拟耗时操作 Thread.Sleep(5000); return “Agent处理结果”; }

  核心要点：所有涉及更新UI控件（如lblStatus.Text = ...）的操作，必须在UI线程上执行。async/await配合Task.Run可以轻松将耗时操作丢到后台，完成后自动返回UI线程更新界面，流畅无比。

• 使用BackgroundWorker组件（较旧但直观）：

  private void StartAgentWithBackgroundWorker() { BackgroundWorker worker = new BackgroundWorker(); worker.WorkerReportsProgress = true; worker.DoWork += (s, e) => { // 在后台线程执行 for (int i = 0; i <= 100; i++) { Thread.Sleep(50); worker.ReportProgress(i); // 报告进度 } e.Result = “处理完成”; }; worker.ProgressChanged += (s, e) => { // 此事件在UI线程被触发，可以安全更新控件 progressBar1.Value = e.ProgressPercentage; }; worker.RunWorkerCompleted += (s, e) => { // 任务完成，在UI线程更新 MessageBox.Show(e.Result.ToString()); }; worker.RunWorkerAsync(); }

场景三：常驻后台服务（Systemd, Windows Service）对于需要7x24小时运行的服务器端Agent，需要将其包装成系统服务。

• Linux (Systemd):

  # /etc/systemd/system/my-agent.service [Unit] Description=My AI Agent Service After=network.target [Service] Type=simple User=agentuser WorkingDirectory=/opt/my-agent ExecStart=/usr/bin/python3 /opt/my-agent/main.py Restart=on-failure RestartSec=10 [Install] WantedBy=multi-user.target

  使用sudo systemctl start my-agent启动，journalctl -u my-agent -f查看日志。

• Windows (Windows Service): 可以使用NSSM(Non-Sucking Service Manager)这个神器，将任何exe或脚本轻松安装为服务，无需编写C#代码。

  nssm install MyAgentService “C:\Python39\python.exe” “C:\MyAgent\main.py” nssm start MyAgentService

避坑指南：

1. 线程安全：在后台线程中访问共享资源（如全局配置、数据库连接池）时，务必使用锁（lockin C#,threading.Lockin Python）或其他同步机制，防止数据竞争。
2. 异常处理：后台任务的异常不会直接崩溃主程序，但必须被捕获并妥善记录到日志中，否则任务会无声无息地失败。
3. 资源泄漏：确保任务完成后，释放所有打开的文件句柄、数据库连接、网络连接等。对于长时间运行的服务，要监控内存使用，防止内存泄漏。
4. Winform卡死的根本原因：除了直接在UI线程执行耗时操作外，在后台线程中直接调用UI控件（如textBox1.Text = “xxx”）也会引发跨线程访问异常或死锁。务必通过Control.Invoke或Dispatcher.Invoke（WPF）来安全更新UI。

4. 定时唤醒：让Agent学会“闹钟”

定时唤醒是自动化Agent的标志性能力。无论是每天凌晨1点拉取数据，还是每25分钟检查一次邮箱，都需要可靠的调度机制。

4.1 Cron表达式：定时任务的通用语言

Cron表达式是一个字符串，包含5个或6个（有时包含秒）时间字段，用空格分隔。它几乎是一切定时任务调度的基础。

标准格式（5位）：分钟 小时 日 月 星期扩展格式（6位）：秒 分钟 小时 日 月 星期

特殊字符详解：

• *：任意值。在“分钟”字段表示每分钟。
• ,：指定多个值。10,20,30在“分钟”字段表示第10、20、30分钟。
• -：指定范围。9-17在“小时”字段表示上午9点到下午5点。
• /：指定增量。*/15在“分钟”字段表示每15分钟（0,15,30,45）。0/5也表示从第0分钟开始每5分钟。
• ?：仅在“日”和“星期”字段使用，表示“不指定值”。因为这两个字段互斥，指定了日期就不能再指定星期几。
• L：最后一天。在“日”字段表示月末最后一天；在“星期”字段6L表示最后一个星期五。
• W：最近的工作日。15W表示当月15日最近的工作日（如果15日是周六，则触发14日周五；如果是周日，则触发16日周一）。
• #：第几个星期几。6#3表示每月的第三个星期五。

常用示例：

• 0 0 1 * * ?：每天凌晨1点整执行一次。这是摘要描述中提到的表达式。
• 0 */25 * * * ?：每25分钟执行一次（在分钟数为0,25,50时触发）。对应热词“cron 每25分钟”。
• 0 0 0 * * ?：每天0点执行一次。
• 0 30 9 ? * MON-FRI：每周一到周五上午9:30执行。
• 0 0 12 1 * ?：每月1号中午12点执行。

注意：不同系统、不同库对Cron表达式的支持略有差异。例如，Linux系统的crontab通常只支持5位格式（不含秒），而Quartz、Spring等框架支持6位。务必查阅你所用工具的文档。

4.2 实现定时唤醒的三种模式

模式一：操作系统级Cron（最经典）在Linux服务器上，使用crontab -e编辑定时任务。

# 每天凌晨1点运行你的Agent脚本 0 1 * * * /usr/bin/python3 /path/to/your/agent.py >> /var/log/agent.log 2>&1 # 每25分钟运行一次 */25 * * * * /usr/bin/python3 /path/to/your/agent.py >> /var/log/agent.log 2>&1

优点：简单、可靠、与语言无关。缺点：任务调度分散，不好集中管理和监控；不适合需要复杂上下文传递的Agent。

模式二：使用调度库（应用内集成）这是最灵活的方式，调度逻辑和业务逻辑在同一进程内。

• Python Schedule库：

  import schedule import time def job(): print(“Agent被定时唤醒了！”) # 在这里调用你的Agent核心函数 # 每天01:00执行 schedule.every().day.at(“01:00”).do(job) # 每25分钟执行 schedule.every(25).minutes.do(job) while True: schedule.run_pending() time.sleep(1) # 避免CPU空转

  这个库语法直观，适合轻量级应用。但需要注意，它是阻塞式的，time.sleep(1)会占用线程。

• Python APScheduler： 功能更强大，支持持久化存储任务、分布式调度、Cron表达式等。

  from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore jobstores = { ‘default’: SQLAlchemyJobStore(url=‘sqlite:///jobs.sqlite’) } scheduler = BackgroundScheduler(jobstores=jobstores) # 使用Cron表达式添加任务 scheduler.add_job(agent_task, ‘cron’, hour=1, id=‘daily_task’) # 或者直接使用字符串表达式 scheduler.add_job(agent_task, ‘cron’, minute=‘*/25’, id=‘frequent_task’) scheduler.start() # 主程序可以继续做其他事

  关键优势：任务定义可以持久化到数据库，即使程序重启，定时任务也不会丢失。

• Java Spring的 @Scheduled：

  import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @Component public class ScheduledAgent { // 每天凌晨1点执行 @Scheduled(cron = “0 0 1 * * ?”) public void dailyTask() { // Agent逻辑 } // 每25分钟执行 @Scheduled(cron = “0 */25 * * * ?”) public void frequentTask() { // Agent逻辑 } }

  需要在启动类加上@EnableScheduling注解。Spring会将任务托管到线程池，非常方便。

模式三：基于消息队列的延迟任务这不是严格的“定时”，而是“延迟执行”。适用于“30分钟后重试失败任务”、“2小时后给用户发送提醒”这类场景。前面提到的Redis Sorted Set (ZADD+ZRANGEBYSCORE) 或 RabbitMQ的延迟交换机插件、死信队列都能实现。

如何选择？

• 简单、独立的任务：选操作系统Cron。
• 需要与应用程序状态紧密交互、任务需持久化：选APScheduler、Quartz、Spring@Scheduled。
• 需要高精度、分布式调度：考虑使用专门的分布式任务调度框架，如Airflow（更适合数据管道）、Celery Beat（配合Celery）或XXL-JOB。
• 仅仅是延迟触发：用消息队列的延迟功能。

4.3 定时任务的高级考量与避坑

1. 任务幂等性：定时任务可能因为各种原因（如执行时间过长、服务器时间漂移、手动触发）被重复执行。你的Agent任务逻辑必须保证幂等性，即同一任务执行多次的结果与执行一次相同。例如，通过任务ID和状态锁来防止重复处理同一数据。
2. 任务执行时长超过间隔：如果你的任务需要跑30分钟，但定时是每25分钟一次，就会发生任务堆积。解决方案：a) 加分布式锁，确保同一时间只有一个实例运行；b) 改用更长的时间间隔；c) 将大任务拆分成可独立执行的小任务。
3. 错过执行（Missed Fire）：服务器在任务触发时间点宕机了怎么办？好的调度器（如APScheduler、Quartz）有misfire_grace_time（错过容忍时间）配置，可以在服务器恢复后补执行。你需要根据业务决定是忽略、立即执行还是只执行最后一次。
4. 时间与时区：服务器时区、数据库时区和Cron表达式使用的时区必须一致！最好全部使用UTC时间，在展示给用户时再转换。这是最容易出错的点之一。
5. 监控与告警：定时任务必须配日志和监控。任务成功/失败要有记录，失败后最好能自动重试，并设置失败阈值，超过后发送告警（邮件、钉钉、Slack）。

5. 实战整合：构建一个完整的持久化Agent服务

现在，我们把任务持久化、后台执行和定时唤醒组合起来，设计一个能投入生产环境的小型Agent服务框架。我们以Python为例，使用FastAPI提供Web API，Celery处理后台任务，APScheduler负责定时触发，Redis作为Broker和结果后端，MySQL存储任务状态。

5.1 系统架构与组件职责

1. FastAPI (Web Layer)：提供RESTful API，接收用户创建任务的请求，并立即返回一个任务ID。它不执行耗时操作。
2. Celery (Task Queue Layer)：真正的任务执行者。Worker进程从Redis消息队列中取出任务并执行。它负责调用Agent的核心逻辑。
3. APScheduler (Scheduler Layer)：内嵌在Web服务中，负责按Cron表达式定时向Celery队列发送任务消息。
4. Redis (Message Broker & Cache)：作为Celery的消息中间件，传递任务；同时缓存活跃任务的上下文，加速读取。
5. MySQL (Persistent Storage)：持久化存储所有任务的元数据、最终状态和结果。提供任务查询和管理能力。

5.2 核心代码实现

第一步：定义数据模型（models.py）

from sqlalchemy import Column, Integer, String, DateTime, Text, Enum from sqlalchemy.ext.declarative import declarative_base import enum Base = declarative_base() class TaskStatus(enum.Enum): PENDING = “pending” RUNNING = “running” SUCCESS = “success” FAILED = “failed” class Task(Base): __tablename__ = ‘tasks’ id = Column(String(64), primary_key=True) # 使用UUID name = Column(String(255)) status = Column(Enum(TaskStatus), default=TaskStatus.PENDING) progress = Column(Integer, default=0) input_params = Column(Text) # JSON字符串 result = Column(Text) # JSON字符串 checkpoint = Column(Text) # 序列化的Agent上下文 created_at = Column(DateTime) updated_at = Column(DateTime) scheduled_for = Column(DateTime) # 定时任务计划执行时间

第二步：配置Celery与任务（celery_app.py）

from celery import Celery from celery.utils.log import get_task_logger from .models import Task, TaskStatus, SessionLocal import json import uuid logger = get_task_logger(__name__) # 创建Celery应用，指定Broker和Backend app = Celery(‘agent_worker’, broker=‘redis://localhost:6379/1’, backend=‘redis://localhost:6379/2’) @app.task(bind=True) def execute_agent_task(self, task_id, agent_type, params): """执行Agent任务的Celery Task""" db = SessionLocal() try: task = db.query(Task).filter(Task.id == task_id).first() if not task: logger.error(f“Task {task_id} not found.”) return task.status = TaskStatus.RUNNING db.commit() # 1. 加载检查点（如果存在） context = {} if task.checkpoint: context = json.loads(task.checkpoint) logger.info(f“Resumed task {task_id} from checkpoint.”) # 2. 这里是你的Agent核心逻辑 # 模拟一个长任务，并定期保存进度和检查点 total_steps = 100 for i in range(context.get(‘current_step’, 0), total_steps): # 模拟工作 # ... 你的Agent处理逻辑 ... # 3. 定期更新进度和保存检查点（例如每10步） progress = int((i + 1) / total_steps * 100) self.update_state(state=‘PROGRESS’, meta={‘progress’: progress}) if (i + 1) % 10 == 0: checkpoint_data = { ‘current_step’: i + 1, ‘intermediate_data’: ‘...’, # 你的中间数据 } task.progress = progress task.checkpoint = json.dumps(checkpoint_data) db.commit() logger.info(f“Task {task_id} checkpoint saved at step {i+1}.”) # 4. 任务完成 task.status = TaskStatus.SUCCESS task.progress = 100 task.result = json.dumps({“message”: “Task completed successfully”}) task.checkpoint = None # 清理检查点 db.commit() logger.info(f“Task {task_id} completed.”) return {“task_id”: task_id, “status”: “success”} except Exception as e: logger.exception(f“Task {task_id} failed: {e}”) if db: task.status = TaskStatus.FAILED task.result = json.dumps({“error”: str(e)}) db.commit() raise self.retry(exc=e, countdown=60) # 失败后60秒重试 finally: if db: db.close()

第三步：FastAPI Web接口与调度器（main.py）

from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from .celery_app import execute_agent_task from .models import Task, TaskStatus, SessionLocal, engine from .scheduler import scheduler # 假设scheduler在另一个模块初始化 import uuid from datetime import datetime Base.metadata.create_all(bind=engine) # 创建表 app = FastAPI() class TaskRequest(BaseModel): name: str agent_type: str params: dict schedule_cron: str = None # 可选，Cron表达式 @app.post(“/tasks”) async def create_task(request: TaskRequest, background_tasks: BackgroundTasks): """创建即时或定时任务""" task_id = str(uuid.uuid4()) db = SessionLocal() task = Task( id=task_id, name=request.name, status=TaskStatus.PENDING, input_params=json.dumps(request.params), created_at=datetime.utcnow(), updated_at=datetime.utcnow() ) db.add(task) db.commit() db.close() if request.schedule_cron: # 定时任务：添加到APScheduler scheduler.add_job( func=execute_agent_task.delay, # 注意：这里传递的是Celery的delay方法 trigger=‘cron’, args=[task_id, request.agent_type, request.params], id=task_id, replace_existing=True, **parse_cron(request.schedule_cron) # 将Cron字符串解析为字典参数 ) return {“task_id”: task_id, “message”: “Scheduled task created”, “schedule”: request.schedule_cron} else: # 即时任务：发送到Celery队列 execute_agent_task.delay(task_id, request.agent_type, request.params) return {“task_id”: task_id, “message”: “Task queued for immediate execution”} @app.get(“/tasks/{task_id}”) async def get_task_status(task_id: str): """查询任务状态""" db = SessionLocal() task = db.query(Task).filter(Task.id == task_id).first() db.close() if not task: return {“error”: “Task not found”} return { “id”: task.id, “status”: task.status.value, “progress”: task.progress, “result”: json.loads(task.result) if task.result else None, “created_at”: task.created_at.isoformat() if task.created_at else None, “updated_at”: task.updated_at.isoformat() if task.updated_at else None, } # 启动时加载APScheduler @app.on_event(“startup”) async def startup_event(): scheduler.start() # 可以从数据库加载未完成的定时任务重新调度 @app.on_event(“shutdown”) async def shutdown_event(): scheduler.shutdown()

第四步：独立调度器模块（scheduler.py）

from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from apscheduler.executors.pool import ThreadPoolExecutor jobstores = { ‘default’: SQLAlchemyJobStore(url=‘sqlite:///jobs.sqlite’) # 或你的MySQL URL } executors = { ‘default’: ThreadPoolExecutor(20), } job_defaults = { ‘coalesce’: False, # 是否合并多次未执行的触发 ‘max_instances’: 3, # 同一个任务允许的最大并发实例数 } scheduler = BackgroundScheduler( jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=‘UTC’ # 统一使用UTC时区 )

5.3 部署与运维要点

1. 启动服务：

   # 终端1：启动Web服务 uvicorn main:app --host 0.0.0.0 --port 8000 # 终端2：启动Celery Worker celery -A celery_app worker --loglevel=info --concurrency=4 # 终端3：启动Celery Beat（如果需要Celery内置的定时，本例中用APScheduler替代） # celery -A celery_app beat --loglevel=info

2. 进程管理：使用Supervisor或Systemd来管理这三个进程，确保它们崩溃后能自动重启。

3. 监控：

   • 任务队列：使用Flower监控Celery队列和Worker状态。
   • 数据库：监控MySQL连接数和慢查询。
   • 日志：所有组件（FastAPI, Celery, APScheduler）的日志集中收集到ELK或Graylog。

4. 高可用考虑：生产环境中，Redis、MySQL建议做主从或集群。Celery Worker可以水平扩展多个。APScheduler在多个Web实例上运行时，需要使用支持分布式锁的JobStore（如基于数据库的），或者只在一个实例上启用调度器。

这个框架提供了一个坚实的起点，你可以根据具体Agent的逻辑填充execute_agent_task函数中的核心处理部分。它解决了持久化（MySQL+检查点）、后台执行（Celery）、定时唤醒（APScheduler）三大核心问题，并且具备了基本的可观测性和可靠性。