# 第三章 TodoWrite (待办写入)


`s01 > s02 > [ s03 ] s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12`

>“本专栏基于开源项目 `learn-claude-code` 的官方文档。原文档非常硬核，为了方便像我一样的新手小白理解，我对文档进行了**逐行精读**，并加入了很多**中文注释、大白话解释和踩坑记录**。希望这套‘咀嚼版’教程能帮你推开 AI Agent 开发的大门。”


项目地址：https://github.com/shareAI-lab/learn-claude-code

>"没有计划的 agent 走哪算哪" -- 先列步骤再动手, 完成率翻倍。


## 一、问题-**Models Forget (模型会健忘)**


大模型在处理**长链路任务 (Long-Horizon Tasks)** 时的致命缺陷：多步任务中, 模型会丢失进度 -- 重复做过的事、跳步、跑偏。对话越长越严重: 工具结果不断填满上下文, 系统提示的影响力逐渐被稀释。一个 10 步重构可能做完 1-3 步就开始即兴发挥, 因为 4-10 步已经被挤出注意力了。

## 二、解决方案


```
+--------+      +-------+      +---------+
|  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
```
### 1. 显式状态管理（Explicit State with `todo` tool）


它不仅让 LLM 思考，还强迫 LLM **把思考过程写下来**。


- **引入 `TodoManager` 类**：这是一个简单的 Python 类，用来存储任务列表。

- **增加 `todo` 工具**：允许模型调用这个工具来创建、更新任务列表。
	- 模型必须明确标出：哪个任务 `pending`（待办），哪个 `in_progress`（进行中），哪个 `completed`（已完成）。
	- **效果**：即使对话历史很长，模型的“当前任务状态”始终是清晰的，存储在 `TODO` 变量里，并在每次工具调用后返回给模型看。


### 2. 强制提醒机制（The "Nag" Reminder）


光有工具模型可能不用，或者用着用着就忘了。所以 `s03` 增加了一个**监督逻辑**。

## 三、工作原理


1. TodoManager 存储带状态的项目。同一时间只允许一个 `in_progress`。这是对模型的一种**行为约束（Constraint）**。 大模型有时候思维很发散，“双线程”工作：一边想改文件 A，一边又想顺手把文件 B 也修了。这通常会导致混乱和顾此失彼。这里强制规定：**你此时此刻只能做 ONE thing**


```
class TodoManager:
    def update(self, items: list) -> str:
        validated, in_progress_count = [], 0
        for item in items:
            status = item.get("status", "pending")
            if status == "in_progress":
                in_progress_count += 1
            # <-- 状态必须是 pending, in_progress, completed 三者之一
            validated.append({"id": item["id"], "text": item["text"],
                              "status": status})
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress")
        self.items = validated
        return self.render()
```

1. `todo` 工具和其他工具一样加入 dispatch map。


```
TOOL_HANDLERS = {
    # ...base tools...
    "todo": lambda **kw: TODO.update(kw["items"]),
}
```

1. 显式状态与计数器逻辑：代码使用 `used_todo` 标志位监测 todo 工具的使用。


- 如果用了 todo，计数器 `rounds_since_todo` 清零 。

- 如果没用 (比如只用了 bash)，计数器 +1 。

- 积攒到 3 次，上面的 nag reminder 就会被插入到用户消息中，强迫模型反思进度。


```

&nbsp; &nbsp; &nbsp; &nbsp; used_todo = False

&nbsp; &nbsp; &nbsp; &nbsp; for block in response.content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # ...

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if block.name == "todo":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; used_todo = True

&nbsp; &nbsp; &nbsp; &nbsp; rounds_since_todo = 0 if used_todo else rounds_since_todo + 1
```

1. nag reminder: 模型连续 3 轮以上不调用 `todo` 时注入提醒。


```
if rounds_since_todo >= 3 and messages:
    last = messages[-1]
    if last["role"] == "user" and isinstance(last.get("content"), list):
        last["content"].insert(0, {
            "type": "text",
            "text": "<reminder>Update your todos.</reminder>",
        })
```
"同时只能有一个 in_progress" 强制顺序聚焦。nag reminder 制造问责压力 -- 你不更新计划, 系统就追着你问。

## 四、相对 s02 的变更


|组件|之前(s02)|之后(s03)|
|:-:|:-:|:-:|
|Tools|4|5 (+todo)|
|规划|无|带状态的 TodoManager|
|Nag 注入|无|3 轮后注入 `<reminder>`|
|Agent loop|简单分发|+ rounds_since_todo 计数器|

## 五、试一试


```
cd learn-claude-code
python agents/s03_todo_write.py
```
试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):


1. `Refactor the file hello.py: add type hints, docstrings, and a main guard`

2. `Create a Python package with __init__.py, utils.py, and tests/test_utils.py`

3. `Review all Python files and fix any style issues`


### 完整代码


```
#!/usr/bin/env python3
import os

import subprocess

from pathlib import Path

  

from anthropic import Anthropic

from dotenv import load_dotenv

  

load_dotenv(override=True)

  

if os.getenv("ANTHROPIC_BASE_URL"):

&nbsp; &nbsp; os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)

  

WORKDIR = Path.cwd()

client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))

MODEL = os.environ["MODEL_ID"]

  

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."""

  
  

# -- TodoManager: structured state the LLM writes to --

class TodoManager:

&nbsp; &nbsp; def __init__(self):

&nbsp; &nbsp; &nbsp; &nbsp; self.items = []

  

&nbsp; &nbsp; def update(self, items: list) -> str:

&nbsp; &nbsp; &nbsp; &nbsp; if len(items) > 20:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; raise ValueError("Max 20 todos allowed")

&nbsp; &nbsp; &nbsp; &nbsp; validated = []

&nbsp; &nbsp; &nbsp; &nbsp; in_progress_count = 0

&nbsp; &nbsp; &nbsp; &nbsp; for i, item in enumerate(items):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; text = str(item.get("text", "")).strip()

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; status = str(item.get("status", "pending")).lower()

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; item_id = str(item.get("id", str(i + 1)))

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if not text:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; raise ValueError(f"Item {item_id}: text required")

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if status not in ("pending", "in_progress", "completed"):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; raise ValueError(f"Item {item_id}: invalid status '{status}'")

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if status == "in_progress":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; in_progress_count += 1

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; validated.append({

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "id": item_id,

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "text": text,

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "status": status

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; })

&nbsp; &nbsp; &nbsp; &nbsp; if in_progress_count > 1:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; raise ValueError("Only one task can be in_progress at a time")

&nbsp; &nbsp; &nbsp; &nbsp; self.items = validated

&nbsp; &nbsp; &nbsp; &nbsp; return self.render()

  

&nbsp; &nbsp; def render(self) -> str:

&nbsp; &nbsp; &nbsp; &nbsp; if not self.items:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return "No todos."

&nbsp; &nbsp; &nbsp; &nbsp; lines = []

&nbsp; &nbsp; &nbsp; &nbsp; for item in self.items:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; marker = {"pending": "[ ]", "in_progress": "[>]", "completed": "[x]"}[item["status"]]

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; lines.append(f"{marker} #{item['id']}: {item['text']}")

&nbsp; &nbsp; &nbsp; &nbsp; done = sum(1 for t in self.items if t["status"] == "completed")

&nbsp; &nbsp; &nbsp; &nbsp; lines.append(f"\n({done}/{len(self.items)} completed)")

&nbsp; &nbsp; &nbsp; &nbsp; return "\n".join(lines)

  
  

TODO = TodoManager()

  
  

# -- Tool implementations --

def safe_path(p: str) -> Path:

&nbsp; &nbsp; path = (WORKDIR / p).resolve()

&nbsp; &nbsp; if not path.is_relative_to(WORKDIR):

&nbsp; &nbsp; &nbsp; &nbsp; raise ValueError(f"Path escapes workspace: {p}")

&nbsp; &nbsp; return path

  

def run_bash(command: str) -> str:

&nbsp; &nbsp; dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]

&nbsp; &nbsp; if any(d in command for d in dangerous):

&nbsp; &nbsp; &nbsp; &nbsp; return "Error: Dangerous command blocked"

&nbsp; &nbsp; try:

&nbsp; &nbsp; &nbsp; &nbsp; r = subprocess.run(command, shell=True, cwd=WORKDIR,

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;capture_output=True, text=True, timeout=120)

&nbsp; &nbsp; &nbsp; &nbsp; out = (r.stdout + r.stderr).strip()

&nbsp; &nbsp; &nbsp; &nbsp; return out[:50000] if out else "(no output)"

&nbsp; &nbsp; except subprocess.TimeoutExpired:

&nbsp; &nbsp; &nbsp; &nbsp; return "Error: Timeout (120s)"

  

def run_read(path: str, limit: int = None) -> str:

&nbsp; &nbsp; try:

&nbsp; &nbsp; &nbsp; &nbsp; lines = safe_path(path).read_text().splitlines()

&nbsp; &nbsp; &nbsp; &nbsp; if limit and limit < len(lines):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; lines = lines[:limit] + [f"... ({len(lines) - limit} more)"]

&nbsp; &nbsp; &nbsp; &nbsp; return "\n".join(lines)[:50000]

&nbsp; &nbsp; except Exception as e:

&nbsp; &nbsp; &nbsp; &nbsp; return f"Error: {e}"

  

def run_write(path: str, content: str) -> str:

&nbsp; &nbsp; try:

&nbsp; &nbsp; &nbsp; &nbsp; fp = safe_path(path)

&nbsp; &nbsp; &nbsp; &nbsp; fp.parent.mkdir(parents=True, exist_ok=True)

&nbsp; &nbsp; &nbsp; &nbsp; fp.write_text(content)

&nbsp; &nbsp; &nbsp; &nbsp; return f"Wrote {len(content)} bytes"

&nbsp; &nbsp; except Exception as e:

&nbsp; &nbsp; &nbsp; &nbsp; return f"Error: {e}"

  

def run_edit(path: str, old_text: str, new_text: str) -> str:

&nbsp; &nbsp; try:

&nbsp; &nbsp; &nbsp; &nbsp; fp = safe_path(path)

&nbsp; &nbsp; &nbsp; &nbsp; content = fp.read_text()

&nbsp; &nbsp; &nbsp; &nbsp; if old_text not in content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return f"Error: Text not found in {path}"

&nbsp; &nbsp; &nbsp; &nbsp; fp.write_text(content.replace(old_text, new_text, 1))

&nbsp; &nbsp; &nbsp; &nbsp; return f"Edited {path}"

&nbsp; &nbsp; except Exception as e:

&nbsp; &nbsp; &nbsp; &nbsp; return f"Error: {e}"

  
  

TOOL_HANDLERS = {

&nbsp; &nbsp; "bash": &nbsp; &nbsp; &nbsp; lambda **kw: run_bash(kw["command"]),

&nbsp; &nbsp; "read_file": &nbsp;lambda **kw: run_read(kw["path"], kw.get("limit")),

&nbsp; &nbsp; "write_file": lambda **kw: run_write(kw["path"], kw["content"]),

&nbsp; &nbsp; "edit_file": &nbsp;lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),

&nbsp; &nbsp; "todo": &nbsp; &nbsp; &nbsp; lambda **kw: TODO.update(kw["items"]),

}

  

TOOLS = [

&nbsp; &nbsp; {"name": "bash", "description": "Run a shell command.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},

&nbsp; &nbsp; {"name": "read_file", "description": "Read file contents.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},

&nbsp; &nbsp; {"name": "write_file", "description": "Write content to file.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},

&nbsp; &nbsp; {"name": "edit_file", "description": "Replace exact text in file.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},

&nbsp; &nbsp; {"name": "todo", "description": "Update task list. Track progress on multi-step tasks.",

&nbsp; &nbsp; &nbsp;"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"]}},

]

  
  

# -- Agent loop with nag reminder injection --

def agent_loop(messages: list):

&nbsp; &nbsp; rounds_since_todo = 0

&nbsp; &nbsp; while True:

&nbsp; &nbsp; &nbsp; &nbsp; # Nag reminder is injected below, alongside tool results

&nbsp; &nbsp; &nbsp; &nbsp; response = client.messages.create(

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; model=MODEL, system=SYSTEM, messages=messages,

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tools=TOOLS, max_tokens=8000,

&nbsp; &nbsp; &nbsp; &nbsp; )

&nbsp; &nbsp; &nbsp; &nbsp; messages.append({"role": "assistant", "content": response.content})

&nbsp; &nbsp; &nbsp; &nbsp; if response.stop_reason != "tool_use":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return

&nbsp; &nbsp; &nbsp; &nbsp; results = []

&nbsp; &nbsp; &nbsp; &nbsp; used_todo = False

&nbsp; &nbsp; &nbsp; &nbsp; for block in response.content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if block.type == "tool_use":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; handler = TOOL_HANDLERS.get(block.name)

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; output = handler(**block.input) if handler else f"Unknown tool: {block.name}"

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; except Exception as e:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; output = f"Error: {e}"

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print(f"> {block.name}: {str(output)[:200]}")

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if block.name == "todo":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; used_todo = True

&nbsp; &nbsp; &nbsp; &nbsp; rounds_since_todo = 0 if used_todo else rounds_since_todo + 1

&nbsp; &nbsp; &nbsp; &nbsp; if rounds_since_todo >= 3:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; results.insert(0, {"type": "text", "text": "<reminder>Update your todos.</reminder>"})

&nbsp; &nbsp; &nbsp; &nbsp; messages.append({"role": "user", "content": results})

  
  

if __name__ == "__main__":

&nbsp; &nbsp; history = []

&nbsp; &nbsp; while True:

&nbsp; &nbsp; &nbsp; &nbsp; try:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; query = input("\033[36ms03 >> \033[0m")

&nbsp; &nbsp; &nbsp; &nbsp; except (EOFError, KeyboardInterrupt):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; break

&nbsp; &nbsp; &nbsp; &nbsp; if query.strip().lower() in ("q", "exit", ""):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; break

&nbsp; &nbsp; &nbsp; &nbsp; history.append({"role": "user", "content": query})

&nbsp; &nbsp; &nbsp; &nbsp; agent_loop(history)

&nbsp; &nbsp; &nbsp; &nbsp; response_content = history[-1]["content"]

&nbsp; &nbsp; &nbsp; &nbsp; if isinstance(response_content, list):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; for block in response_content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if hasattr(block, "text"):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print(block.text)

&nbsp; &nbsp; &nbsp; &nbsp; print()
```

s01 > s02 > [ s03 ] s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（3）TodoWrite (待办写入)

本文介绍 learn-claude-code 项目第3阶段（s03）的 TodoWrite（待办写入） 功能，用于解决大模型在长链路任务中"健忘"的问题。

编程语言

开源项目learn-claude-code详解AI Agent开发技巧，通过TodoManager类解决大模型多步任务健忘问题。采用显式状态管理和强制提醒机制，确保任务执行顺序性。提供完整Python代码实现，包含bash、文件读写等工具函数，帮助开发者构建更稳定的AI Agent系统。

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（3）TodoWrite (待办写入)-腾讯云开发者社区-腾讯云

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（3）TodoWrite (待办写入)

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（3）TodoWrite (待办写入)

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐