企业官网建设流程全解析

1. 从 s13 到 s14：为什么 Agent 需要一个 Cron Scheduler

如果你跟着 learn-claude-code 的教程一路走到 s13，会发现 Agent 已经能后台跑任务了：模型调用 build、test、install 这类耗时命令时，不再阻塞主循环，而是丢到后台线程执行，再通过通知机制把结果带回。但 s13 有一个隐含前提——任务必须由用户或模型在当前这一轮对话里触发。换句话说，后台能力只是把「已经被触发的任务」放到后台执行，而不是主动在未来某个时间点触发任务。

这就是 s14 Cron Scheduler 要解决的问题。它要的不是「让工具跑得更快」，而是「让 Agent 可以按时间表生产新的工作」。每天早上 9 点自动检查 PR、每 30 分钟自动查看 CI 状态、每周自动生成一次报告——这些需求靠用户实时输入是做不到的，必须引入一个新的触发源：时间。

s14 的做法很克制：它没有推翻前面的主流程，而是在已有 Agent Loop 外侧加了一个独立的调度层。当时间匹配某个 cron 表达式时，调度器把对应任务放入队列；当 Agent 空闲时，队列处理器再把这个任务交给正常的 agent_loop 执行。这样一来，Agent 的行为不再只能由用户实时输入触发，也可以由预先注册的时间表触发。

这篇笔记聚焦 Claude Code 并发场景下 s14_new Cron Scheduler 的配置落地：从 settings.json 骨架到任务触发链路，给出可复制的配置片段与验证动作，帮你在本地跑通定时调度并观察并发行为。适合已经跑过 s01 到 s13、想补齐 Concurrency 这一部分内容的读者。

2. 前置准备：TaoToken 接入与本地环境

在动手改配置之前，先把模型接入这一层准备好。s14 的调度链路本身不依赖特定模型服务，但你要验证「定时任务触发后模型能正常推理并调用工具」，就需要一个稳定的 API 入口。我这边用的是 TaoToken，它的接口兼容主流 SDK 调用方式，配置成本低。

官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数，直接写 https://taotoken.net/api 即可。

你需要先拿到 API Key。进入控制台创建密钥：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，然后在 API Keys 页面生成：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。生成后把 Key 写进环境变量，不要硬编码进代码。

export TAOTOKEN_API_KEY="sk-你的密钥" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

本地环境方面，确认 Python 版本在 3.10 以上，因为 s14 用到了str | None这种联合类型注解。另外确认工作目录可写，因为 durable 任务会写入.scheduled_tasks.json。

python --version # Python 3.10.x 或更高 ls -la # 确认当前目录有写权限

如果你还没跑过前面的章节，建议先跑通 s13 的后台任务，再进入 s14。因为 s14 的 queue processor 和 s13 的后台线程在并发模型上是叠加关系，跳过 s13 直接看 s14 容易在锁的层面卡住。

3. settings.json 骨架与 CronJob 数据结构

s14 的配置骨架分两部分：一部分是运行时的数据结构定义，一部分是持久化文件。先看数据结构。

from dataclasses import dataclass, asdict from pathlib import Path import threading WORKDIR = Path.cwd() DURABLE_PATH = WORKDIR / ".scheduled_tasks.json" @dataclass class CronJob: id: str cron: str # "0 9 * * *" prompt: str # message to inject when fired recurring: bool # True = recurring, False = one-shot durable: bool # True = persist to disk scheduled_jobs: dict[str, CronJob] = {} cron_queue: list[CronJob] = [] cron_lock = threading.Lock() agent_lock = threading.Lock() _last_fired: dict[str, str] = {} # job_id -> "YYYY-MM-DD HH:MM"

这里最关键的是cron和prompt两个字段。cron决定任务什么时候触发，比如0 9 * * 1-5表示工作日早上 9 点；prompt则是任务触发后要重新递进 Agent Loop 的用户请求。

scheduled_jobs保存所有已注册的定时任务，cron_queue保存已经到时间、但还没交给 Agent 执行的任务。这里特意引入了两个锁：cron_lock用来保护调度任务和队列，agent_lock用来保证同一时间只有一个 Agent Loop 在运行，避免用户输入和定时任务同时抢占同一个会话。

_last_fired是一个容易被忽略的细节。因为调度线程是每秒检查一次，如果某个任务在 09:00 这一分钟内一直匹配 cron 表达式，就可能被重复触发几十次。所以代码用YYYY-MM-DD HH:MM作为分钟级标记，保证同一个任务在同一分钟只触发一次。

持久化文件.scheduled_tasks.json的结构就是 CronJob 的序列化结果：

[ { "id": "cron_960034", "cron": "*/2 * * * *", "prompt": "Print the current date. Run: date", "recurring": true, "durable": true } ]

如果你希望任务在进程重启后还能恢复，就把durable设为true；如果只是当前会话临时用，设为false即可，进程结束就消失。

4. 可复制配置：cron 匹配、校验与调度线程

这一节给出可以直接复制运行的配置片段。先看 cron 字段匹配逻辑。

def _cron_field_matches(field: str, value: int) -> bool: """Match a single cron field against a value.""" if field == "*": return True if field.startswith("*/"): step = int(field[2:]) return step > 0 and value % step == 0 if "," in field: return any(_cron_field_matches(f.strip(), value) for f in field.split(",")) if "-" in field: lo, hi = field.split("-", 1) return int(lo) <= value <= int(hi) return value == int(field)

这一段负责判断单个字段是否匹配当前时间值。*表示任意值都匹配，*/5表示每 5 个单位触发一次，1-5表示一个范围，1,3,5表示多个离散值。

在此基础上，cron_matches()会把五个字段组合起来，判断一个完整的 cron 表达式是否命中当前时间。

from datetime import datetime def cron_matches(cron_expr: str, dt: datetime) -> bool: """Check if a 5-field cron expression matches the given datetime. Standard cron semantics: DOM and DOW use OR when both are constrained.""" fields = cron_expr.strip().split() if len(fields) != 5: return False minute, hour, dom, month, dow = fields dow_val = (dt.weekday() + 1) % 7 # Python Monday=0 -> cron Sunday=0 m = _cron_field_matches(minute, dt.minute) h = _cron_field_matches(hour, dt.hour) dom_ok = _cron_field_matches(dom, dt.day) month_ok = _cron_field_matches(month, dt.month) dow_ok = _cron_field_matches(dow, dow_val) if not (m and h and month_ok): return False dom_unconstrained = dom == "*" dow_unconstrained = dow == "*" if dom_unconstrained and dow_unconstrained: return True if dom_unconstrained: return dow_ok if dow_unconstrained: return dom_ok return dom_ok or dow_ok

这里有一个容易踩的坑：Python 的weekday()中周一是 0，而 cron 语义中通常是周日是 0。所以代码用dow_val = (dt.weekday() + 1) % 7做转换。另外，day-of-month 和 day-of-week 在标准 cron 语义中，当两者都被约束时通常采用 OR 逻辑，代码最后的return dom_ok or dow_ok就是在模拟这个行为。

接下来是校验逻辑。cron 任务不是一次性的普通工具调用，而是会在未来自动触发，如果表达式本身是错的，错误会被延迟到未来才发生，排查成本更高。所以 s14 在注册任务时就先检查字段数量、数值范围、步长、区间等内容。

def validate_cron(cron_expr: str) -> str | None: """Validate a cron expression. Returns error message or None.""" fields = cron_expr.strip().split() if len(fields) != 5: return f"Expected 5 fields, got {len(fields)}" bounds = [(0, 59), (0, 23), (1, 31), (1, 12), (0, 6)] names = ["minute", "hour", "day-of-month", "month", "day-of-week"] for i, (field, (lo, hi), name) in enumerate(zip(fields, bounds, names)): err = _validate_cron_field(field, lo, hi) if err: return f"{name}: {err}" return None

然后是调度线程本体。这是 s14 最核心的新增逻辑。

import time def cron_scheduler_loop(): """Independent daemon thread: poll every 1s, fire matching jobs.""" while True: time.sleep(1) now = datetime.now() minute_marker = now.strftime("%Y-%m-%d %H:%M") with cron_lock: for job in list(scheduled_jobs.values()): try: if cron_matches(job.cron, now): if _last_fired.get(job.id) != minute_marker: cron_queue.append(job) _last_fired[job.id] = minute_marker print(f"[cron fire] {job.id} -> {job.prompt[:40]}") if not job.recurring: scheduled_jobs.pop(job.id, None) if job.durable: save_durable_jobs() except Exception as e: print(f"[cron error] {job.id}: {e}")

这一段可以看作 s14 的「时钟」。它每秒醒来一次，拿当前时间和所有scheduled_jobs做匹配。如果某个任务命中，就把它追加到cron_queue。注意调度线程不直接调用 LLM，也不直接执行工具，只是把「到期任务」放入队列。这样调度系统和 Agent Loop 就解耦了。

对于一次性任务，任务会在触发后删除：

if not job.recurring: scheduled_jobs.pop(job.id, None) if job.durable: save_durable_jobs()

所以 recurring 任务会一直保留，下一次命中时间继续触发；one-shot 任务触发一次就自动消失。

程序启动后，会先加载持久化任务，然后启动调度线程：

load_durable_jobs() threading.Thread(target=cron_scheduler_loop, daemon=True).start() print("[cron] scheduler thread started")

这里的daemon=True表示这是一个后台守护线程。主进程还在，它就继续跑；主进程退出，它也会一起结束。

5. 验证请求：从注册到触发的完整链路

配置写好后，怎么验证它真的跑通了？我试过用下面这个 prompt 来观察整个链路：

Schedule a task to print the current date every 2 minutes

模型不会直接执行date命令，而是先把这个需求转换成一次schedule_cron工具调用：

ToolUseBlock( name="schedule_cron", input={ "cron": "*/2 * * * *", "prompt": "Print the current date. Run: date", "recurring": True } )

随后schedule_job()会创建一个新的 CronJob 对象，并写入scheduled_jobs。调试中可以看到任务被注册后返回了：

Scheduled cron_960034: */2 * * * * -> Print the current date. Run: date

这说明第一次循环完成的是「创建定时任务」，而不是「执行定时任务」。真正的执行要等cron_scheduler_loop后续命中时间后再触发。

当时间命中 cron 表达式时，系统打印出：

[cron fire] cron_960034 -> Print the current date. Run: date

紧接着出现：

[queue processor] delivering scheduled work [inject cron] Print the current date. Run: date

这说明队列处理线程接管了这个到期任务，并把它作为[Scheduled] ...消息注入到 Agent Loop 中。之后模型像处理普通用户请求一样生成 bash 工具调用，执行date，最终返回当前系统时间。

进入agent_loop()后，代码首先执行：

fired = consume_cron_queue() for job in fired: messages.append({ "role": "user", "content": f"[Scheduled] {job.prompt}" })

调试中可以看到messages[-1]变成了：

{ "role": "user", "content": "[Scheduled] Print the current date. Run: date" }

这一步非常关键。它说明 Cron Scheduler 并不是绕过 Agent Loop 去执行命令，而是把「到期任务」重新包装成一条普通的 user message。只不过这条消息不是用户现场输入的，而是由调度线程在命中时间后注入进来的。

模型看到[Scheduled] Print the current date. Run: date后，开始正常推理，并生成 bash 工具调用：

ToolUseBlock( name="bash", input={"command": "date"} )

工具执行后，results 中出现了标准 tool_result：

{ "type": "tool_result", "tool_use_id": "call_00_LpyfpqRtU2bWKhZfBWsI6131", "content": "Fri Jun 5 08:54:19 AM CST 2026" }

这说明date命令已经真实执行，并把当前系统时间返回给 Agent Loop。从用户角度看，这个任务已经在指定的 2 分钟周期内自动运行了一次；从代码角度看，它本质上仍然是普通的工具调用结果回填。

由于这个任务的 cron 表达式是*/2 * * * *，并且recurring=True，所以它不会在第一次触发后被删除，而是会继续保留在scheduled_jobs中，等待下一个匹配的分钟再次触发。第二次触发时，新的 fired 仍然是同一个任务，再次被注入为[Scheduled] ...消息，模型再次执行date，返回新的时间。

这证明了 recurring 定时任务的周期性触发逻辑是有效的。更重要的是，调试结果也证明了_last_fired的作用：同一个任务不会在同一分钟里重复疯狂触发，而是在下一个匹配分钟才再次进入cron_queue。

6. 本篇常见错排查

6.1 任务注册成功但从不触发

最常见的原因是 cron 表达式字段数不对。s14 只接受标准五段式：分 时 日 月 周。如果你写了六段（带秒），validate_cron会直接返回Expected 5 fields, got 6。检查你注册时返回的消息，如果以Error:开头，说明校验没过。

另一个原因是时区。datetime.now()取的是本地时间，如果你的机器时区和预期不符，任务会在错误的时间触发。验证方法是在调度线程里临时打印now，确认它和你预期的时间一致。

6.2 同一分钟内任务被触发多次

如果你去掉了_last_fired的判断，或者minute_marker的格式写错，就会出现这个问题。minute_marker必须是YYYY-MM-DD HH:MM这种分钟级精度，不能带秒。带秒的话每次循环 marker 都不同，判断就失效了。

6.3 用户输入和定时任务同时执行导致会话错乱

这是并发场景下最容易踩的坑。s14 用agent_lock保证同一时间只有一个 Agent Loop 在跑。如果你在main里处理用户输入时没有加锁：

with agent_lock: run_agent_turn_locked(query)

那么用户请求还没跑完，queue processor 又启动了一轮 Agent Loop，session_history和session_context就会被并发修改。表现是消息顺序错乱、上下文丢失、甚至报 KeyError。

6.4 durable 任务取消后重启又回来了

取消 durable 任务时，除了从scheduled_jobs里 pop，还必须同步更新磁盘文件：

def cancel_job(job_id: str) -> str: with cron_lock: job = scheduled_jobs.pop(job_id, None) if not job: return f"Job {job_id} not found" if job.durable: save_durable_jobs() return f"Cancelled {job_id}"

如果漏了save_durable_jobs()，内存里取消了，但.scheduled_tasks.json里还在，下次启动load_durable_jobs()又会把它加载回来。

6.5 调度线程被单个坏任务拖死

cron_scheduler_loop里对每个 job 的匹配都包了 try/except。如果你自己改代码时去掉了这个 try，某个 job 的 cron 表达式解析抛异常，整个调度线程就会退出，之后所有任务都不再触发。表现是「第一个任务正常，后面全部静默」。排查方法是看有没有[cron error]输出。

7. 语义一致的 CTA 与后续学习路径

如果你在排障过程中发现是模型接入层的问题，比如请求超时、鉴权失败、返回格式异常，优先去检查 API Key 和接入配置。API Keys 管理页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ，接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。这两个页面能覆盖大部分接入类报错。

如果你想先验证模型本身在定时任务场景下的推理表现，比如让它把自然语言调度需求转成 cron 表达式，可以直接在模型对话页面测试：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。输入类似「每个工作日早上 9 点检查 PR」这样的需求，看它生成的 cron 表达式是否符合预期，再决定要不要写进schedule_cron。

如果你打算把 s14 的调度能力用到长期编码或 Agent 工作流里，比如让 Agent 每天自动跑测试、定期检查依赖更新，那更适合用 Coding Plan 来管理调用配额和并发：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。定时任务的特点是低频但长期，配额规划比单次调用更重要。

回到 s14 本身，这一节最值得记住的工程思想是：调度和执行必须解耦。调度器只负责判断「什么时候该做」，不负责「具体怎么做」；队列只负责缓存已触发任务，不负责模型推理；Agent Loop 只负责消费任务，不负责轮询时间。正是因为这几层职责清晰分开，s14 才能在不破坏原有 Agent Loop 的情况下，为系统增加按时间自动行动的能力。

同时也要清楚 durable 和真正系统级调度的区别。.scheduled_tasks.json只能保证任务定义跨重启保留，但不能保证进程关闭期间任务仍然执行。教学版的 cron scheduler 是进程内调度器，它适合解释 Agent 内部如何处理定时任务；如果要做生产环境中的长期调度，还需要结合系统级定时器、服务守护、锁机制和多实例协调。

下一篇会继续看 s19 MCP Plugin 章节，把插件扩展这一块补上。