作者：逆境不可逃

技术永无止境

希望我的内容可以帮助到你！！！！！

---

大家吼 ! 我是 逆境不可逃 今天给大家带来文章《【与我学 ClaudeCode】规划与协调篇 之 TodoWrite 的 神奇之处》.

Learn-Claude-Code 官方地址 : https://github.com/shareAI-lab/learn-claude-code

前言

顺着前面s01 Agent循环、s02工具分发的底层架构一路进阶，本篇正式进入规划与协调核心体系。 从简单任务清单、子智能体上下文隔离，到按需技能加载，再到磁盘持久化任务依赖图，一步步让 Claude 智能体从 “只会被动执行” 进化为会规划、会拆分、懂依赖、能协同的成熟工程级 Agent。

整体学习路线： s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

---

 TodoWrite 任务清单机制：先定计划，再动手执行

1、问题根源：为什么早期 Agent 长任务必崩

几乎所有 v0/v1 版本的 Agent 都面临同一个致命问题：

多步任务中，模型会逐渐丢失进度 —— 重复做过的事、跳步、跑偏。对话越长越严重：工具结果不断填满上下文，系统提示的影响力逐渐被稀释。一个 10 步重构可能做完 1-3 步就开始即兴发挥，因为 4-10 步已经被挤出注意力窗口了。

传统的内部思维链 (CoT) 规划完全无法解决这个问题，因为：

• 计划是不可见的：用户和开发者都不知道模型在想什么
• 计划是短暂的：一旦思维链滚出上下文窗口，计划就永久丢失
• 计划是不可控的：模型随时可能偏离计划，没有任何约束

2、三大核心设计决策

TodoWrite 通过三个反直觉但极其有效的设计决策，从根本上解决了上述问题。每个决策都明确对比了替代方案的缺陷。

a. 强制计划外化：让计划可见可追踪

核心设计：不让模型在思维链里默默规划，而是强制通过todo工具将计划写入独立于 LLM 上下文的外部状态 (TodoManager)。每个计划项都有明确的状态：pending(待办)、in_progress(进行中)、completed(已完成)。

三大不可替代的好处：

1. 用户可见：执行前就能知道 Agent 打算做什么，彻底打破 "黑盒" 运行，出问题可提前干预
2. 开发者可调试：通过检查计划状态就能精准定位 Agent 卡在哪一步，大幅降低调试难度
3. 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_result

b. 核心工作机制

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 时，系统会自动在工具结果中注入<reminder>Update 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(override=True)
if os.getenv("ANTHROPIC_BASE_URL"):
    os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)

# 工作目录：所有文件操作都限制在此目录下，防止路径逃逸
WORKDIR = Path.cwd()
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]

b. 系统提示词

SYSTEM = f"""You 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(f"Item {item_id}: text required")
            # 验证状态只能是三个合法值
            if status not in ("pending", "in_progress", "completed"):
                raise ValueError(f"Item {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(f"Path 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, shell=True, cwd=WORKDIR,
                           capture_output=True, text=True, timeout=120)
        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(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        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 f"Unknown tool: {block.name}"
                except Exception as e:
                    output = f"Error: {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": "<reminder>Update 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.整个程序的执行流程

1. 程序启动，加载环境变量，初始化 Claude 客户端和 TodoManager
2. 显示命令行提示符，等待用户输入
3. 用户输入一个任务（比如 "创建一个计算器程序"）
4. 把用户输入添加到对话历史，调用agent_loop
5. agent_loop把对话历史发给 Claude
6. Claude 返回响应，说要调用todo工具，创建任务列表
7. 程序执行todo工具，更新任务状态，把结果返回给 Claude
8. Claude 返回响应，说要调用write_file工具，创建文件
9. 程序执行write_file工具，写入代码，把结果返回给 Claude
10. Claude 返回响应，说要调用todo工具，把第一个任务标记为已完成，第二个标记为进行中
11. 重复步骤 5-10，直到所有任务完成
12. Claude 返回最终回答，程序把回答打印到控制台
13. 回到步骤 2，等待用户下一个输入

5、TodoWrite 的核心优势与创新

1. 状态外化是根本突破：计划存储在独立于 LLM 上下文的外部状态中，彻底解决了长任务进度丢失问题
2. 硬约束远胜软提示：通过代码层面的强制约束来引导 LLM 行为，比系统提示词的软约束有效 100 倍
3. 极致的可观测性：用户和开发者都能清晰看到 Agent 的计划和进度，不再是黑盒
4. 简单到极致：整个系统只有 177 行代码，没有复杂的架构，但是经验上效果极好
5. 保留模型自主性：系统只提供规划框架和约束，具体计划内容完全由模型自己制定，保留了 LLM 的灵活性和创造力

6、运行示例

假设用户输入："给我写一个简单的计算器程序，支持加减乘除"

Agent 的典型执行流程：

1. LLM 首先调用todo工具，制定初始计划：

   [>] #1: 创建calculator.py文件
   [ ] #2: 实现加法函数
   [ ] #3: 实现减法函数
   [ ] #4: 实现乘法函数
   [ ] #5: 实现除法函数（处理除零错误）
   [ ] #6: 添加命令行交互界面
   [ ] #7: 测试程序
   

2. 调用write_file工具创建文件
3. 完成后调用todo工具，将 #1 标记为completed，#2 标记为in_progress
4. 调用edit_file工具添加加法函数
5. 以此类推，直到所有任务完成
6. 如果 LLM 连续 3 轮没有调用todo工具，系统会自动注入提醒
7. LLM 看到提醒后，会立即调用todo工具更新进度

---