【与我学 ClaudeCode】规划与协调篇 之 TodoWrite 的 神奇之处
作者逆境不可逃技术永无止境希望我的内容可以帮助到你大家吼 ! 我是 逆境不可逃 今天给大家带来文章《【与我学 ClaudeCode】规划与协调篇 之 TodoWrite 的 神奇之处》.Learn-Claude-Code 官方地址 :https://github.com/shareAI-lab/learn-claude-code前言顺着前面s01 Agent循环、s02工具分发的底层架构一路进阶本篇正式进入规划与协调核心体系。 从简单任务清单、子智能体上下文隔离到按需技能加载再到磁盘持久化任务依赖图一步步让 Claude 智能体从 “只会被动执行” 进化为会规划、会拆分、懂依赖、能协同的成熟工程级 Agent。整体学习路线s01 s02s03 s04 s05 s06 | s07 s08 s09 s10 s11 s12TodoWrite 任务清单机制先定计划再动手执行1、问题根源为什么早期 Agent 长任务必崩几乎所有 v0/v1 版本的 Agent 都面临同一个致命问题多步任务中模型会逐渐丢失进度 —— 重复做过的事、跳步、跑偏。对话越长越严重工具结果不断填满上下文系统提示的影响力逐渐被稀释。一个 10 步重构可能做完 1-3 步就开始即兴发挥因为 4-10 步已经被挤出注意力窗口了。传统的内部思维链 (CoT) 规划完全无法解决这个问题因为计划是不可见的用户和开发者都不知道模型在想什么计划是短暂的一旦思维链滚出上下文窗口计划就永久丢失计划是不可控的模型随时可能偏离计划没有任何约束2、三大核心设计决策TodoWrite 通过三个反直觉但极其有效的设计决策从根本上解决了上述问题。每个决策都明确对比了替代方案的缺陷。a. 强制计划外化让计划可见可追踪核心设计不让模型在思维链里默默规划而是强制通过todo工具将计划写入独立于 LLM 上下文的外部状态 (TodoManager)。每个计划项都有明确的状态pending(待办)、in_progress(进行中)、completed(已完成)。三大不可替代的好处用户可见执行前就能知道 Agent 打算做什么彻底打破 黑盒 运行出问题可提前干预开发者可调试通过检查计划状态就能精准定位 Agent 卡在哪一步大幅降低调试难度Agent 自身可引用即使早期上下文已经滚出窗口计划仍存在于外部状态中永远不会丢失替代方案的致命缺陷用内部 CoT 规划v0/v1 的做法计划是不可见且短暂的。Claude 的扩展思考也是一样 —— 一旦思考内容滚出上下文计划就没了而且用户和下游工具永远检查不到。b. 单任务强制聚焦同一时间只做一件事核心设计TodoManager 通过代码硬约束强制任何时候最多只能有一个任务处于in_progress状态。如果模型想开始第二个任务必须先完成或放弃当前任务。解决的隐蔽失败模式试图交替处理多个任务的模型往往会丢失状态产出大量半成品。LLM 的上下文切换能力极差会搞混不同任务的细节记不清自己在做哪个。替代方案的致命缺陷允许多个进行中的任务看起来更灵活但在实践中会导致输出质量灾难性下降。单聚焦约束是一个简单但极其有效的护栏能显著提升任务完成率。c. 20 条计划上限防止过度规划核心设计TodoWrite 将计划项数量严格限制在 20 条以内这是对 过度规划 的刻意约束。为什么 20 条是黄金数字不加限制时模型倾向于把任务分解成越来越细粒度的步骤产出 50 条甚至更多计划每一步都微不足道冗长的计划极其脆弱如果第 15 步失败剩下的 35 步可能全部作废20 条以内的短计划保持在正确的抽象层级更容易在现实偏离计划时做出调整经验证明绝大多数真实的编码任务都可以用 5-15 个有意义的步骤表达替代方案的致命缺陷没有上限会导致荒谬的详细计划动态上限与任务复杂度成正比更聪明但会大幅增加系统复杂度。固定 20 条是一个简单的启发式规则经验上效果极好。3、系统整体架构与工作原理a. 架构流程图-------- ------- --------- | User | --- | LLM | --- | Tools | | prompt | | | | todo | -------- ------ -------- ^ | | tool_result | ---------------- | ---------------------- | TodoManager state | | [ ] task A | | [] task B - doing | | [x] task C | ----------------------- | if rounds_since_todo 3: inject reminder into tool_resultb. 核心工作机制TodoWrite 由三个相互配合的核心组件构成(1) TodoManager外部状态管理器存储所有带状态的任务项强制执行三大约束最多 20 条任务、同一时间最多一个进行中任务、状态只能是三个合法值提供update()方法验证并更新任务列表违反约束会抛出明确错误提供render()方法将任务列表渲染成人类可读的格式返回给 LLM(2) Todo 工具与 LLM 的交互接口todo工具和其他基础工具bash、read_file、write_file 等完全平等LLM 通过调用todo工具来制定、更新和跟踪计划工具调用的结果渲染后的任务列表会被加入上下文让 LLM 始终知道当前进度(3) Nag 提醒系统问责机制系统维护一个rounds_since_todo计数器记录 LLM 连续多少轮没有调用 todo 工具每轮循环结束后如果调用了 todo 工具计数器重置为 0否则加 1当计数器达到 3 时系统会自动在工具结果中注入reminderUpdate your todos./reminder这个机制制造了持续的 问责压力模型不更新计划系统就会一直提醒它4、完整代码逐行解析a. 初始化与配置import os import subprocess from pathlib import Path from anthropic import Anthropic from dotenv import load_dotenv # 加载环境变量支持自定义API端点 load_dotenv(overrideTrue) if os.getenv(ANTHROPIC_BASE_URL): os.environ.pop(ANTHROPIC_AUTH_TOKEN, None) # 工作目录所有文件操作都限制在此目录下防止路径逃逸 WORKDIR Path.cwd() client Anthropic(base_urlos.getenv(ANTHROPIC_BASE_URL)) MODEL os.environ[MODEL_ID]b. 系统提示词SYSTEM fYou are a coding agent at {WORKDIR}. Use the todo tool to plan multi-step tasks. Mark in_progress before starting, completed when done. Prefer tools over prose.明确告诉 LLM 必须使用 todo 工具来规划任务强调 开始前标记为 in_progress完成后标记为 completed要求 优先使用工具而不是文字描述这是 Agent 的核心原则c. TodoManager 核心实现class TodoManager: def __init__(self): self.items [] # 存储所有任务项 def update(self, items: list) - str: # 约束1最多20个任务 if len(items) 20: raise ValueError(Max 20 todos allowed) validated [] in_progress_count 0 for i, item in enumerate(items): text str(item.get(text, )).strip() status str(item.get(status, pending)).lower() item_id str(item.get(id, str(i 1))) # 验证任务文本不能为空 if not text: raise ValueError(fItem {item_id}: text required) # 验证状态只能是三个合法值 if status not in (pending, in_progress, completed): raise ValueError(fItem {item_id}: invalid status {status}) # 统计进行中的任务数量 if status in_progress: in_progress_count 1 validated.append({id: item_id, text: text, status: status}) # 约束2同一时间最多一个进行中的任务 if in_progress_count 1: raise ValueError(Only one task can be in_progress at a time) self.items validated return self.render() def render(self) - str: 将任务列表渲染成易读的格式 if not self.items: return No todos. lines [] for item in self.items: marker {pending: [ ], in_progress: [], completed: [x]}[item[status]] lines.append(f{marker} #{item[id]}: {item[text]}) # 显示完成进度 done sum(1 for t in self.items if t[status] completed) lines.append(f\n({done}/{len(self.items)} completed)) return \n.join(lines) # 全局唯一的TodoManager实例 TODO TodoManager()d. 基础工具实现def safe_path(p: str) - Path: 安全路径检查防止路径逃逸攻击 path (WORKDIR / p).resolve() if not path.is_relative_to(WORKDIR): raise ValueError(fPath escapes workspace: {p}) return path def run_bash(command: str) - str: 执行shell命令过滤危险操作 dangerous [rm -rf /, sudo, shutdown, reboot, /dev/] if any(d in command for d in dangerous): return Error: Dangerous command blocked try: r subprocess.run(command, shellTrue, cwdWORKDIR, capture_outputTrue, textTrue, timeout120) out (r.stdout r.stderr).strip() return out[:50000] if out else (no output) except subprocess.TimeoutExpired: return Error: Timeout (120s) # 其他工具read_file、write_file、edit_file略e. 工具注册# 工具处理函数映射 TOOL_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]), todo: lambda **kw: TODO.update(kw[items]), } # 工具描述和输入schema传给Anthropic API TOOLS [ # 其他工具略 {name: todo, description: Update task list. Track progress on multi-step tasks., input_schema: {type: object, properties: {items: {type: array, items: {type: object, properties: {id: {type: string}, text: {type: string}, status: {type: string, enum: [pending, in_progress, completed]}}, required: [id, text, status]}}}, required: [items]}}, ]f. Agent 主循环含 Nag 提醒def agent_loop(messages: list): rounds_since_todo 0 # Nag提醒计数器 while True: # 调用LLM生成回复 response client.messages.create( modelMODEL, systemSYSTEM, messagesmessages, toolsTOOLS, max_tokens8000, ) messages.append({role: assistant, content: response.content}) # 如果不是工具调用结束本轮循环 if response.stop_reason ! tool_use: return results [] used_todo False # 执行所有工具调用 for block in response.content: if block.type tool_use: handler TOOL_HANDLERS.get(block.name) try: output handler(**block.input) if handler else fUnknown tool: {block.name} except Exception as e: output fError: {e} print(f {block.name}:) print(str(output)[:200]) results.append({type: tool_result, tool_use_id: block.id, content: str(output)}) # 标记是否使用了todo工具 if block.name todo: used_todo True # 更新Nag计数器 rounds_since_todo 0 if used_todo else rounds_since_todo 1 # 触发Nag提醒连续3轮没调用todo if rounds_since_todo 3: results.append({type: text, text: reminderUpdate your todos./reminder}) # 将工具结果和提醒加入上下文 messages.append({role: user, content: results})g. 交互式命令行入口if __name__ __main__: history [] # 全局对话历史 while True: try: query input(\033[36ms03 \033[0m) except (EOFError, KeyboardInterrupt): break if query.strip().lower() in (q, exit, ): break history.append({role: user, content: query}) agent_loop(history) # 打印最终回复 response_content history[-1][content] if isinstance(response_content, list): for block in response_content: if hasattr(block, text): print(block.text) print()h.整个程序的执行流程程序启动加载环境变量初始化 Claude 客户端和 TodoManager显示命令行提示符等待用户输入用户输入一个任务比如 创建一个计算器程序把用户输入添加到对话历史调用agent_loopagent_loop把对话历史发给 ClaudeClaude 返回响应说要调用todo工具创建任务列表程序执行todo工具更新任务状态把结果返回给 ClaudeClaude 返回响应说要调用write_file工具创建文件程序执行write_file工具写入代码把结果返回给 ClaudeClaude 返回响应说要调用todo工具把第一个任务标记为已完成第二个标记为进行中重复步骤 5-10直到所有任务完成Claude 返回最终回答程序把回答打印到控制台回到步骤 2等待用户下一个输入5、TodoWrite 的核心优势与创新状态外化是根本突破计划存储在独立于 LLM 上下文的外部状态中彻底解决了长任务进度丢失问题硬约束远胜软提示通过代码层面的强制约束来引导 LLM 行为比系统提示词的软约束有效 100 倍极致的可观测性用户和开发者都能清晰看到 Agent 的计划和进度不再是黑盒简单到极致整个系统只有 177 行代码没有复杂的架构但是经验上效果极好保留模型自主性系统只提供规划框架和约束具体计划内容完全由模型自己制定保留了 LLM 的灵活性和创造力6、运行示例假设用户输入给我写一个简单的计算器程序支持加减乘除Agent 的典型执行流程LLM 首先调用todo工具制定初始计划[] #1: 创建calculator.py文件 [ ] #2: 实现加法函数 [ ] #3: 实现减法函数 [ ] #4: 实现乘法函数 [ ] #5: 实现除法函数处理除零错误 [ ] #6: 添加命令行交互界面 [ ] #7: 测试程序调用write_file工具创建文件完成后调用todo工具将 #1 标记为completed#2 标记为in_progress调用edit_file工具添加加法函数以此类推直到所有任务完成如果 LLM 连续 3 轮没有调用todo工具系统会自动注入提醒LLM 看到提醒后会立即调用todo工具更新进度