跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

调度任务

最后更新 查看 MarkdownAgent 设置

调度在未来执行的任务——无论是几秒后、特定日期/时间，还是循环 cron 计划。调度任务在 Agent 重启后仍然保留，并持久化到 SQLite。

调度任务可执行与用户请求或消息相同的任何操作：发起请求、查询数据库、发送邮件、读写状态。调度任务可调用 Agent 上的任意常规方法。

概览

调度系统支持四种模式：

模式 语法 用例
Delayed this.schedule(60, ...) 60 秒后运行
Scheduled this.schedule(new Date(...), ...) 在指定时间运行
Cron this.schedule("0 8 * * *", ...) 按 recurring schedule 运行
Interval this.scheduleEvery(30, ...) 每 30 秒运行

底层实现使用 Durable Object alarms 在正确时间唤醒 agent。任务存储在 SQLite 表中并按顺序执行。

快速入门

import { Agent } from "agents";

export class ReminderAgent extends Agent {
	async onRequest(request) {
		const url = new URL(request.url);

		// Schedule in 30 seconds
		await this.schedule(30, "sendReminder", {
			message: "Check your email",
		});

		// Schedule at specific time
		await this.schedule(new Date("2025-02-01T09:00:00Z"), "sendReminder", {
			message: "Monthly report due",
		});

		// Schedule recurring (every day at 8am)
		await this.schedule("0 8 * * *", "dailyDigest", {
			userId: url.searchParams.get("userId"),
		});

		return new Response("Scheduled!");
	}

	async sendReminder(payload) {
		console.log(`Reminder: ${payload.message}`);
		// Send notification, email, etc.
	}

	async dailyDigest(payload) {
		console.log(`Sending daily digest to ${payload.userId}`);
		// Generate and send digest
	}
}
import { Agent } from "agents";

export class ReminderAgent extends Agent {
	async onRequest(request: Request) {
		const url = new URL(request.url);

		// Schedule in 30 seconds
		await this.schedule(30, "sendReminder", {
			message: "Check your email",
		});

		// Schedule at specific time
		await this.schedule(new Date("2025-02-01T09:00:00Z"), "sendReminder", {
			message: "Monthly report due",
		});

		// Schedule recurring (every day at 8am)
		await this.schedule("0 8 * * *", "dailyDigest", {
			userId: url.searchParams.get("userId"),
		});

		return new Response("Scheduled!");
	}

	async sendReminder(payload: { message: string }) {
		console.log(`Reminder: ${payload.message}`);
		// Send notification, email, etc.
	}

	async dailyDigest(payload: { userId: string }) {
		console.log(`Sending daily digest to ${payload.userId}`);
		// Generate and send digest
	}
}

调度模式

延迟执行

传入数字以在秒为单位的延迟后调度任务：

// Run in 10 seconds
await this.schedule(10, "processTask", { taskId: "123" });

// Run in 5 minutes (300 seconds)
await this.schedule(300, "sendFollowUp", { email: "[email protected]" });

// Run in 1 hour
await this.schedule(3600, "checkStatus", { orderId: "abc" });
// Run in 10 seconds
await this.schedule(10, "processTask", { taskId: "123" });

// Run in 5 minutes (300 seconds)
await this.schedule(300, "sendFollowUp", { email: "[email protected]" });

// Run in 1 hour
await this.schedule(3600, "checkStatus", { orderId: "abc" });

用例：

  • 防抖快速事件
  • 延迟通知（「购物车中还有商品」）
  • 带退避的重试
  • 速率限制

定时执行

传入 Date 对象以在特定时间调度任务：

// Run tomorrow at noon
const tomorrow = new Date();
tomorrow.setDate(tomorrow.getDate() + 1);
tomorrow.setHours(12, 0, 0, 0);
await this.schedule(tomorrow, "sendReminder", { message: "Meeting time!" });

// Run at a specific timestamp
await this.schedule(new Date("2025-06-15T14:30:00Z"), "triggerEvent", {
	eventId: "conference-2025",
});

// Run in 2 hours using Date math
const twoHoursFromNow = new Date(Date.now() + 2 * 60 * 60 * 1000);
await this.schedule(twoHoursFromNow, "checkIn", {});
// Run tomorrow at noon
const tomorrow = new Date();
tomorrow.setDate(tomorrow.getDate() + 1);
tomorrow.setHours(12, 0, 0, 0);
await this.schedule(tomorrow, "sendReminder", { message: "Meeting time!" });

// Run at a specific timestamp
await this.schedule(new Date("2025-06-15T14:30:00Z"), "triggerEvent", {
	eventId: "conference-2025",
});

// Run in 2 hours using Date math
const twoHoursFromNow = new Date(Date.now() + 2 * 60 * 60 * 1000);
await this.schedule(twoHoursFromNow, "checkIn", {});

用例：

  • 预约提醒
  • 截止日期通知
  • 定时内容发布
  • 基于时间的触发器

循环（cron）

传入 cron 表达式字符串以进行循环调度：

// Every day at 8:00 AM
await this.schedule("0 8 * * *", "dailyReport", {});

// Every hour
await this.schedule("0 * * * *", "hourlyCheck", {});

// Every Monday at 9:00 AM
await this.schedule("0 9 * * 1", "weeklySync", {});

// Every 15 minutes
await this.schedule("*/15 * * * *", "pollForUpdates", {});

// First day of every month at midnight
await this.schedule("0 0 1 * *", "monthlyCleanup", {});
// Every day at 8:00 AM
await this.schedule("0 8 * * *", "dailyReport", {});

// Every hour
await this.schedule("0 * * * *", "hourlyCheck", {});

// Every Monday at 9:00 AM
await this.schedule("0 9 * * 1", "weeklySync", {});

// Every 15 minutes
await this.schedule("*/15 * * * *", "pollForUpdates", {});

// First day of every month at midnight
await this.schedule("0 0 1 * *", "monthlyCleanup", {});

Cron 语法： minute hour day month weekday

字段 取值 特殊字符
分钟 0-59 * , - /
小时 0-23 * , - /
日（月） 1-31 * , - /
月 1-12 * , - /
星期 0-6 (0=Sunday) * , - /

常见模式：

"* * * * *"; // Every minute
"*/5 * * * *"; // Every 5 minutes
"0 * * * *"; // Every hour (on the hour)
"0 0 * * *"; // Every day at midnight
"0 8 * * 1-5"; // Weekdays at 8am
"0 0 * * 0"; // Every Sunday at midnight
"0 0 1 * *"; // First of every month
"* * * * *"; // Every minute
"*/5 * * * *"; // Every 5 minutes
"0 * * * *"; // Every hour (on the hour)
"0 0 * * *"; // Every day at midnight
"0 8 * * 1-5"; // Weekdays at 8am
"0 0 * * 0"; // Every Sunday at midnight
"0 0 1 * *"; // First of every month

用例：

  • 日报/周报
  • 周期性清理作业
  • 轮询外部服务
  • 健康检查
  • 订阅续期

Cron 调度默认具有幂等性——使用相同 cron 表达式、回调与载荷多次调用 schedule() 会返回现有调度，而非创建重复项。因此在 onStart() 中设置 cron 调度是安全的。

Interval

使用 scheduleEvery() 以固定间隔（秒）运行任务。与 cron 不同，间隔支持亚分钟精度与任意时长：

// Poll every 30 seconds
await this.scheduleEvery(30, "poll", { source: "api" });

// Health check every 45 seconds
await this.scheduleEvery(45, "healthCheck", {});

// Sync every 90 seconds (1.5 minutes - cannot be expressed in cron)
await this.scheduleEvery(90, "syncData", { destination: "warehouse" });
// Poll every 30 seconds
await this.scheduleEvery(30, "poll", { source: "api" });

// Health check every 45 seconds
await this.scheduleEvery(45, "healthCheck", {});

// Sync every 90 seconds (1.5 minutes - cannot be expressed in cron)
await this.scheduleEvery(90, "syncData", { destination: "warehouse" });

与 cron 的主要区别：

特性 Cron Interval
最小粒度 1 分钟 1 秒
任意间隔 否（须符合 cron 模式） 是
固定调度 是（例如「每天 8:00」） 否（相对于启动时间）
重叠防护 否 是（内置）

幂等性：

scheduleEvery() 在回调名称、间隔与载荷组合上具有幂等性——使用相同参数多次调用不会创建重复调度。因此在每次 Durable Object 唤醒时都会运行的 onStart() 中调用是安全的：

class MyAgent extends Agent {
	async onStart() {
		// Safe to call on every wake — only one schedule is created
		await this.scheduleEvery(30, "poll", { source: "api" });
	}
}
class MyAgent extends Agent {
	async onStart() {
		// Safe to call on every wake — only one schedule is created
		await this.scheduleEvery(30, "poll", { source: "api" });
	}
}

不同的间隔或载荷会创建新的独立调度。

重叠防护：

若回调执行时间超过间隔，下一次执行会被跳过（不会排队）。这可防止资源失控：

class PollingAgent extends Agent {
	async poll() {
		// If this takes 45 seconds and interval is 30 seconds,
		// the next poll is skipped (with a warning logged)
		const data = await slowExternalApi();
		await this.processData(data);
	}
}

// Set up 30-second interval
await this.scheduleEvery(30, "poll", {});
class PollingAgent extends Agent {
	async poll() {
		// If this takes 45 seconds and interval is 30 seconds,
		// the next poll is skipped (with a warning logged)
		const data = await slowExternalApi();
		await this.processData(data);
	}
}

// Set up 30-second interval
await this.scheduleEvery(30, "poll", {});

发生 skip 时，日志中会出现警告：

Skipping interval schedule abc123: previous execution still running

错误恢复：

若回调抛出错误，间隔仍会继续——仅该次执行失败：

class SyncAgent extends Agent {
	async syncData() {
		// Even if this throws, the interval keeps running
		const response = await fetch("https://api.example.com/data");
		if (!response.ok) throw new Error("Sync failed");
		// ...
	}
}
class SyncAgent extends Agent {
	async syncData() {
		// Even if this throws, the interval keeps running
		const response = await fetch("https://api.example.com/data");
		if (!response.ok) throw new Error("Sync failed");
		// ...
	}
}

用例：

  • 亚分钟级轮询（每 10、30、45 秒）
  • 无法映射到 cron 的 interval（每 90 秒、每 7 分钟）
  • 精确控制的 API 速率限制轮询
  • 实时数据同步

管理调度任务

获取 schedule

按 ID 检索调度任务：

const schedule = await this.getScheduleById(scheduleId);

if (schedule) {
	console.log(
		`Task ${schedule.id} will run at ${new Date(schedule.time * 1000)}`,
	);
	console.log(`Callback: ${schedule.callback}`);
	console.log(`Type: ${schedule.type}`); // "scheduled" | "delayed" | "cron" | "interval"
} else {
	console.log("Schedule not found");
}
const schedule = await this.getScheduleById(scheduleId);

if (schedule) {
	console.log(
		`Task ${schedule.id} will run at ${new Date(schedule.time * 1000)}`,
	);
	console.log(`Callback: ${schedule.callback}`);
	console.log(`Type: ${schedule.type}`); // "scheduled" | "delayed" | "cron" | "interval"
} else {
	console.log("Schedule not found");
}

列出 schedule

使用可选筛选条件查询调度任务：

// Get all scheduled tasks
const allSchedules = await this.listSchedules();

// Get only cron jobs
const cronJobs = await this.listSchedules({ type: "cron" });

// Get tasks in the next hour
const upcoming = await this.listSchedules({
	timeRange: {
		start: new Date(),
		end: new Date(Date.now() + 60 * 60 * 1000),
	},
});

// Get a specific task by ID
const specific = await this.listSchedules({ id: "abc123" });

// Combine filters
const upcomingCronJobs = await this.listSchedules({
	type: "cron",
	timeRange: {
		start: new Date(),
		end: new Date(Date.now() + 24 * 60 * 60 * 1000),
	},
});
// Get all scheduled tasks
const allSchedules = await this.listSchedules();

// Get only cron jobs
const cronJobs = await this.listSchedules({ type: "cron" });

// Get tasks in the next hour
const upcoming = await this.listSchedules({
	timeRange: {
		start: new Date(),
		end: new Date(Date.now() + 60 * 60 * 1000),
	},
});

// Get a specific task by ID
const specific = await this.listSchedules({ id: "abc123" });

// Combine filters
const upcomingCronJobs = await this.listSchedules({
	type: "cron",
	timeRange: {
		start: new Date(),
		end: new Date(Date.now() + 24 * 60 * 60 * 1000),
	},
});

取消 schedule

在执行前移除调度任务：

const cancelled = await this.cancelSchedule(scheduleId);

if (cancelled) {
	console.log("Schedule cancelled successfully");
} else {
	console.log("Schedule not found (may have already executed)");
}
const cancelled = await this.cancelSchedule(scheduleId);

if (cancelled) {
	console.log("Schedule cancelled successfully");
} else {
	console.log("Schedule not found (may have already executed)");
}

示例：可取消的提醒

class ReminderAgent extends Agent {
	async setReminder(userId, message, delaySeconds) {
		const schedule = await this.schedule(delaySeconds, "sendReminder", {
			userId,
			message,
		});

		// Store the schedule ID so user can cancel later
		this.sql`
      INSERT INTO user_reminders (user_id, schedule_id, message)
      VALUES (${userId}, ${schedule.id}, ${message})
    `;

		return schedule.id;
	}

	async cancelReminder(scheduleId) {
		const cancelled = await this.cancelSchedule(scheduleId);

		if (cancelled) {
			this.sql`DELETE FROM user_reminders WHERE schedule_id = ${scheduleId}`;
		}

		return cancelled;
	}

	async sendReminder(payload) {
		// Send the reminder...

		// Clean up the record
		this.sql`DELETE FROM user_reminders WHERE user_id = ${payload.userId}`;
	}
}
class ReminderAgent extends Agent {
	async setReminder(userId: string, message: string, delaySeconds: number) {
		const schedule = await this.schedule(delaySeconds, "sendReminder", {
			userId,
			message,
		});

		// Store the schedule ID so user can cancel later
		this.sql`
      INSERT INTO user_reminders (user_id, schedule_id, message)
      VALUES (${userId}, ${schedule.id}, ${message})
    `;

		return schedule.id;
	}

	async cancelReminder(scheduleId: string) {
		const cancelled = await this.cancelSchedule(scheduleId);

		if (cancelled) {
			this.sql`DELETE FROM user_reminders WHERE schedule_id = ${scheduleId}`;
		}

		return cancelled;
	}

	async sendReminder(payload: { userId: string; message: string }) {
		// Send the reminder...

		// Clean up the record
		this.sql`DELETE FROM user_reminders WHERE user_id = ${payload.userId}`;
	}
}

Schedule 对象

创建或检索 schedule 时，会得到 Schedule 对象：

type Schedule<T> = {
	id: string; // Unique identifier
	callback: string; // Method name to call
	payload: T; // Data passed to the callback
	time: number; // Unix timestamp (seconds) of next execution
} & (
	| { type: "scheduled" } // One-time at specific date
	| { type: "delayed"; delayInSeconds: number } // One-time after delay
	| { type: "cron"; cron: string } // Recurring (cron expression)
	| { type: "interval"; intervalSeconds: number } // Recurring (fixed interval)
);

示例：

const schedule = await this.schedule(60, "myTask", { foo: "bar" });

console.log(schedule);
// {
//   id: "abc123xyz",
//   callback: "myTask",
//   payload: { foo: "bar" },
//   time: 1706745600,
//   type: "delayed",
//   delayInSeconds: 60
// }
const schedule = await this.schedule(60, "myTask", { foo: "bar" });

console.log(schedule);
// {
//   id: "abc123xyz",
//   callback: "myTask",
//   payload: { foo: "bar" },
//   time: 1706745600,
//   type: "delayed",
//   delayInSeconds: 60
// }

模式

从回调重新调度

对于动态循环调度，在回调内调度下一次运行：

class PollingAgent extends Agent {
	async startPolling(intervalSeconds) {
		await this.schedule(intervalSeconds, "poll", { interval: intervalSeconds });
	}

	async poll(payload) {
		try {
			const data = await fetch("https://api.example.com/updates");
			await this.processUpdates(await data.json());
		} catch (error) {
			console.error("Polling failed:", error);
		}

		// Schedule the next poll (regardless of success/failure)
		await this.schedule(payload.interval, "poll", payload);
	}

	async stopPolling() {
		// Cancel all polling schedules
		const schedules = await this.listSchedules({ type: "delayed" });
		for (const schedule of schedules) {
			if (schedule.callback === "poll") {
				await this.cancelSchedule(schedule.id);
			}
		}
	}
}
class PollingAgent extends Agent {
	async startPolling(intervalSeconds: number) {
		await this.schedule(intervalSeconds, "poll", { interval: intervalSeconds });
	}

	async poll(payload: { interval: number }) {
		try {
			const data = await fetch("https://api.example.com/updates");
			await this.processUpdates(await data.json());
		} catch (error) {
			console.error("Polling failed:", error);
		}

		// Schedule the next poll (regardless of success/failure)
		await this.schedule(payload.interval, "poll", payload);
	}

	async stopPolling() {
		// Cancel all polling schedules
		const schedules = await this.listSchedules({ type: "delayed" });
		for (const schedule of schedules) {
			if (schedule.callback === "poll") {
				await this.cancelSchedule(schedule.id);
			}
		}
	}
}

指数退避重试

class RetryAgent extends Agent {
	async attemptTask(payload) {
		try {
			await this.doWork(payload.taskId);
			console.log(
				`Task ${payload.taskId} succeeded on attempt ${payload.attempt}`,
			);
		} catch (error) {
			if (payload.attempt >= payload.maxAttempts) {
				console.error(
					`Task ${payload.taskId} failed after ${payload.maxAttempts} attempts`,
				);
				return;
			}

			// Exponential backoff: 2^attempt seconds (2s, 4s, 8s, 16s...)
			const delaySeconds = Math.pow(2, payload.attempt);

			await this.schedule(delaySeconds, "attemptTask", {
				...payload,
				attempt: payload.attempt + 1,
			});

			console.log(`Retrying task ${payload.taskId} in ${delaySeconds}s`);
		}
	}

	async doWork(taskId) {
		// Your actual work here
	}
}
class RetryAgent extends Agent {
	async attemptTask(payload: {
		taskId: string;
		attempt: number;
		maxAttempts: number;
	}) {
		try {
			await this.doWork(payload.taskId);
			console.log(
				`Task ${payload.taskId} succeeded on attempt ${payload.attempt}`,
			);
		} catch (error) {
			if (payload.attempt >= payload.maxAttempts) {
				console.error(
					`Task ${payload.taskId} failed after ${payload.maxAttempts} attempts`,
				);
				return;
			}

			// Exponential backoff: 2^attempt seconds (2s, 4s, 8s, 16s...)
			const delaySeconds = Math.pow(2, payload.attempt);

			await this.schedule(delaySeconds, "attemptTask", {
				...payload,
				attempt: payload.attempt + 1,
			});

			console.log(`Retrying task ${payload.taskId} in ${delaySeconds}s`);
		}
	}

	async doWork(taskId: string) {
		// Your actual work here
	}
}

自毁 Agent

可以在调度回调内安全调用 this.destroy()：

class TemporaryAgent extends Agent {
	async onStart() {
		// Self-destruct in 24 hours
		await this.schedule(24 * 60 * 60, "cleanup", {});
	}

	async cleanup() {
		// Perform final cleanup
		console.log("Agent lifetime expired, cleaning up...");

		// This is safe to call from a scheduled callback
		await this.destroy();
	}
}
class TemporaryAgent extends Agent {
	async onStart() {
		// Self-destruct in 24 hours
		await this.schedule(24 * 60 * 60, "cleanup", {});
	}

	async cleanup() {
		// Perform final cleanup
		console.log("Agent lifetime expired, cleaning up...");

		// This is safe to call from a scheduled callback
		await this.destroy();
	}
}

AI 辅助调度

SDK 包含使用 AI 解析自然语言调度请求的工具。

getSchedulePrompt()

返回用于将自然语言解析为调度参数的系统提示词：

import { getSchedulePrompt, scheduleSchema } from "agents";
import { generateObject } from "ai";
import { openai } from "@ai-sdk/openai";

class SmartScheduler extends Agent {
	async parseScheduleRequest(userInput) {
		const result = await generateObject({
			model: openai("gpt-4o"),
			system: getSchedulePrompt({ date: new Date() }),
			prompt: userInput,
			schema: scheduleSchema,
		});

		return result.object;
	}

	async handleUserRequest(input) {
		// Parse: "remind me to call mom tomorrow at 3pm"
		const parsed = await this.parseScheduleRequest(input);

		// parsed = {
		//   description: "call mom",
		//   when: {
		//     type: "scheduled",
		//     date: "2025-01-30T15:00:00Z"
		//   }
		// }

		if (parsed.when.type === "scheduled" && parsed.when.date) {
			await this.schedule(new Date(parsed.when.date), "sendReminder", {
				message: parsed.description,
			});
		} else if (parsed.when.type === "delayed" && parsed.when.delayInSeconds) {
			await this.schedule(parsed.when.delayInSeconds, "sendReminder", {
				message: parsed.description,
			});
		} else if (parsed.when.type === "cron" && parsed.when.cron) {
			await this.schedule(parsed.when.cron, "sendReminder", {
				message: parsed.description,
			});
		}
	}

	async sendReminder(payload) {
		console.log(`Reminder: ${payload.message}`);
	}
}
import { getSchedulePrompt, scheduleSchema } from "agents";
import { generateObject } from "ai";
import { openai } from "@ai-sdk/openai";

class SmartScheduler extends Agent {
	async parseScheduleRequest(userInput: string) {
		const result = await generateObject({
			model: openai("gpt-4o"),
			system: getSchedulePrompt({ date: new Date() }),
			prompt: userInput,
			schema: scheduleSchema,
		});

		return result.object;
	}

	async handleUserRequest(input: string) {
		// Parse: "remind me to call mom tomorrow at 3pm"
		const parsed = await this.parseScheduleRequest(input);

		// parsed = {
		//   description: "call mom",
		//   when: {
		//     type: "scheduled",
		//     date: "2025-01-30T15:00:00Z"
		//   }
		// }

		if (parsed.when.type === "scheduled" && parsed.when.date) {
			await this.schedule(new Date(parsed.when.date), "sendReminder", {
				message: parsed.description,
			});
		} else if (parsed.when.type === "delayed" && parsed.when.delayInSeconds) {
			await this.schedule(parsed.when.delayInSeconds, "sendReminder", {
				message: parsed.description,
			});
		} else if (parsed.when.type === "cron" && parsed.when.cron) {
			await this.schedule(parsed.when.cron, "sendReminder", {
				message: parsed.description,
			});
		}
	}

	async sendReminder(payload: { message: string }) {
		console.log(`Reminder: ${payload.message}`);
	}
}

scheduleSchema

用于验证已解析调度数据的 Zod schema。在 when.type 上使用 discriminated union，使每个变体仅包含所需字段：

import { scheduleSchema } from "agents";

// The schema is a discriminated union:
// {
//   description: string,
//   when:
//     | { type: "scheduled", date: string }       // ISO 8601 date string
//     | { type: "delayed", delayInSeconds: number }
//     | { type: "cron", cron: string }
//     | { type: "no-schedule" }
// }
import { scheduleSchema } from "agents";

// The schema is a discriminated union:
// {
//   description: string,
//   when:
//     | { type: "scheduled", date: string }       // ISO 8601 date string
//     | { type: "delayed", delayInSeconds: number }
//     | { type: "cron", cron: string }
//     | { type: "no-schedule" }
// }

调度 vs Queue vs Workflows

特性 Queue Scheduling Workflows
何时 立即（FIFO） 未来时间 未来时间
执行 顺序 在 scheduled 时间 多步
重试 内置 内置 自动
持久化 SQLite SQLite Workflow 引擎
Recurring 否 是（cron） 否（使用 scheduling）
复杂逻辑 否 否 是
人工审批 否 否 是

使用 Queue 当：

  • 需要后台处理且不阻塞响应
  • 任务应尽快运行但无需阻塞
  • 顺序很重要（FIFO）

使用 Scheduling 当：

  • 任务需要在特定时间运行
  • 需要 recurring 作业（cron）
  • 延迟执行（debouncing、重试）

使用 Workflows 当：

  • 带依赖的多步流程
  • 带退避的自动重试
  • 人工介入审批
  • 长时间运行任务（数分钟到数小时）

API 参考

schedule()

async schedule<T>(
  when: Date | string | number,
  callback: keyof this,
  payload?: T,
  options?: { retry?: RetryOptions; idempotent?: boolean }
): Promise<Schedule<T>>

调度未来执行的任务。

参数：

  • when - 何时执行：number（秒延迟）、Date（特定时间）或 string（cron 表达式）
  • callback - 要调用的方法名
  • payload - 传给回调的数据（须可 JSON 序列化）
  • options.retry - 可选重试配置。详见 重试
  • options.idempotent - 按回调 + 载荷去重。cron 调度默认为 true，延迟与基于 Date 的调度默认为 false

返回值： 含任务详情的 Schedule 对象

幂等性：

Cron 调度默认具有幂等性——使用相同回调、cron 表达式与载荷多次调用 schedule("0 * * * *", "tick") 会返回现有调度，而非创建重复项。设置 idempotent: false 可覆盖此行为。

对于延迟与基于 Date 的调度，设置 idempotent: true 可启用相同去重行为（按回调 + 载荷匹配）。在 onStart() 中调用 schedule() 时尤其有用，可避免 Durable Object 重启后累积重复行：

class MyAgent extends Agent {
	async onStart() {
		// Without idempotent: true, this creates a new row on every DO restart
		await this.schedule(3600, "hourlyCleanup", {}, { idempotent: true });
	}
}
class MyAgent extends Agent {
	async onStart() {
		// Without idempotent: true, this creates a new row on every DO restart
		await this.schedule(3600, "hourlyCleanup", {}, { idempotent: true });
	}
}

scheduleEvery()

async scheduleEvery<T>(
  intervalSeconds: number,
  callback: keyof this,
  payload?: T,
  options?: { retry?: RetryOptions }
): Promise<Schedule<T>>

调度以固定间隔重复运行的任务。

参数：

  • intervalSeconds - 执行间隔秒数（须大于 0）
  • callback - 要调用的方法名
  • payload - 传给回调的数据（须可 JSON 序列化）
  • options.retry - 可选重试配置。详见 重试

返回值： type: "interval" 的 Schedule 对象

行为：

  • 首次执行在 intervalSeconds 之后（非立即）
  • 若回调仍在运行且下次执行到期，则跳过（重叠防护）
  • 若回调抛出错误，间隔仍继续
  • 使用 cancelSchedule(id) 取消以停止整个间隔

getScheduleById()

async getScheduleById(id: string): Promise<Schedule<unknown> | undefined>

按 ID 获取调度任务。未找到时返回 undefined。此方法在顶层 Agent 与子 Agent 中均可用。

listSchedules()

async listSchedules(criteria?: {
  id?: string;
  type?: "scheduled" | "delayed" | "cron" | "interval";
  timeRange?: { start?: Date; end?: Date };
}): Promise<Schedule<unknown>[]>

获取符合条件的调度任务。此方法在顶层 Agent 与子 Agent 中均可用。

getSchedule()

getSchedule<T>(id: string): Schedule<T> | undefined

已弃用。同步按 ID 获取调度任务。此方法仅适用于顶层 Agent。请改用 await this.getScheduleById(id)。

getSchedules()

getSchedules<T>(criteria?: {
  id?: string;
  type?: "scheduled" | "delayed" | "cron" | "interval";
  timeRange?: { start?: Date; end?: Date };
}): Schedule<T>[]

已弃用。同步获取符合条件的调度任务。此方法仅适用于顶层 Agent。请改用 await this.listSchedules(criteria)。

cancelSchedule()

async cancelSchedule(id: string): Promise<boolean>

取消调度任务。取消成功返回 true，未找到返回 false。

keepAlive()

async keepAlive(): Promise<() => void>

通过持有 30 秒 alarm 支持的心跳引用，防止 Durable Object 因不活动被驱逐。返回释放函数，调用时释放心跳。释放函数具有幂等性——多次调用安全。

工作完成后务必调用释放函数——否则心跳会无限持续。

const dispose = await this.keepAlive();
try {
	// Long-running work that must not be interrupted
	const result = await longRunningComputation();
	await sendResults(result);
} finally {
	dispose();
}
const dispose = await this.keepAlive();
try {
	// Long-running work that must not be interrupted
	const result = await longRunningComputation();
	await sendResults(result);
} finally {
	dispose();
}

keepAliveWhile()

async keepAliveWhile<T>(fn: () => Promise<T>): Promise<T>

在保持 Durable Object 存活的同时运行异步函数。函数运行前自动启动心跳，完成时停止（无论成功或抛出）。返回函数返回值。

这是使用 keepAlive 的推荐方式——可保证清理。

const result = await this.keepAliveWhile(async () => {
	const data = await longRunningComputation();
	return data;
});
const result = await this.keepAliveWhile(async () => {
	const data = await longRunningComputation();
	return data;
});

保持 Agent 存活

Durable Objects 在不活动一段时间后会被驱逐（通常 70–140 秒内无传入请求、WebSocket 消息或 alarm）。在长时间运行操作期间——流式 LLM 响应、等待外部 API、运行多步计算——agent 可能在执行中途被驱逐。

keepAlive() 通过持有内存心跳引用并直接使用 Durable Object alarm 系统来防止此情况。alarm 触发本身会重置不活动计时器。

  • 心跳不会与你自己的调度冲突，因为 alarm 系统通过单个 alarm 槽复用。
  • 不会创建调度行，心跳对 listSchedules() 不可见。
  • 多个并发 keepAlive() 调用使用引用计数，因此一个释放函数不会释放另一调用方的心跳。
  • 在子 Agent 内，keepAlive() 将心跳引用委托给顶层父 Agent，因为 facet 没有独立的 alarm 槽。

多个并发调用方

每次 keepAlive() 调用返回独立的释放函数：

const dispose1 = await this.keepAlive();
const dispose2 = await this.keepAlive();

// Both heartbeats are active
dispose1(); // Only cancels the first heartbeat
// Agent is still alive via dispose2's heartbeat

dispose2(); // Now the agent can go idle
const dispose1 = await this.keepAlive();
const dispose2 = await this.keepAlive();

// Both heartbeats are active
dispose1(); // Only cancels the first heartbeat
// Agent is still alive via dispose2's heartbeat

dispose2(); // Now the agent can go idle

AIChatAgent

AIChatAgent 在流式响应期间自动调用 keepAlive()。使用 AIChatAgent 时无需自行添加——每个 LLM 流默认受保护，不会因空闲驱逐。

何时使用 keepAlive

场景 使用 keepAlive?
通过 AIChatAgent 流式 LLM 响应 否 — 已内置
自定义 Agent 中的长时间计算 是
等待慢速外部 API 调用 是
多步 tool 执行 是
短 request-response handler 否 — 不需要
通过 scheduling 或 workflow 的后台工作 否 — alarm 已保持 DO 活跃

限制

  • 最大任务数： 受 SQLite 存储限制（每个任务一行）。每个 agent 的实际上限为数万条。
  • 任务大小： 每个任务（含 payload）最大 2MB。
  • 最小延迟： 0 秒（在下次 alarm tick 运行）
  • Cron 精度： 分钟级（非秒级）
  • Interval 精度： 秒级
  • Cron 作业： 执行后自动为下次 occurrence 重新调度
  • Interval 作业： 执行后重新调度为 now + intervalSeconds；若仍在运行则跳过

后续步骤

推送通知

使用 scheduling 与 web-push 发送 browser 推送通知。

这篇文档对您有帮助吗？