learn-claude-code 后臺(tái)任務(wù)機(jī)制詳解慢命令進(jìn)后臺(tái)線程通知隊(duì)列注入 Agent Loop【免費(fèi)下載鏈接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1項(xiàng)目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code本篇基于 learn-claude-code 倉庫的后臺(tái)任務(wù)章節(jié)文檔 docs/zh/s08-background-tasks.md完整講解后臺(tái)執(zhí)行這一 harness 層機(jī)制的動(dòng)機(jī)、架構(gòu)與實(shí)現(xiàn)BackgroundManager如何用守護(hù)線程運(yùn)行耗時(shí)命令、如何用線程安全的通知隊(duì)列在每輪 LLM 調(diào)用前注入結(jié)果并結(jié)合 agents/s08_background_tasks.py 的源碼給出可運(yùn)行的配置與實(shí)操步驟。讀完你能掌握在不阻塞 Agent Loop 的前提下并行執(zhí)行npm install、pytest等慢命令的完整方案。問題阻塞式循環(huán)里模型只能干等文檔開頭給出的問題是有些命令要跑好幾分鐘——npm install、pytest、docker build。在阻塞式循環(huán)中工具調(diào)用必須等命令返回才能繼續(xù)模型只能干等用戶說裝依賴順便建個(gè)配置文件Agent 卻只能一個(gè)一個(gè)來。learn-claude-code 的口號(hào)是 Bash is all you need它用 17 課漸進(jìn)式地把一個(gè) claude code 風(fēng)格的 agent harness 從 0 搭到 1。后臺(tái)任務(wù)文檔中的 s08現(xiàn) 17 課體系中的 s11要解決的正是其中運(yùn)行長任務(wù)階段的問題。倉庫 README 給這一課的定位是「慢操作丟后臺(tái)agent 繼續(xù)想下一步」—— 后臺(tái)線程跑命令完成后注入通知屬于Harness 層的機(jī)制模型繼續(xù)思考harness 負(fù)責(zé)等待。解決方案主線程 后臺(tái)線程 通知隊(duì)列文檔給出的架構(gòu)示意主線程跑 agent loop后臺(tái)線程跑子進(jìn)程完成后把結(jié)果排入隊(duì)列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 --關(guān)鍵洞察就一句話Fire and forget —— 命令在跑的時(shí)候agent 不阻塞。只有子進(jìn)程 I/O 被并行化agent loop 本身保持單線程。工作原理BackgroundManager 源碼逐段解析文檔給出了四步工作原理下面結(jié)合 agents/s08_background_tasks.py 的完整實(shí)現(xiàn)逐一展開。1. 線程安全的任務(wù)注冊表 通知隊(duì)列BackgroundManageragents/s08_background_tasks.py#L50-L54只維護(hù)三個(gè)狀態(tài)class BackgroundManager: def __init__(self): self.tasks {} # task_id - {status, result, command} self._notification_queue [] # completed task results self._lock threading.Lock()tasks任務(wù)注冊表task_id - {status, result, command}供check_background工具隨時(shí)查詢狀態(tài)_notification_queue已完成任務(wù)的結(jié)果隊(duì)列等待在 LLM 調(diào)用前被排空_lock保護(hù)隊(duì)列的threading.Lock因?yàn)楹笈_(tái)線程寫、主線程讀。2. run()啟動(dòng)守護(hù)線程立即返回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]}實(shí)現(xiàn)上有三個(gè)值得注意的細(xì)節(jié)task_id 取uuid4的前 8 位agents/s08_background_tasks.py#L58足夠區(qū)分并發(fā)任務(wù)又便于模型在后續(xù)輪次里引用daemonTrue守護(hù)線程不會(huì)阻止進(jìn)程退出——即使主循環(huán)結(jié)束殘留的后臺(tái)命令線程也隨進(jìn)程一起被回收立即返回字符串Background task xxxx started: ...這個(gè)字符串就是回給 LLM 的tool_result模型拿到 task_id 后可以繼續(xù)干別的。3. _execute()子進(jìn)程執(zhí)行、超時(shí)保護(hù)、結(jié)果截?cái)嗑€程目標(biāo)是_executeagents/s08_background_tasks.py#L66-L89它是整個(gè)機(jī)制中防御性最強(qiáng)的部分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], })關(guān)鍵參數(shù)與行為參數(shù) / 行為取值作用shellTrue—支持、管道等 shell 語法與bash工具一致cwdWORKDIR進(jìn)程啟動(dòng)時(shí)的當(dāng)前目錄后臺(tái)任務(wù)與主循環(huán)共享工作區(qū)timeout300300 秒超時(shí)被捕獲為timeout狀態(tài)不會(huì)無限掛起結(jié)果存儲(chǔ)上限[:50000]字符防止pytest全量輸出撐爆tasks注冊表通知隊(duì)列 preview[:500]字符注入對(duì)話的只是結(jié)果摘要控制 token 開銷空輸出(no output)占位保證 LLM 一定能讀到有意義的反饋狀態(tài)機(jī)running / completed / timeout / error異常也走統(tǒng)一狀態(tài)模型可據(jù)此決策注意截?cái)嗟膬杉?jí)設(shè)計(jì)完整輸出≤50000 字符留在tasks[task_id][result]里模型可以后續(xù)用check_background task_id查詢進(jìn)入通知隊(duì)列的只有 500 字符的摘要。4. check() 與 drain_notifications()兩種結(jié)果獲取方式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時(shí)列出全部任務(wù)帶task_id時(shí)返回單個(gè)任務(wù)的狀態(tài)與結(jié)果運(yùn)行中顯示(running)drain_notifications()是推模式的觸發(fā)點(diǎn)加鎖、復(fù)制、清空原子地完成取出并清空保證同一條通知只被注入一次。Agent Loop 集成每次 LLM 調(diào)用前排空隊(duì)列文檔第 4 步是集成點(diǎn)。在 agents/s08_background_tasks.py#L188-L215 中agent_loop在每一輪調(diào)用 LLM 之前先排空通知隊(duì)列并把結(jié)果包裝成一條background-results消息追加進(jìn)對(duì)話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, ) ...這個(gè)注入位置是設(shè)計(jì)的核心不喚醒模型——后臺(tái)完成不會(huì)打斷正在進(jìn)行的推理而是搭車下一次client.messages.createXML 標(biāo)簽包裹background-results.../background-results讓模型能區(qū)分這是系統(tǒng)事件而非用戶輸入每行通知帶task_id與狀態(tài)模型能把它和之前background_run返回的 task_id 對(duì)應(yīng)起來。配合系統(tǒng)提示SYSTEM You are a coding agent at {WORKDIR}. Use background_run for long-running commands.agents/s08_background_tasks.py#L46引導(dǎo)模型主動(dòng)把慢命令分給后臺(tái)。工具集6 個(gè)工具與分發(fā)表s08 的工具面是6 個(gè)4 個(gè)基礎(chǔ)文件/命令工具 2 個(gè)后臺(tái)專用工具通過TOOL_HANDLERS分發(fā)表注冊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)), }工具說明關(guān)鍵約束源碼可查bash阻塞式 shell 命令120s 超時(shí)危險(xiǎn)命令黑名單rm -rf /、sudo、shutdown等直接攔截read_file讀文件可選limit行數(shù)路徑必須resolve后仍在WORKDIR內(nèi)safe_path校驗(yàn)write_file寫文件自動(dòng)建父目錄輸出上限 50000 字符edit_file精確替換一段文本找不到即報(bào)錯(cuò)只替換第一處replace(..., 1)background_run后臺(tái)線程跑命令立即返回 task_id300s 超時(shí)結(jié)果截?cái)嘁娚衔腸heck_background查詢單個(gè)任務(wù)或列出全部task_id可省略bash與background_run形成對(duì)照前者 120 秒超時(shí)、同步返回輸出后者 300 秒超時(shí)、異步返回 task_id。模型可以按命令預(yù)期耗時(shí)自行分流。相對(duì) s07Task System的變更文檔給出的對(duì)比表繼承原文檔組件之前 (s07)之后 (s08)Tools86 (基礎(chǔ) background_run check)執(zhí)行方式僅阻塞阻塞 后臺(tái)線程通知機(jī)制無每輪排空的隊(duì)列并發(fā)無守護(hù)線程s07 是任務(wù)系統(tǒng)agents/s07_task_system.pytask_create / task_update / task_list / task_get四個(gè)任務(wù)工具 4 個(gè)基礎(chǔ)工具 8 個(gè)而 s08 用background_run / check_background兩個(gè)后臺(tái)工具換掉了任務(wù)工具回到 6 個(gè)工具。兩課解決的是不同問題s07 讓目標(biāo)在壓縮后存活落盤 JSON 依賴圖s08 讓慢命令不阻塞循環(huán)。實(shí)操運(yùn)行與推薦 prompt環(huán)境配置運(yùn)行前提見 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 兼容服務(wù)商的模型 # ANTHROPIC_BASE_URL... # 可選指向兼容端點(diǎn)設(shè)置后會(huì)清掉 ANTHROPIC_AUTH_TOKEN運(yùn)行cd learn-claude-code python agents/s08_background_tasks.py程序以交互式 REPL 啟動(dòng)提示符s08 輸入q/exit退出。文檔推薦的三個(gè)測試 prompt英文 prompt 對(duì) LLM 效果更好也可以用中文Run sleep 5 echo done in the background, then create a file while it runs—— 驗(yàn)證后臺(tái)跑命令的同時(shí)創(chuàng)建文件的并行能力Start 3 background tasks: sleep 2, sleep 4, sleep 6. Check their status.—— 驗(yàn)證多任務(wù)并發(fā)與check_background的狀態(tài)查詢Run pytest in the background and keep working on other things—— 用真實(shí)慢命令驗(yàn)證通知注入時(shí)機(jī)。觀察要點(diǎn)background_run是否立即返回 task_id后續(xù)輪次的background-results里是否出現(xiàn)對(duì)應(yīng) task_id 的完成通知check_background不帶參數(shù)時(shí)是否列出全部任務(wù)。延伸從 legacy s08 到現(xiàn) 17 課體系的 s11需要說明一點(diǎn)版本關(guān)系本文檔屬于 legacy 12 課軌道agents/docs/。倉庫 README 給出了映射表——old s08 對(duì)應(yīng) new s11Background Tasks?，F(xiàn)行 17 課實(shí)現(xiàn)位于 s11_background_tasks/s11_background_tasks/README.zh.md、s11_background_tasks/code.py機(jī)制在兩點(diǎn)上演進(jìn)顯式參數(shù)取代獨(dú)立工具不再有background_run工具而是給bash的 schema 增加run_in_background布爾參數(shù)should_run_background()只在tool_name bash且參數(shù)明確為True時(shí)進(jìn)入后臺(tái)路徑不做關(guān)鍵詞猜測通知不復(fù)用 tool_use_id后臺(tái)命令先返回帶bg_id的占位tool_result保持一個(gè)tool_use只對(duì)應(yīng)一個(gè)tool_result完成結(jié)果在后續(xù)輪次以獨(dú)立的task_notification事件注入格式為task_id.../task_idstatuscompleted/status。這些行為有自動(dòng)化測試背書tests/test_background_tasks.py 驗(yàn)證了后臺(tái)執(zhí)行必須顯式聲明、權(quán)限檢查先于后臺(tái)分發(fā)rm -rf類命令即使帶run_in_background: true也會(huì)被 Permission 攔截、且不會(huì)創(chuàng)建后臺(tái)任務(wù)、完成結(jié)果只在后續(xù) LLM 調(diào)用前被收集一次collect_background_results()第二次調(diào)用返回空列表。小結(jié)機(jī)制BackgroundManager用守護(hù)線程 鎖保護(hù)的通知隊(duì)列把子進(jìn)程 I/O 并行化agent loop 保持單線程關(guān)鍵參數(shù)后臺(tái)命令 300s 超時(shí)、結(jié)果存儲(chǔ) 50000 字符、通知摘要 500 字符、task_id 為 8 位 UUID 前綴集成點(diǎn)每次client.messages.create之前drain_notifications()以background-results消息注入可驗(yàn)證路徑文檔 docs/zh/s08-background-tasks.mdlegacy 實(shí)現(xiàn) agents/s08_background_tasks.py現(xiàn)行實(shí)現(xiàn) s11_background_tasks/code.py測試 tests/test_background_tasks.py。這一課給出的模式可以抽象為通用結(jié)論在 LLM 驅(qū)動(dòng)的循環(huán)里并行化應(yīng)該發(fā)生在 I/O 層子進(jìn)程、網(wǎng)絡(luò)請(qǐng)求而不是循環(huán)本身結(jié)果回收統(tǒng)一掛到下一次模型調(diào)用前這個(gè)天然同步點(diǎn)既能避免阻塞又不需要引入回調(diào)、事件總線等更復(fù)雜的并發(fā)設(shè)施?！久赓M(fèi)下載鏈接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1項(xiàng)目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考