learn-claude-code 后台任务机制详解慢命令进后台线程通知队列注入 Agent Loop【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code本篇基于 learn-claude-code 仓库的后台任务章节文档 docs/zh/s08-background-tasks.md完整讲解后台执行这一 harness 层机制的动机、架构与实现BackgroundManager如何用守护线程运行耗时命令、如何用线程安全的通知队列在每轮 LLM 调用前注入结果并结合 agents/s08_background_tasks.py 的源码给出可运行的配置与实操步骤。读完你能掌握在不阻塞 Agent Loop 的前提下并行执行npm install、pytest等慢命令的完整方案。问题阻塞式循环里模型只能干等文档开头给出的问题是有些命令要跑好几分钟——npm install、pytest、docker build。在阻塞式循环中工具调用必须等命令返回才能继续模型只能干等用户说装依赖顺便建个配置文件Agent 却只能一个一个来。learn-claude-code 的口号是 Bash is all you need它用 17 课渐进式地把一个 claude code 风格的 agent harness 从 0 搭到 1。后台任务文档中的 s08现 17 课体系中的 s11要解决的正是其中运行长任务阶段的问题。仓库 README 给这一课的定位是「慢操作丢后台agent 继续想下一步」—— 后台线程跑命令完成后注入通知属于Harness 层的机制模型继续思考harness 负责等待。解决方案主线程 后台线程 通知队列文档给出的架构示意主线程跑 agent loop后台线程跑子进程完成后把结果排入队列Main thread Background thread ----------------- ----------------- | agent loop | | subprocess runs | | ... | | ... | | [LLM call] ---------- | enqueue(result) | | ^drain queue | ----------------- ----------------- Timeline: Agent --[spawn A]--[spawn B]--[other work]---- | | v v [A runs] [B runs] (parallel) | | -- results injected before next LLM call --关键洞察就一句话Fire and forget —— 命令在跑的时候agent 不阻塞。只有子进程 I/O 被并行化agent loop 本身保持单线程。工作原理BackgroundManager 源码逐段解析文档给出了四步工作原理下面结合 agents/s08_background_tasks.py 的完整实现逐一展开。1. 线程安全的任务注册表 通知队列BackgroundManageragents/s08_background_tasks.py#L50-L54只维护三个状态class BackgroundManager: def __init__(self): self.tasks {} # task_id - {status, result, command} self._notification_queue [] # completed task results self._lock threading.Lock()tasks任务注册表task_id - {status, result, command}供check_background工具随时查询状态_notification_queue已完成任务的结果队列等待在 LLM 调用前被排空_lock保护队列的threading.Lock因为后台线程写、主线程读。2. run()启动守护线程立即返回def run(self, command: str) - str: Start a background thread, return task_id immediately. task_id str(uuid.uuid4())[:8] self.tasks[task_id] {status: running, result: None, command: command} thread threading.Thread( targetself._execute, args(task_id, command), daemonTrue ) thread.start() return fBackground task {task_id} started: {command[:80]}实现上有三个值得注意的细节task_id 取uuid4的前 8 位agents/s08_background_tasks.py#L58足够区分并发任务又便于模型在后续轮次里引用daemonTrue守护线程不会阻止进程退出——即使主循环结束残留的后台命令线程也随进程一起被回收立即返回字符串Background task xxxx started: ...这个字符串就是回给 LLM 的tool_result模型拿到 task_id 后可以继续干别的。3. _execute()子进程执行、超时保护、结果截断线程目标是_executeagents/s08_background_tasks.py#L66-L89它是整个机制中防御性最强的部分def _execute(self, task_id: str, command: str): Thread target: run subprocess, capture output, push to queue. try: r subprocess.run( command, shellTrue, cwdWORKDIR, capture_outputTrue, textTrue, timeout300 ) output (r.stdout r.stderr).strip()[:50000] status completed except subprocess.TimeoutExpired: output Error: Timeout (300s) status timeout except Exception as e: output fError: {e} status error self.tasks[task_id][status] status self.tasks[task_id][result] output or (no output) with self._lock: self._notification_queue.append({ task_id: task_id, status: status, command: command[:80], result: (output or (no output))[:500], })关键参数与行为参数 / 行为取值作用shellTrue—支持、管道等 shell 语法与bash工具一致cwdWORKDIR进程启动时的当前目录后台任务与主循环共享工作区timeout300300 秒超时被捕获为timeout状态不会无限挂起结果存储上限[:50000]字符防止pytest全量输出撑爆tasks注册表通知队列 preview[:500]字符注入对话的只是结果摘要控制 token 开销空输出(no output)占位保证 LLM 一定能读到有意义的反馈状态机running / completed / timeout / error异常也走统一状态模型可据此决策注意截断的两级设计完整输出≤50000 字符留在tasks[task_id][result]里模型可以后续用check_background task_id查询进入通知队列的只有 500 字符的摘要。4. check() 与 drain_notifications()两种结果获取方式def check(self, task_id: str None) - str: Check status of one task or list all. if task_id: t self.tasks.get(task_id) if not t: return fError: Unknown task {task_id} return f[{t[status]}] {t[command][:60]}\n{t.get(result) or (running)} lines [] for tid, t in self.tasks.items(): lines.append(f{tid}: [{t[status]}] {t[command][:60]}) return \n.join(lines) if lines else No background tasks. def drain_notifications(self) - list: Return and clear all pending completion notifications. with self._lock: notifs list(self._notification_queue) self._notification_queue.clear() return notifscheck()是拉模式不带task_id时列出全部任务带task_id时返回单个任务的状态与结果运行中显示(running)drain_notifications()是推模式的触发点加锁、复制、清空原子地完成取出并清空保证同一条通知只被注入一次。Agent Loop 集成每次 LLM 调用前排空队列文档第 4 步是集成点。在 agents/s08_background_tasks.py#L188-L215 中agent_loop在每一轮调用 LLM 之前先排空通知队列并把结果包装成一条background-results消息追加进对话def agent_loop(messages: list): while True: # Drain background notifications and inject as system message before LLM call notifs BG.drain_notifications() if notifs and messages: notif_text \n.join( f[bg:{n[task_id]}] {n[status]}: {n[result]} for n in notifs ) messages.append({role: user, content: fbackground-results\n{notif_text}\n/background-results}) response client.messages.create( modelMODEL, systemSYSTEM, messagesmessages, toolsTOOLS, max_tokens8000, ) ...这个注入位置是设计的核心不唤醒模型——后台完成不会打断正在进行的推理而是搭车下一次client.messages.createXML 标签包裹background-results.../background-results让模型能区分这是系统事件而非用户输入每行通知带task_id与状态模型能把它和之前background_run返回的 task_id 对应起来。配合系统提示SYSTEM You are a coding agent at {WORKDIR}. Use background_run for long-running commands.agents/s08_background_tasks.py#L46引导模型主动把慢命令分给后台。工具集6 个工具与分发表s08 的工具面是6 个4 个基础文件/命令工具 2 个后台专用工具通过TOOL_HANDLERS分发表注册agents/s08_background_tasks.py#L163-L170TOOL_HANDLERS { bash: lambda **kw: run_bash(kw[command]), read_file: lambda **kw: run_read(kw[path], kw.get(limit)), write_file: lambda **kw: run_write(kw[path], kw[content]), edit_file: lambda **kw: run_edit(kw[path], kw[old_text], kw[new_text]), background_run: lambda **kw: BG.run(kw[command]), check_background: lambda **kw: BG.check(kw.get(task_id)), }工具说明关键约束源码可查bash阻塞式 shell 命令120s 超时危险命令黑名单rm -rf /、sudo、shutdown等直接拦截read_file读文件可选limit行数路径必须resolve后仍在WORKDIR内safe_path校验write_file写文件自动建父目录输出上限 50000 字符edit_file精确替换一段文本找不到即报错只替换第一处replace(..., 1)background_run后台线程跑命令立即返回 task_id300s 超时结果截断见上文check_background查询单个任务或列出全部task_id可省略bash与background_run形成对照前者 120 秒超时、同步返回输出后者 300 秒超时、异步返回 task_id。模型可以按命令预期耗时自行分流。相对 s07Task System的变更文档给出的对比表继承原文档组件之前 (s07)之后 (s08)Tools86 (基础 background_run check)执行方式仅阻塞阻塞 后台线程通知机制无每轮排空的队列并发无守护线程s07 是任务系统agents/s07_task_system.pytask_create / task_update / task_list / task_get四个任务工具 4 个基础工具 8 个而 s08 用background_run / check_background两个后台工具换掉了任务工具回到 6 个工具。两课解决的是不同问题s07 让目标在压缩后存活落盘 JSON 依赖图s08 让慢命令不阻塞循环。实操运行与推荐 prompt环境配置运行前提见 requirements.txt 与 .env.examplepip install -r requirements.txt # anthropic0.25.0, python-dotenv1.0.0, pyyaml6.0 cp .env.example .env.env中需要配置ANTHROPIC_API_KEYsk-ant-xxx # 必填 MODEL_IDclaude-sonnet-4-6 # 必填也可换 Anthropic 兼容服务商的模型 # ANTHROPIC_BASE_URL... # 可选指向兼容端点设置后会清掉 ANTHROPIC_AUTH_TOKEN运行cd learn-claude-code python agents/s08_background_tasks.py程序以交互式 REPL 启动提示符s08 输入q/exit退出。文档推荐的三个测试 prompt英文 prompt 对 LLM 效果更好也可以用中文Run sleep 5 echo done in the background, then create a file while it runs—— 验证后台跑命令的同时创建文件的并行能力Start 3 background tasks: sleep 2, sleep 4, sleep 6. Check their status.—— 验证多任务并发与check_background的状态查询Run pytest in the background and keep working on other things—— 用真实慢命令验证通知注入时机。观察要点background_run是否立即返回 task_id后续轮次的background-results里是否出现对应 task_id 的完成通知check_background不带参数时是否列出全部任务。延伸从 legacy s08 到现 17 课体系的 s11需要说明一点版本关系本文档属于 legacy 12 课轨道agents/docs/。仓库 README 给出了映射表——old s08 对应 new s11Background Tasks。现行 17 课实现位于 s11_background_tasks/s11_background_tasks/README.zh.md、s11_background_tasks/code.py机制在两点上演进显式参数取代独立工具不再有background_run工具而是给bash的 schema 增加run_in_background布尔参数should_run_background()只在tool_name bash且参数明确为True时进入后台路径不做关键词猜测通知不复用 tool_use_id后台命令先返回带bg_id的占位tool_result保持一个tool_use只对应一个tool_result完成结果在后续轮次以独立的task_notification事件注入格式为task_id.../task_idstatuscompleted/status。这些行为有自动化测试背书tests/test_background_tasks.py 验证了后台执行必须显式声明、权限检查先于后台分发rm -rf类命令即使带run_in_background: true也会被 Permission 拦截、且不会创建后台任务、完成结果只在后续 LLM 调用前被收集一次collect_background_results()第二次调用返回空列表。小结机制BackgroundManager用守护线程 锁保护的通知队列把子进程 I/O 并行化agent loop 保持单线程关键参数后台命令 300s 超时、结果存储 50000 字符、通知摘要 500 字符、task_id 为 8 位 UUID 前缀集成点每次client.messages.create之前drain_notifications()以background-results消息注入可验证路径文档 docs/zh/s08-background-tasks.mdlegacy 实现 agents/s08_background_tasks.py现行实现 s11_background_tasks/code.py测试 tests/test_background_tasks.py。这一课给出的模式可以抽象为通用结论在 LLM 驱动的循环里并行化应该发生在 I/O 层子进程、网络请求而不是循环本身结果回收统一挂到下一次模型调用前这个天然同步点既能避免阻塞又不需要引入回调、事件总线等更复杂的并发设施。【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考