s13：后台任务（Background Tasks）

核心思想

基于线程的后台任务执行系统——核心目标是让耗时操作异步执行，避免阻塞 Agent 的主循环，并通过通知机制将结果异步注入对话流。

关键区别：s13 的后台执行 ≠ 自动触发（那是 s14 的事）。s13 解决的是「命令太慢会卡住主循环」的问题。

---

1. 后台任务生命周期

工具调用请求
    │
    ▼
should_run_background() ──否──→ 同步执行（原有路径）
    │
    是
    ▼
start_background_task()
    │
    ├─→ 分配 bg_id（bg_0001, bg_0002...）
    ├─→ 创建守护线程（daemon=True）
    └─→ 立即返回占位符（不等待）
    │
    ▼
主循环继续（处理其他工具调用或下一轮 LLM）
    │
    ▼
下一轮循环开始时
    │
    ▼
collect_background_results()
    │
    ├─→ 检查已完成的后台任务
    ├─→ 生成  通知
    └─→ 注入到 user 消息中

---

2. 后台任务存储结构

# 全局递增计数器，生成唯一的后台任务 ID
_bg_counter = 0

# bg_id → {tool_use_id, command, status}
background_tasks: dict[str, dict] = {}

# bg_id → output（执行结果字符串）
background_results: dict[str, str] = {}

# 线程锁，保护共享字典的读写操作
background_lock = threading.Lock()

四个核心数据结构的作用：

变量	类型	作用
_bg_counter	int	全局递增计数器，保证每个 bg_id 唯一
background_tasks	dict[str, dict]	任务元信息：记录每个后台任务的 ID、对应工具调用 ID、命令内容、运行状态
background_results	dict[str, str]	任务结果：bg_id → 执行输出的字符串
background_lock	threading.Lock	线程锁：保护上面两个字典的并发读写安全

---

3. 慢操作检测 — is_slow_operation()

判断逻辑

教学内容做了简化，只对 bash 工具进行检测（其他工具通常快速返回）。具体步骤：

1. 提取命令字符串并转为小写
2. 检查是否包含已知的慢操作关键词

慢操作关键词分类

类别	关键词	说明
包管理器	pip install, npm install, cargo build	安装依赖通常需要下载+编译
构建工具	build, compile, make	编译项目耗时较长
测试框架	test, pytest	测试套件可能跑很久
部署操作	deploy, docker build	镜像构建和部署
通用长时	install	兜底匹配

代码实现

def is_slow_operation(tool_name: str, tool_input: dict) -> bool:
    """判断一个工具调用是否属于慢操作"""
    if tool_name != "bash":
        return False  # 只检测 bash 工具
    cmd = tool_input.get("command", "").lower()
    slow_keywords = [
        "install", "build", "test", "deploy", "compile",
        "docker build", "pip install", "npm install",
        "cargo build", "pytest", "make"
    ]
    return any(kw in cmd for kw in slow_keywords)

---

4. 触发决策 — should_run_background()

一条命令是否进入后台执行，采用两级判断策略：

1. 显式优先：模型在工具调用参数中传了 run_in_background=True → 直接后台
2. 启发式兜底：模型未指定时，通过关键词自动判断（命令含 install/build/test 等）→ 后台

def should_run_background(tool_name: str, tool_input: dict) -> bool:
    # 第一级：模型显式指定 run_in_background: true
    if tool_input.get("run_in_background"):
        return True
    # 第二级：模型未指定时，通过关键词检测自动判断
    return is_slow_operation(tool_name, tool_input)

设计原则：显式请求永远优先于启发式规则。模型"知道"该后台时就让它自己决定；模型"不知道"时系统兜底自动判断。

---

5. 后台任务启动 — start_background_task()

核心公式

后台执行 = 守护线程（Daemon Thread）+ 生命周期字典

把工具执行包进一个 worker 函数，丢进 daemon=True 的线程里，返回一个 bg_id 给调用方。

什么是守护线程？

守护线程（Daemon Thread）是一种特殊线程，其生命周期与主线程绑定——主线程结束时，所有守护线程自动跟着退出，不会阻止进程关闭。这非常适合辅助任务：可以随时中断、不需要优雅退出。

代码实现

def start_background_task(block) -> str:
    """启动一个后台任务，返回 bg_id"""
    global _bg_counter
    _bg_counter += 1
    bg_id = f"bg_{_bg_counter:04d}"          # 格式：bg_0001, bg_0002, ...
    cmd = block.input.get("command", block.name)

    # 闭包：捕获 block 和 bg_id，在线程内执行实际工具调用
    def worker():
        result = execute_tool(block)           # 实际执行工具
        with background_lock:                  # 加锁写结果
            background_tasks[bg_id]["status"] = "completed"
            background_results[bg_id] = result

    # 先注册任务状态（running），再启动线程
    with background_lock:
        background_tasks[bg_id] = {
            "tool_use_id": block.id,
            "command": cmd,
            "status": "running",
        }

    thread = threading.Thread(target=worker, daemon=True)
    thread.start()
    print(f"  \033[33m[background] dispatched {bg_id}: {cmd[:40]}\033[0m")
    return bg_id

执行顺序要点：

1. 先分配 bg_id
2. 先把状态写入字典（status: "running"）
3. 再启动线程
4. 立即返回 bg_id（不等待线程完成）

---

6. 结果收集 — collect_background_results()

收集机制

每轮循环结束时扫描已完成的任务，把结果格式化成 <task_notification> 注入对话。这样 LLM 在后续轮次中能被动感知后台任务的完成。

为什么用 <task_notification> 而不是伪装成 tool_result？

因为 Messages API 要求一个 tool_use 严格对应一个 tool_result——原来的调用已经用占位符回复过了，后台结果只能以"新事件"的身份出现，不能冒充之前的 tool_result。

主循环的行为变化：快同步、慢异步

Turn 1: bash "npm install" (后台)
        ↓
     占位 tool_result → LLM 继续（不等待安装完成）
        ↓
Turn 2: read_file "package.json" (同步)
        ↓
     文件结果 + collect 发现 bg_0001 已完成
        ↓
     注入  到 user 消息
        ↓
LLM 在同一条消息里看到：
  「文件内容 + 安装完成通知」→ 继续下一步

效果：慢操作在后台跑着，快操作照常同步执行——两不耽误。