Background Tasks - 后台任务与异步执行

问题背景：慢命令阻塞#

同步执行慢命令会导致：

错误理解：后台就是另一条主循环。

正确理解：主循环保持单线程，后台是另一条执行线。

主循环流程变更#

流程说明：

• 主循环：background_run → 登记任务 → 立刻返回 task_id → 继别工作
• 后台执行线：真正执行命令 → 写入通知 → 写磁盘文件
• 下一轮：drain_notifications → 摘要注入 → 模型收到结果

验证你的理解#

Q1：为什么”后台是另一条主循环”是错误理解？

:::details 答案 主循环必须保持单线程，否则状态管理混乱。

如果多条主循环：

1. 多个 messages[] → 状态不一致
2. 多个 tool_use → 权限检查混乱
3. 多个 stop_reason → 循环控制失控

正确设计：主循环单线程，后台是另一条执行线。并行的是等待时间，不是主循环本身。 :::

RuntimeTaskRecord：执行状态管理#

1
task = {
2
    "id": "a1b2c3d4",
3
    "command": "pytest",
4
    "status": "running",  # running/completed/failed/timeout
5
    "started_at": 1710000000.0,
6
    "result_preview": "",
7
    "output_file": "",
8
}

职责：管执行状态（谁在跑、跑到哪）。

Notification：结果投递通道#

1
notification = {
2
    "type": "background_completed",
3
    "task_id": "a1b2c3d4",
4
    "status": "completed",
5
    "preview": "tests passed",  # 只放摘要，不放全文
6
}

职责：管结果投递（完成了、通知主循环）。

职责分离原则#

为什么分离？

验证你的理解#

Q2：为什么 notifications 只放 preview，不放完整输出？

:::details 答案 防止上下文撑爆 + 防止注意力稀释。

如果通知放完整输出：

1. pytest 可能输出数千行 → messages[] 撑爆
2. 模型注意力被稀释 → 核心任务权重下降
3. 多个后台任务完成 → 上下文爆炸

正确设计：通知只放 preview（500字摘要），完整输出写磁盘文件。模型需要全文时用 read_file 读取。 :::

drain_notifications 流程#

流程：

1. drain_notifications 排空通知队列
2. 把通知摘要 append 到 messages (user role)
3. 再调用模型
4. 模型收到后台结果

摘要注入位置#

1
# 主循环入口
2
def agent_loop():
3
    # 1. 先 drain_notifications 排空通知队列
4
    notifications = drain_notifications()
5
    for notif in notifications:
6
        # 2. 把通知摘要 append 到 messages (user role)
7
        messages.append({
8
            "role": "user",
9
            "content": f"Background task {notif['task_id']} completed: {notif['preview']}"
10
        })
11

12
    # 3. 再调用模型
13
    response = call_llm(messages)

关键点：通知注入在调用模型前——模型本轮能看到后台结果。

验证你的理解#

Q3：为什么通知注入在调用模型前，而不是调用后？

:::details 答案 模型本轮能看到后台结果，不需要等到下一轮。

如果注入在调用模型后：

1. 本轮模型看不到后台结果
2. 用户问”后台任务跑完了吗” → 模型不知道
3. 需要下一轮才能看到 → 响应延迟

正确设计：drain 在调用模型前 → 模型本轮能看到结果 → 实时响应。 :::

background_run：发起后台任务#

1
def background_run(command: str) -> str:
2
    # 1. 登记 RuntimeTaskRecord
3
    task_id = generate_task_id()
4
    task = {
5
        "id": task_id,
6
        "command": command,
7
        "status": "running",
8
        "started_at": time.time(),
9
    }
10
    tasks[task_id] = task
11

12
    # 2. 写磁盘持久化
13
    save_task_to_disk(task)
14

15
    # 3. 启动后台线程执行
16
    start_background_thread(command, task_id)
17

18
    # 4. 立刻返回 task_id
19
    return task_id

关键设计：立刻返回 task_id，不是同步卡住。

background_check：主动查询状态#

1
def background_check(task_id: str) -> dict:
2
    task = tasks.get(task_id)
3
    if not task:
4
        return {"error": "Task not found"}
5

6
    return {
7
        "status": task["status"],
8
        "preview": task["result_preview"],
9
        "output_file": task["output_file"],
10
    }

用途：模型主动查询任务状态（notification 已 drain 或未 drain）。

验证你的理解#

Q4：background_run 为什么立刻返回 task_id，而不是等待执行完成？

:::details 答案 主循环不阻塞，继续推进别的工作。

如果等待执行完成：

1. pytest 可能跑10分钟 → 主循环阻塞10分钟
2. 模型在等待期间什么都做不了
3. 用户被整轮流程堵住

正确设计：background_run 立刻返回 task_id → 主循环继续 → 后台独立执行 → 下一轮看到结果。 :::

与后续课程的关系#

核心洞察：Background Tasks 是”等待不阻塞”的机制。理解 L13，就理解了 Agent 如何”主循环单线程 + 后台并行等待”。后续课程都在扩展后台机制——定时任务、任务系统。

L13 核心概念的理解升级路径#

如何检验你达到了 L2？#

用这个问题自测：

如果通知队列放完整输出（数千行 pytest 日志），会发生什么？完整链条是什么？

:::details L2 层级回答 上下文撑爆 + 注意力稀释 + 多任务累积爆炸。

完整链条：

1. pytest 输出数千行日志
2. notification 包含完整输出
3. drain 注入 messages → messages[] 撑爆（数万 tokens）
4. Transformer 注意力权重归一化 → pytest 日志权重累加
5. 核心任务权重被稀释 → 输出可能跑偏
6. 多个后台任务完成 → 通知队列累积 → messages[] 更大
7. 触发 Context Compact → 摘要生成 → 丢失推理细节

正确设计：通知只放 preview（500字摘要），完整输出写磁盘文件。模型需要全文时 read_file 读取。 :::

本章目标：让每个核心概念都从 L0 升级到 L2。理解 L13，不只是记住任务表和通知队列，而是掌握”如何在单线程主循环上并行等待”的设计原则。

主循环与后台执行线的关系#

时间线与并行关系#

数据流向#

前台 vs 后台对比#

Task vs Background Task 边界#