企业官网建设流程全解析

• 后端
• 人工智能
• AI Agent

Aegra 是一款开源的自托管 AI Agent 后端（LangGraph Platform 的开源替代），内置了Cron 定时任务调度器：你只需写一个标准的 Cron 表达式并指定时区，Aegra 就会自动按时触发你的 AI Agent 执行任务——每天晨报、周报摘要、定期数据巡检，都可以交给 Agent 自动完成，无需任何外部调度工具。

一、什么是 Aegra Cron 定时任务

Aegra 的定时任务由一个后台轮询调度器驱动：每个服务器实例内部都有一个轮询循环，定期检查 PostgreSQL 中到期的 Cron 任务，并为到期任务自动创建一次 Agent 运行（Run）。

核心特性一览：

• ⏰Cron 表达式调度：支持标准 5 字段表达式，也可开启秒级 6 字段表达式
• 🌍时区支持：传入 IANA 时区名（如Asia/Shanghai），任务按当地时间触发
• 🧵两种模式：无状态（每次新建会话）或绑定线程（持续同一对话上下文）
• 🗄️存储于 PostgreSQL：任务持久化，重启不丢失；多实例部署下通过租约（lease）机制避免重复触发
• 🎛️全生命周期管理：支持创建、更新、暂停、恢复、删除与搜索

调度器源码位于 libs/aegra-api/src/aegra_api/services/cron_scheduler.py，官方文档见 docs/guides/cron.mdx。

二、Cron 表达式怎么写：5 字段格式速查

Aegra 使用标准的 Cron 表达式语法，格式如下：

┌──── 分钟 (0-59) │ ┌──── 小时 (0-23) │ │ ┌──── 日期 (1-31) │ │ │ ┌──── 月份 (1-12) │ │ │ │ ┌──── 星期 (0-6，0=周日) * * * * *

常用 Cron 表达式对照表

💡 秒级调度（6 字段，如*/30 * * * * *表示每 30 秒）默认被拒绝，需将环境变量CRON_ALLOW_SECONDS_SCHEDULE设为true才允许开启，因为它会成倍增加调度压力和数据库写入。

三、三步创建你的第一个定时任务

以 Aegra 内置的 examples/cron_example/ 示例为例，这个示例 Agent 每分钟"打卡"一次，把运行计数写入会话历史。

第 1 步：启动 Aegra 服务（调度器默认开启）：

uv run aegra dev --no-db-check

第 2 步：创建线程和 Assistant（绑定线程模式）：

client = get_client(url="http://localhost:2026") thread = await client.threads.create() assistant = await client.assistants.create(graph_id="cron_example")

第 3 步：创建 Cron 任务：

run = await client.crons.create_for_thread( thread_id=thread["thread_id"], assistant_id=assistant["assistant_id"], schedule="* * * * *", # 每分钟触发 input={"messages": []}, ) print(f"Cron ID: {run['cron_id']}")

注意两点：

• input是必填项：每次触发都会基于这个载荷启动一次运行，即使 Agent 不需要输入也要传input={}，否则会被 422 拒绝。
• 创建成功后会立即执行第一次运行，响应返回的是Run对象；如果想先创建但暂不运行，传入enabled=False即可跳过首次触发。

完整可运行的示例脚本见 examples/cron_example/setup.py。

四、无状态 vs 线程绑定：两种触发方式

模式 1：无状态 Cron（默认）

每次触发都在全新线程中运行，适合独立、互不依赖的周期性任务（如每日早报）。默认会在运行结束后自动删除临时线程；如果想保留下来事后检查，传入on_run_completed="keep"：

run = await client.crons.create( assistant_id="agent", schedule="0 9 * * *", on_run_completed="keep", # 保留线程以便查看 input={"messages": [{"role": "user", "content": "Daily report"}]}, )

⚠️ 选择keep后线程会不断累积，长期部署建议配合线程 TTL 策略或单独的清理任务。

模式 2：线程绑定 Cron

把任务绑定到指定线程，每次触发延续同一会话，Agent 可以"记得"上次运行发生了什么——适合周报汇总、持续监控类场景。绑定线程删除时，关联的 Cron 会被级联自动删除。

五、时区设置：让任务按当地时间触发

默认情况下 Cron 表达式按UTC时间计算。想让任务按纽约或上海时间触发，只需传入一个 IANA 时区名：

run = await client.crons.create( assistant_id="agent", schedule="0 9 * * *", # 纽约时间 09:00 timezone="America/New_York", input={"messages": [{"role": "user", "content": "Morning standup"}]}, )

• ✅ 时区会随任务记录一并存储，内部自动转换为 UTC 调度
• ❌ 传入无效的时区名会返回 HTTP 422 错误
• 🔄 更新schedule时如果不改时区，原时区会被自动保留

六、管理定时任务：更新、暂停与搜索

更新任务参数

支持更新的字段包括：schedule、enabled、input、config、metadata、timezone、durability：

updated = await client.crons.update( cron_id="<cron_id>", schedule="0 10 * * *", # 改为每天 10 点 enabled=False, # 同时暂停 )

暂停与恢复

暂停不会删除任务，只是让调度器忽略它，随时可恢复：

await client.crons.update(cron_id="<cron_id>", enabled=False) # 暂停 await client.crons.update(cron_id="<cron_id>", enabled=True) # 恢复

搜索、统计与删除

crons = await client.crons.search(assistant_id="agent") # 按 Assistant 查 total = await client.crons.count(metadata={"team": "research"}) # 按元数据统计 await client.crons.delete(cron_id="<cron_id>") # 永久删除

API 层实现见 libs/aegra-api/src/aegra_api/api/crons.py，请求模型定义见 libs/aegra-api/src/aegra_api/models/crons.py。

七、高级配置：调度器环境变量

通过环境变量可精细调整调度行为，完整说明见 docs/reference/environment-variables.mdx：

超过单用户上限时，创建新任务会返回 HTTP 429。

八、注意事项与最佳实践

• 🔁至少一次（at-least-once）投递语义：若任务在"认领后、提交前"发生故障，租约到期后另一轮轮询会重新触发该任务。因此请让 Agent 的运行逻辑保持幂等，避免重复执行产生副作用。
• 🚫定时触发不走请求级鉴权：创建任务时会完整走鉴权链，但每次定时触发是以任务所有者身份在后台执行，不会再触发@auth.on(...)处理器。如果需要对每次运行做权限控制，请写在 Graph 内部。
• 🏷️metadata 不会复制到触发的 Run 上：metadata仅用于任务记录本身的搜索与过滤。
• 📅到期自动停止：设置end_time（必须为未来时间）后，任务在end_time之后会自动失效。

九、相关文件与文档索引

掌握 Cron 表达式、时区设置和两种线程模式后，你就可以让 Aegra 里的 AI Agent 在任何你需要的时间点自动上岗了——从每日晨报到每周巡检，全部自动化。

• 后端
• 人工智能
• AI Agent