⚙️ 从零打造 AI Agent：任务运行时篇（S12-S14）

基础打好了，系统也加固了，现在要让它能管理复杂项目、异步执行任务、定时触发工作。

---

🎬 开场：你的 Agent 只会单打独斗吗？

现在你的 Agent 已经：

• ✅ 有了核心循环，会把结果喂回模型
• ✅ 会用工具，能读写文件、执行命令
• ✅ 有了安全防护，不会乱删东西
• ✅ 有了记忆，跨会话也能记住一些事
• ✅ 有了系统加固，更安全、更稳定

但它还是… 一个人干活。

想象你要盖一栋大楼：

• 只雇一个工人 → 他要搬砖，要砌墙，要刷漆，要装水管…
• 项目大了，根本忙不过来
• 而且他不知道"谁在等谁"，经常返工

你的 Agent 也面临同样的问题。

---

📚 上集回顾：系统加固

上篇文章我们讲了给 Agent 加固：

章节	核心问题	解决方案
S07 权限系统	如何防止危险操作	deny → mode → allow → ask
S08 Hook 系统	如何扩展而不改主循环	事件 + payload + 返回码
S09 记忆系统	如何跨会话记住信息	user/feedback/project/reference
S10 系统提示词	如何组织输入	core + tools + skills + memory + dynamic
S11 错误恢复	如何遇错不崩	分类 → 恢复 → 预算

今天，我们给它加上任务管理、异步执行、定时触发的能力。

---

📋 S12：任务系统 - 持久化工作图

🎬 精彩开场：项目管理软件

想象你要管理一个大型项目，有 100 个任务：

没有任务系统的情况：

• 任务清单写在脑子里
• “张三做任务 1，李四做任务 2…”
• 任务 3 依赖任务 1，任务 4 依赖任务 2…
• 问：“任务 10 现在谁在做？”
• 答：“呃… 让我想想…”

有任务系统的情况：

任务1: 写需求文档 [张三][进行中][解锁任务3]
任务2: 设计架构 [李四][已完成][解锁任务4]
任务3: 写代码 [待认领][被任务1阻塞]
任务4: 写测试 [待认领][被任务2阻塞]

• 任何时候都能看到：谁在做什么、被什么卡住
• 自动追踪依赖关系
• 多人可以协作

这就是任务系统：把"工作"变成一张可观察、可管理的图。

核心问题

复杂项目有很多任务，它们之间有依赖关系：

• 任务 A 完成后，任务 B 才能开始
• 任务 C 和 D 可以并行
• 任务 E 完成后，才能做任务 F

用简单的 todo 列表管不住。

解决方案

用任务图管理依赖关系：

任务1: 写需求文档 [张三][进行中][解锁任务3]
任务2: 设计架构 [李四][已完成][解锁任务4]
任务3: 写代码 [待认领][被任务1阻塞]

关键数据结构

TaskRecord = {
    "id": 1,
    "subject": "写解析器",
    "status": "pending",  # pending/in_progress/completed
    "blockedBy": [],      # 还在等谁
    "blocks": [],       # 完成后解锁谁
    "owner": "",        # 谁在做
}

就绪判断规则

def is_ready(task: dict) -> bool:
    return task["status"] == "pending" and not task["blockedBy"]

只有同时满足：

1. 任务还没开始
2. 前置依赖全部完成

才算"可以开始"。

工作流程

1. 创建任务
2. 设置依赖关系
3. 系统自动追踪"谁在等谁"
4. 完成任务时自动解锁后续任务

核心代码

# 创建任务
task = create_task("写解析器", blocked_by=[task1.id])

# 自动解锁
def complete(task_id: int):
    task["status"] = "completed"
    
    # 自动解锁被它阻塞的任务
    for other in all_tasks:
        if task_id in other["blockedBy"]:
            other["blockedBy"].remove(task_id)

新手最容易踩的4个坑

1. 只会创建任务，不会维护依赖 → 最后还是一张普通清单
2. 任务只放内存，不落盘 → 系统重启就丢了
3. 完成任务后不自动解锁 → 系统永远不知道下一步谁可以开工
4. 把工作目标和运行中的执行混成一层 → Task 是目标，Runtime Task 是执行状态

---

🔄 S13：后台任务 - 把任务目标和运行槽位分开

🎬 精彩开场：点外卖 vs 在餐厅等

在餐厅等的情况（同步）：

• 你坐下来点菜
• 服务员说"需要等 30 分钟"
• 你就一直坐着等
• 等了 29 分钟，服务员说"还要 5 分钟"
• 你继续等… 崩溃

点外卖的情况（异步）：

• 你下单点外卖
• 去做其他事
• 30 分钟后，外卖到了通知你

后台任务就是"点外卖"模式：慢操作在后台跑，主循环继续做其他事情，完了再通知。

核心问题

有些操作很慢：

• npm install 可能要 10 分钟
• pytest 可能要 5 分钟
• 构建 Docker 镜像可能要 30 分钟

如果主循环一直同步等待，用户什么都做不了。

解决方案

慢命令在后台跑，主循环继续做其他事，完了再通知。

工作流程

主循环
    |
    +-- background_run("pytest")
    |   -> 立刻返回 task_id
    |
    +-- 继续别的工作
    |
    +-- 下一轮前排空通知
        -> 把摘要注入 messages

后台执行线
    |
    +-- 真正执行 pytest
    +-- 完成后写入通知队列

关键数据结构

# 后台任务记录
RuntimeTaskRecord = {
    "id": "a1b2c3d4",
    "command": "pytest",
    "status": "running",
    "preview": "",
}

# 通知
Notification = {
    "type": "background_completed",
    "task_id": "a1b2c3d4",
    "status": "completed",
    "preview": "3 passed",
}

为什么完整输出不要塞进 prompt

# 错误！上下文爆炸
messages.append({"role": "user", "content": full_log_output})

# 正确！只放摘要
messages.append({
    "role": "user",
    "content": "[bg:a1b2c3] completed - 3 passed"
})

核心代码

# 启动后台任务
def run(command: str) -> str:
    task_id = new_id()
    thread = threading.Thread(target=_execute, args=(task_id, command))
    thread.start()
    return task_id  # 立刻返回，主循环继续

# 完成时写通知
def _execute(task_id: str, command: str):
    result = subprocess.run(...)
    notifications.append({
        "task_id": task_id,
        "status": "completed",
        "preview": result.stdout[:500]
    })

Task vs Background Task

机制	回答什么问题
Task	要做什么、谁依赖谁、现在总体进度如何
Background Task	哪个命令正在跑、跑到什么状态、结果什么时候回来

新手最容易踩的3个坑

1. 以为"后台"就是更复杂的主循环 → 主循环永远只有一条
2. 长日志全文塞进上下文 → 上下文会爆炸
3. 没有通知队列 → 主循环不知道结果回来了

---

⏰ S14：定时调度 - 让时间也能触发工作

🎬 精彩开场：闹钟的故事

想象你每天早上起床：

没有定时调度：

• 你妈妈每天早上 7 点敲门叫你
• 如果妈妈不在，你就睡过头了
• 你完全依赖"有人提醒"

有闹钟的情况：

• 你设置闹钟：明天早上 7 点
• 闹钟响了，自动叫醒你
• 不需要妈妈，也不需要任何人提醒

定时调度就是给 AI 装一个"闹钟"——到了指定时间，自动触发新工作，不需要用户每次手动提醒。

核心问题

有些任务不是"现在做"，而是"将来某个时间做"：

• 每天早上跑一次测试
• 每周一生成报告
• 30 分钟后提醒我检查结果

如果每次都要手动触发，太累了。

解决方案

把"时间"也变成一种触发入口。

三部分架构

1. 调度记录（记住未来）
2. 定时检查器（看时间是否到了）
3. 通知队列（结果通知主循环）

Cron 表达式

分 时 日 月 周
*/5 * * * *   → 每 5 分钟
0 9 * * 1      → 每周一 9 点
30 14 * * *   → 每天 14:30

工作流程

schedule_create("0 9 * * 1", "生成周报")
    ↓
把记录写到文件（持久化）
    ↓
后台检查器每分钟看一次
    ↓
时间到了？
    ↓ 是 → 发通知到队列
    ↓ 否 → 继续等
    ↓
主循环下一轮处理通知

核心代码

# 创建调度
def create(cron_expr: str, prompt: str):
    job = {
        "id": new_id(),
        "cron": cron_expr,
        "prompt": prompt,
        "last_fired_at": None,
    }
    jobs.append(job)

# 定时检查
def check_loop():
    while True:
        now = datetime.now()
        for job in jobs:
            if cron_matches(job["cron"], now):
                queue.put({
                    "type": "scheduled_prompt",
                    "schedule_id": job["id"],
                    "prompt": job["prompt"],
                })
                job["last_fired_at"] = now.timestamp()
        time.sleep(60)

后台任务 vs 定时调度

机制	回答什么问题
后台任务	“已经启动的慢操作，结果什么时候回来？”
定时调度	“一件事应该在未来什么时候开始？”

新手最容易踩的5个坑

1. 一上来沉迷 cron 语法细节 → 重点是"触发后怎么回到主循环"
2. 没有 last_fired_at → 容易短时间重复触发
3. 只放内存，不落盘 → 程序重启就没了
4. 把触发结果直接在后台默默执行 → 先发通知，让主循环决定
5. 误以为定时任务必须绝对准点 → 更重要的是"有计划地触发"

---

🎯 核心公式总结

任务运行时 = 任务系统 + 后台任务 + 定时调度

任务系统 = 任务记录 + 依赖关系 + 自动解锁 + 持久化
后台任务 = 后台执行 + 任务表 + 通知队列 + 摘要返回
定时调度 = 调度记录 + 定时检查 + 通知队列 + 回到主循环

它们之间的关系

任务系统（S12）
    ↓
描述"要做什么"（工作目标）
    ↓
后台任务（S13）
    ↓
让"正在做的"异步执行（运行状态）
    ↓
定时调度（S14）
    ↓
让"未来要做的"定时触发（时间触发）

---

🎯 一句话带走

记住这句话就够了：任务系统管理"做什么"，后台任务管理"怎么做"（异步），定时调度管理"什么时候做"（时间触发）。三者配合，让 Agent 从"单打独斗"变成"能管理复杂项目的团队"。

---

▶️ 下一步

下一篇文章我们将进入多 Agent 平台部分，学习如何建立多个 Agent 协作、制定团队协议、让 Agent 自主工作。

相关文章：

• 《从零打造 AI Agent：核心闭环篇（S01-S06）》
• 《从零打造 AI Agent：系统加固篇（S07-S11）》
• 《从零打造 AI Agent：多 Agent 平台篇（S15-S19）》

---

关注我，一起探索 AI Agent 的世界！