1. 慢操作卡住 Agent Loop 的真实场景Agent Harness 跑起来之后最容易被忽略的坑不是模型选错工具而是某个工具 handler 一跑就是几分钟把整个前台循环钉死在那里。你让 Agent 先执行一次依赖安装再顺手读个配置文件结果安装命令同步阻塞读配置这一步根本轮不到执行模型也只能干等。这就是典型的慢操作阻塞主循环问题。Background Tasks 要解决的就是这件事把耗时工具从 Agent Loop 的主线程里挪出去前台先拿到一个占位结果继续往下走真实输出等后台跑完再回填。适合谁适合已经写过基础 Agent Loop、手里有 Task 系统但还没处理异步工具执行的开发者。Python threading 加 queue 就能搭出一个最小可用的后台任务骨架不需要引入 Celery 这类重型组件。我试过把subprocess.run()直接塞进循环里一条pip install就让整个对话卡住用户那边看起来像死机。后来把慢命令丢进 daemon thread前台立刻返回bg_0001循环继续处理其他 tool call体验完全不一样。下面按可复制的步骤走一遍。2. TaoToken 前置把模型调用和后台任务解耦后台任务骨架本身不依赖具体模型供应商但你要验证「占位结果 完成通知」这条链路得有一个能正常返回 tool_calls 的模型端点。TaoToken 提供 OpenAI 兼容的 Chat Completions 接口assistant 的tool_calls和roletool的tool_call_id配对行为和官方一致适合拿来核对消息协议边界。接入前先在控制台创建 API Key地址是 https://taotoken.net/api-keys 拿到 key 之后配置环境变量。模型对话调试可以用 https://taotoken.net/model-chat 直接发一轮带 tools 的请求观察返回结构里tool_calls[].id长什么样。如果你打算长期跑编码类 AgentCoding Plan 页面 https://taotoken.net/coding-plan 有更完整的额度说明。接入文档在 https://taotoken.net/doc 里面写了 base_url 和鉴权头的写法。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意 base_url 用https://taotoken.net/api不要带多余路径。OpenAI SDK 里这样初始化from openai import OpenAI import os client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], )这一步只是把模型端点准备好后台任务的核心逻辑还是纯 Python threading 和 queue跟供应商无关。这样设计的好处是将来换端点或加本地模型后台骨架不用动。3. 可复制的 daemon thread queue 配置先定义后台任务的状态存储。用两个字典分别记录「在做什么」和「得到了什么」再用一把 Lock 保护跨线程读写。这里我倾向用queue.Queue替代加锁字典来传递完成事件因为 Queue 自带线程安全的 put/get多生产者多消费者场景更省心。import threading import queue import json import time import uuid background_tasks {} # bg_id - 元数据 background_results {} # bg_id - 输出文本 background_lock threading.Lock() completion_queue queue.Queue() # worker 完成后投递 bg_id def start_background_task(tool_call_id: str, command: str, handler) - str: 启动守护线程执行 handler立即返回 bg_id。 bg_id fbg_{uuid.uuid4().hex[:8]} def worker(): try: output handler() status completed except Exception as exc: output ferror: {exc} status failed with background_lock: background_tasks[bg_id][status] status background_results[bg_id] output completion_queue.put(bg_id) # 通知 collector with background_lock: background_tasks[bg_id] { tool_call_id: tool_call_id, command: command, status: running, } thread threading.Thread(targetworker, daemonTrue) thread.start() return bg_id关键点有三个。第一background_tasks[bg_id]必须在thread.start()之前登记否则 worker 极快结束时 collector 可能访问到不存在的 ID。第二execute_tool()这类耗时操作放在锁外执行只在更新状态和结果时持锁否则锁粒度会抵消异步优势。第三daemonTrue只是让主进程退出时不被后台线程拖住不代表任务可靠完成进程关闭时守护线程会被突然停止文件句柄和事务可能没释放。判断一个工具该不该后台化用显式标志加关键字启发SLOW_KEYWORDS [install, build, test, deploy, compile, docker build, pip install, npm install, cargo build, pytest, make] def should_run_background(tool_name: str, tool_input: dict) - bool: if tool_input.get(run_in_background): return True if tool_name ! bash: return False command tool_input.get(command, ).lower() return any(kw in command for kw in SLOW_KEYWORDS)显式标志是模型通过 Tool Schema 提出的意图启发式只是容错。两者都不是可靠的资源调度pytest -q可能几秒结束不含关键字的数据迁移却可能跑几小时。生产系统应让工具元数据声明预期耗时和可后台性由 Harness 最终决定。4. 验证请求与成功结果后台任务跑起来后要验证两件事前台是否立刻拿到占位结果以及完成通知是否正确接回 messages。先写一个模拟慢操作的 handlerdef slow_handler(): time.sleep(3) return build finished: 42 artifacts然后模拟一轮 tool call 处理def handle_tool_call(tool_call_id, command, handler): if should_run_background(bash, {command: command}): bg_id start_background_task(tool_call_id, command, handler) return f[Background task {bg_id} started] Result will be available when complete. return handler() # 前台立即返回 placeholder handle_tool_call(call_abc, pip install requests, slow_handler) print(placeholder) # [Background task bg_1a2b3c4d started] Result will be available when complete.前台拿到占位结果后用原tool_call_id回填roletool消息配对关系保持一对一。真实命令结束后collector 从队列取出完成事件组装成新的 user 消息注入def collect_background_results() - list[str]: notifications [] while True: try: bg_id completion_queue.get_nowait() except queue.Empty: break with background_lock: task background_tasks.pop(bg_id, None) output background_results.pop(bg_id, ) if task is None: continue notifications.append( task_notification\n f task_id{bg_id}/task_id\n f status{task[status]}/status\n f command{task[command]}/command\n f summary{output[:200]}/summary\n /task_notification ) return notifications跑一遍完整流程你会看到T0 收到慢 tool callT1 创建 bg_id 并启动 worker前台回填占位 tool resultT2 循环继续处理其他快工具T3 worker 写入输出并投递队列T4 collector 取出结果组装通知T5 下一轮把通知作为 user 消息送回模型。原 tool call 只配对一次真实结果以新的 observation 形式出现不重复使用同一个tool_call_id。task_notification是 Harness 内部通知协议不是 OpenAI API 的标准 message roleXML 标签只是应用选择的可读包装。用pop()让已收集的结果不会在下一轮重复注入这是最小的进程内去重。5. 本篇常见错排查占位结果没回填模型卡住。assistant 已经输出带 ID 的 tool call后续必须有roletool且tool_call_id对应的结果。哪怕真实输出还没出来也要先回填「已启动」的占位文本否则当前消息组不完整模型无法继续。真实结果又用原 tool_call_id 发了一次。这相当于一个调用返回两次结果破坏消息组一对一关系。真实完成是之后发生的环境事件应组装成新的 user 消息注入不要复用原 ID。worker 里把 execute_tool 放在锁内。其他线程连查状态都要等慢命令结束异步优势被锁粒度抵消。只在更新background_tasks和background_results时持锁。collector 只在工具处理后调用后台完成后没人唤醒。如果后台命令在 Agent 已返回纯文本并退出循环后才完成且之后没有新用户输入通知只能留在队列里等下一次交互。本篇实现的是「后台完成后可在后续轮次看见」还没实现「完成事件立即主动唤醒 Agent」后者需要 event loop 或独立唤醒器。daemon thread 被当成可靠执行。进程退出时守护线程会被突然停止打开的文件、事务和其他资源可能没正常释放。daemonTrue只是让 CLI 退出时不被拖住任务可靠性要靠持久化队列和 worker lease。多进程下用 threading.Lock 协调。threading.Lock只协调当前进程的线程其他进程、机器和重启后的 worker 都看不到这把锁。跨进程要换分布式锁或队列。输出只留前 200 字符完整结果丢了。collector 只把摘要注入通知完整结果被 pop 后没有持久可查的 artifact 地址。生产环境应把完整输出写到对象存储或文件通知里带引用路径。没有取消和超时管理。run_bash()有单次 subprocess 超时却没有对 background job 暴露 cancel、deadline 或进程组终止。慢命令失控时无法回收。6. 下一步从后台执行到主动唤醒后台任务骨架跑通后你已经能把慢操作移出前台让 Agent Loop 继续处理其他动作。但当前方案仍有两个边界一是完成通知依赖后续循环触发不会主动唤醒前台二是状态只在进程内存重启后bg_0001的状态和输出不会恢复。要补齐这些需要引入持久化队列、worker lease、heartbeat、幂等键和 delivery acknowledgement。Task repository 保存业务目标job queue 保存执行实例notification channel 保存可重放事件三者不应只用两个字典模拟。如果你在接入模型端点时遇到 tool_calls 配对问题可以到 https://taotoken.net/api-keys 创建 key 后用 https://taotoken.net/model-chat 发一轮带 tools 的请求核对返回结构接入细节看 https://taotoken.net/doc 。长期跑编码类 Agent 的话https://taotoken.net/coding-plan 有额度说明。下一篇会在这个后台执行层之上加 Cron Scheduler让未来时间点主动唤醒 Agent处理「每天 09:00 自动检查构建」这类需求。