Claude Code 源码拆解(五):让Agent＂持久化＂和＂后台执行＂

一、两个新问题

当Agent运行一段时间后，会遇到两个问题：

问题1：上下文压缩了，任务状态怎么办？

上一篇我们讲了Compact机制：当对话太长时，会把历史压缩成摘要。

但问题是：任务状态也在对话历史里。

第50轮：用户说"帮我重构这个模块"第51轮：Agent创建任务列表...第100轮：上下文压缩第101轮：Agent："好的，请告诉我你要做什么？"

Agent忘记了它正在做什么。

问题2：耗时操作会阻塞Agent

Agent需要运行一个耗时30秒的测试，或者需要安装一个大型依赖包。

Agent: 我要运行测试[30秒等待...]Agent: 测试通过了，现在我要...（但用户已经等得不耐烦了）

Agent被阻塞，用户体验差。

二、解决方案1：任务持久化到磁盘

2.1 核心思想：状态在对话之外

把任务状态存到文件系统，而不是对话历史。

.tasks/  task_1.json  {"id":1, "subject":"重构配置模块", "status":"completed"}  task_2.json  {"id":2, "subject":"更新调用点", "status":"in_progress"}  task_3.json  {"id":3, "subject":"添加测试", "status":"pending", "blockedBy":[2]}

关键： 即使对话被压缩，Agent仍然可以通过task_list工具看到当前任务状态。

2.2 Task数据结构

task = {"id": 3,"subject": "添加测试用例","description": "为新模块添加单元测试","status": "pending",       # pending / in_progress / completed"blockedBy": [2],          # 被哪个任务阻塞"blocks": [],              # 阻塞了哪些任务"owner": "",               # 谁在负责（多Agent时用）}

2.3 依赖图自动解析

classTaskManager:defupdate(self, task_id: int, status: str = None, ...):        task = self._load(task_id)if status:            task["status"] = status# 关键：当任务完成时，自动解除其他任务的阻塞if status == "completed":                self._clear_dependency(task_id)        self._save(task)def_clear_dependency(self, completed_id: int):"""从所有其他任务的blockedBy中移除这个完成的任务"""for f in self.dir.glob("task_*.json"):            task = json.loads(f.read_text())if completed_id in task.get("blockedBy", []):                task["blockedBy"].remove(completed_id)                self._save(task)

效果：

初始状态：  Task 1: [pending]  Task 2: [pending] (blocked by: [1])  Task 3: [pending] (blocked by: [2])完成Task 1后：  Task 1: [completed]  Task 2: [pending]        ← 自动解除阻塞  Task 3: [pending] (blocked by: [2])

2.4 对话压缩后怎么办？

# 上下文被压缩后，Agent可以主动查询SYSTEM = "Use task_list to check current work status."# Agent的第一反应：> task_list()[ ] #2: 更新调用点[ ] #3: 添加测试 (blocked by: [2])> 好的，我继续完成Task 2...

关键洞察：

State that survives compression — because it’s outside the conversation.压缩后状态依然存在——因为它在对话之外。

三、解决方案2：后台执行

3.1 核心思想：Fire and Forget

耗时操作在后台线程执行，不阻塞主循环。

classBackgroundManager:def__init__(self):        self.tasks = {}                    # task_id -> {status, result}        self.notifications = []            # 完成的任务通知队列defrun(self, command: str) -> str:        task_id = str(uuid.uuid4())[:8]        self.tasks[task_id] = {"status": "running", "result": None}# 启动后台线程        thread = threading.Thread(            target=self._execute,            args=(task_id, command),            daemon=True        )        thread.start()returnf"Background task {task_id} started"def_execute(self, task_id: str, command: str):# 在后台线程执行        result = subprocess.run(command, shell=True, capture_output=True)        self.tasks[task_id]["status"] = "completed"        self.tasks[task_id]["result"] = result.stdout# 推送通知        self.notifications.append({"task_id": task_id,"status": "completed","result": result.stdout[:500]        })

3.2 通知注入机制

后台任务完成后，怎么告诉Agent？

defagent_loop(messages: list):whileTrue:# 关键：在每次LLM调用前，注入后台通知        notifs = BG.drain_notifications()if notifs:            notif_text = "\n".join(f"[bg:{n['task_id']}] {n['status']}: {n['result']}"for n in notifs            )            messages.append({"role": "user","content": f"<background-results>\n{notif_text}\n</background-results>"            })            messages.append({"role": "assistant","content": "Noted background results."            })# 然后调用LLM        response = client.messages.create(...)

3.3 执行时间线

Agent ----[spawn: npm install]----[spawn: npm test]----[继续写代码]----              |                          |              v                          v        [后台安装依赖]            [后台运行测试]              |                          |              +---- 通知队列 ----+                                    |                                    v              [注入到对话] <--- "npm install完成"

Agent不需要等，它继续做其他事。等任务完成，结果会自动注入。

四、这两个机制如何配合？

经典组合：

# 1. 创建任务task_create(subject="运行完整测试套件")# 2. 后台执行background_run(command="pytest tests/")# 3. 继续做其他事# ... 写代码 ...# 4. 后台结果注入<background-results>[bg:abc123] completed: 50 tests passed, 2 failed</background-results># 5. 根据结果更新任务task_update(task_id=1, status="completed")

五、这一篇学到了什么

1. Task系统是”外部记忆” —— 状态存在文件里，不怕压缩
2. Background是”异步执行” —— 不阻塞，通知注入
3. 依赖图自动解析 —— 完成任务时自动解除阻塞
4. 两者可以配合 —— Task管规划，Background管执行

下一期，我们将进入最激动人心的部分：多Agent协作。

我们会看到Claude Code如何实现：

• 持久化的”队友”
• 异步通信的”邮箱”
• 自主认领任务的机制

🔔 关注获取更新

这个系列会持续更新，下一期我们将进入多Agent协作的世界。

如果你想知道：

• 如何让多个Agent像团队一样协作？
• 它们之间怎么通信？
• 如何实现”自主认领”任务？

关注公众号，第一时间获取更新。