v2: 使用待办事项的结构化规划

约 300 行代码。+1 个工具。明确的任务跟踪。

v1 有效。但对于复杂的任务，模型可能会失去轨迹。

让它"重构 auth、添加测试、更新文档"，看看会发生什么。没有明确的规划，它会在任务之间跳来跳去，忘记步骤，失去焦点。

v2 添加了一件事：待办事项工具。大约 100 行新代码，从根本上改变了智能体的工作方式。

问题

在 v1 中，计划只存在于模型的"头脑"中：

v1: "我会做 A，然后 B，然后 C" （不可见）
    10 个工具后："等等，我在做什么来着？"

待办事项工具使其明确：

v2:
  [ ] 重构 auth 模块
  [>] 添加单元测试         <- 当前在这里
  [ ] 更新文档

现在你和模型都能看到计划了。

待办事项管理器

一个带约束的列表：

class TodoManager:
    def __init__(self):
        self.items = []  # 最多 20 个

    def update(self, items):
        # 验证：
        # - 每个都需要：content, status, activeForm
        # - 状态：pending | in_progress | completed
        # - 只有一个可以是 in_progress
        # - 没有重复，没有空值

约束很重要：

规则	为什么
最多 20 项	防止无限列表
一个 in_progress	强制聚焦
必填字段	结构化输出

这些不是任意的——它们是护栏。

工具

{
    "name": "TodoWrite",
    "input_schema": {
        "items": [{
            "content": "任务描述",
            "status": "pending | in_progress | completed",
            "activeForm": "现在时：'正在读取文件'"
        }]
    }
}

activeForm 显示现在正在发生什么：

[>] 正在读取认证代码...  <- activeForm
[ ] 添加单元测试

系统提醒

软约束以鼓励使用待办事项：

INITIAL_REMINDER = "<reminder>对多步骤任务使用 TodoWrite。</reminder>"
NAG_REMINDER = "<reminder>超过 10 轮没有待办事项。请更新。</reminder>"

作为上下文注入，而不是命令：

if rounds_without_todo > 10:
    inject_reminder(NAG_REMINDER)

模型看到它们，但不对它们直接回应。

反馈循环

当模型调用 TodoWrite 时：

输入：
  [x] 重构 auth（已完成）
  [>] 添加测试（进行中）
  [ ] 更新文档（待处理）

返回：
  "[x] 重构 auth
   [>] 添加测试
   [ ] 更新文档
   （1/3 已完成）"

模型看到自己的计划。更新它。带着上下文继续。

什么时候待办事项有帮助

不是每个任务都需要它们：

适合	为什么
多步骤工作	5+ 个步骤要跟踪
长对话	20+ 个工具调用
复杂重构	多个文件
教学	可见的"思考"

经验法则：如果你会写一个清单，就使用待办事项。

集成

v2 添加到 v1 而不改变它：

# v1 工具
tools = [bash, read_file, write_file, edit_file]

# v2 添加
tools.append(TodoWrite)
todo_manager = TodoManager()

# v2 跟踪使用情况
if rounds_without_todo > 10:
    inject_reminder()

大约 100 行新代码。相同的智能体循环。

更深的洞察

结构约束并启用。

待办事项约束（最大项、一个 in_progress）启用（可见计划、跟踪进度）。

智能体设计中的模式：

max_tokens 约束 → 启用可管理的响应
工具模式约束 → 启用结构化调用
待办事项约束 → 启用复杂任务完成

好的约束不是限制。它们是脚手架。

学习笔记

按此顺序阅读源代码

打开 v2_todo_agent.py 并首先阅读这些部分：

TodoManager
SYSTEM
TOOLS
run_todo
execute_tool
agent_loop
main

重要的一点是 v2 并没有取代 v1 的智能体循环。它在同一个循环周围添加了一个有状态的工具。

v1:
模型 → 工具 → 结果 → 模型

v2:
模型 → 工具 + TodoWrite → 结果 + 可见计划 → 模型

TodoManager 实际上存储什么

在源代码中，TodoManager 只是一个有一个字段的对象：

self.items = []

模型不会发送像"标记项目 2 完成"这样的小补丁。它每次发送完整的新待办事项列表：

def update(self, items: list) -> str:
    ...
    self.items = validated
    return self.render()

这个设计很简单，对学习很有用：

模型总是拥有完整的当前计划。
主机在接受之前验证计划。
渲染后的计划作为工具结果返回。
渲染后的计划回到上下文，所以模型可以看到进度。

三个必填字段

每个待办事项必须有：

content
status
activeForm

把它们看作同一个任务的三种不同视图：

字段	含义	示例
`content`	稳定的任务名称	`添加单元测试`
`status`	状态机值	`pending`, `in_progress`, `completed`
`activeForm`	智能体现在正在做什么	`正在添加单元测试`

activeForm 很容易被低估。它不只是装饰；它使当前活动在追踪中可读：

[>] 添加单元测试 <- 正在添加单元测试

待办事项列表是一个小型状态机

状态值形成一个微小的状态机：

pending → in_progress → completed

关键护栏是：

只有一个项目可以是 in_progress

那条规则强制聚焦。没有它，模型可以声称同时在做很多事情，这使得计划不那么有用。

TodoWrite 如何成为智能体循环的一部分

TodoWrite 工具只是 TOOLS 中的另一个工具模式：

{
    "name": "TodoWrite",
    "description": "更新任务列表。用于规划和跟踪进度。",
    ...
}

调度器像任何其他工具一样路由它：

if name == "TodoWrite":
    return run_todo(args["items"])

所以核心循环仍然有相同的形状：

模型选择工具
主机执行工具
主机追加 tool_result
模型观察结果

唯一的区别是，一个工具更新内部智能体状态而不是文件系统。

为什么提醒是软的，不是硬的

源代码包括：

INITIAL_REMINDER = "<reminder>对多步骤任务使用 TodoWrite。</reminder>"
NAG_REMINDER = "<reminder>超过 10 轮没有待办事项更新。请更新待办事项。</reminder>"

这是一个重要的设计模式。程序不会强迫每个任务使用待办事项。当任务足够长，可见的计划会有帮助时，它会推动模型。

这就是为什么 v2 仍然感觉灵活：

小任务：不需要清单。
多步骤任务：TodoWrite 创建共享状态。
长任务：提醒减少漂移。

常见失败模式

学习或修改 v2 时注意这些：

打印待办事项是不够的。 它必须作为工具结果返回，以便模型可以观察它。
多个 in_progress 项目减少聚焦。 主机应该拒绝它们。
太多待办事项变成噪音。 最大计数是一个有用的约束。
隐藏的计划不是协作。 用户和模型都需要看到状态。

学习检查

阅读代码后，确保你能回答：

待办事项列表存储在哪里？
为什么 update() 接收完整列表而不是差异？
TodoWrite 在哪里进入工具调度器？
渲染的待办事项列表如何回到模型上下文？
为什么 v2 仍然使用与 v1 相同的智能体循环？

完整源码

#!/usr/bin/env python3
"""
v2_todo_agent.py - Mini Claude Code：结构化规划（约 300 行）

核心理念："让计划可见"
=====================================
v1 对简单任务非常有效。但让它"重构 auth、添加测试、
更新文档"，看看会发生什么。没有明确的规划，模型会：
  - 随机在任务之间跳来跳去
  - 忘记已完成的步骤
  - 中途失去焦点

问题 - "上下文褪色"：
----------------------------
在 v1 中，计划只存在于模型的"头脑"中：

    v1: "我会做 A，然后 B，然后 C" （不可见）
        10 个工具调用后："等等，我在做什么来着？"

解决方案 - TodoWrite 工具：
-----------------------------
v2 添加了一个新工具，从根本上改变了智能体的工作方式：

    v2:
      [ ] 重构 auth 模块
      [>] 添加单元测试         <- 当前正在处理这个
      [ ] 更新文档

现在你和模型都能看到计划了。模型可以：
  - 在工作时更新状态
  - 看到什么完成了，什么是下一个
  - 一次专注于一个任务

关键约束（不是任意的——这些是护栏）：
------------------------------------------------------
    | 规则              | 为什么                              |
    |-------------------|------------------------------------|
    | 最多 20 项        | 防止无限任务列表                     |
    | 一个 in_progress  | 一次只专注于一件事                    |
    | 必填字段           | 确保结构化输出                       |

更深的洞察：
----------------
> "结构约束并启用。"

待办事项约束（最大项、一个 in_progress）启用（可见计划、跟踪进度）。

这种模式出现在智能体设计的各处：
  - max_tokens 约束 → 启用可管理的响应
  - 工具模式约束 → 启用结构化调用
  - 待办事项约束 → 启用复杂任务完成

好的约束不是限制。它们是脚手架。

用法：
    python v2_todo_agent.py
"""

import os
import subprocess
import sys
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

try:
    from anthropic import Anthropic
except ImportError:
    sys.exit("请安装：pip install anthropic python-dotenv")


# =============================================================================
# 配置
# =============================================================================

API_KEY = os.getenv("ANTHROPIC_API_KEY")
BASE_URL = os.getenv("ANTHROPIC_BASE_URL")
MODEL = os.getenv("MODEL_NAME", "claude-sonnet-4-20250514")
WORKDIR = Path.cwd()

# 初始化客户端 - 处理直接 Anthropic 和兼容的 API
client = Anthropic(api_key=API_KEY, base_url=BASE_URL) if BASE_URL else Anthropic(api_key=API_KEY)


# =============================================================================
# TodoManager - v2 中的核心添加
# =============================================================================

class TodoManager:
    """
    管理带有强制约束的结构化任务列表。

    关键设计决策：
    --------------------
    1. 最多 20 项：防止模型创建无尽列表
    2. 一个 in_progress：强制聚焦——一次只能做一件事
    3. 必填字段：每个项目需要 content、status 和 activeForm

    activeForm 字段值得解释：
    - 它是正在发生的事情的现在时形式
    - 当 status 是"in_progress"时显示
    - 示例：content="添加测试", activeForm="正在添加单元测试..."

    这提供了对智能体正在做什么的实时可见性。
    """

    def __init__(self):
        self.items = []

    def update(self, items: list) -> str:
        """
        验证并更新待办事项列表。

        模型每次发送一个完整的新列表。我们验证它，
        存储它，并返回模型将看到的渲染视图。

        验证规则：
        - 每个项目必须有：content, status, activeForm
        - status 必须是：pending | in_progress | completed
        - 一次只有一个项目可以是 in_progress
        - 最多允许 20 个项目

        返回：
            待办事项列表的渲染文本视图
        """
        validated = []
        in_progress_count = 0

        for i, item in enumerate(items):
            # 提取并验证字段
            content = str(item.get("content", "")).strip()
            status = str(item.get("status", "pending")).lower()
            active_form = str(item.get("activeForm", "")).strip()

            # 验证检查
            if not content:
                raise ValueError(f"项目 {i}：需要 content")
            if status not in ("pending", "in_progress", "completed"):
                raise ValueError(f"项目 {i}：无效 status '{status}'")
            if not active_form:
                raise ValueError(f"项目 {i}：需要 activeForm")

            if status == "in_progress":
                in_progress_count += 1

            validated.append({
                "content": content,
                "status": status,
                "activeForm": active_form
            })

        # 强制执行约束
        if len(validated) > 20:
            raise ValueError("最多允许 20 个待办事项")
        if in_progress_count > 1:
            raise ValueError("一次只有一个任务可以是 in_progress")

        self.items = validated
        return self.render()

    def render(self) -> str:
        """
        将待办事项列表渲染为人类可读的文本。

        格式：
            [x] 已完成的任务
            [>] 进行中的任务 <- 正在做某事...
            [ ] 待处理的任务

            （已完成 2/3）

        这个渲染后的文本是模型作为工具结果看到的。
        然后它可以根据其当前状态更新列表。
        """
        if not self.items:
            return "没有待办事项。"

        lines = []
        for item in self.items:
            if item["status"] == "completed":
                lines.append(f"[x] {item['content']}")
            elif item["status"] == "in_progress":
                lines.append(f"[>] {item['content']} <- {item['activeForm']}")
            else:
                lines.append(f"[ ] {item['content']}")

        completed = sum(1 for t in self.items if t["status"] == "completed")
        lines.append(f"\n（已完成 {completed}/{len(self.items)}）")

        return "\n".join(lines)


# 全局待办事项管理器实例
TODO = TodoManager()


# =============================================================================
# 系统提示 - 为 v2 更新
# =============================================================================

SYSTEM = f"""你是位于 {WORKDIR} 的编码智能体。

循环：规划 → 使用工具行动 → 更新待办事项 → 报告。

规则：
- 使用 TodoWrite 跟踪多步骤任务
- 在开始前标记任务 in_progress，完成后标记 completed
- 优先使用工具而非文字。行动，不要只是解释。
- 完成后，摘要更改了什么。"""


# =============================================================================
# 系统提醒 - 软提示以鼓励使用待办事项
# =============================================================================

# 在对话开始时显示
INITIAL_REMINDER = "<reminder>对多步骤任务使用 TodoWrite。</reminder>"

# 如果模型有一段时间没有更新待办事项时显示
NAG_REMINDER = "<reminder>超过 10 轮没有待办事项更新。请更新待办事项。</reminder>"


# =============================================================================
# 工具定义（v1 工具 + TodoWrite）
# =============================================================================

TOOLS = [
    # v1 工具（不变）
    {
        "name": "bash",
        "description": "运行 shell 命令。",
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
    {
        "name": "read_file",
        "description": "读取文件内容。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "limit": {"type": "integer"}
            },
            "required": ["path"],
        },
    },
    {
        "name": "write_file",
        "description": "写入内容到文件。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"],
        },
    },
    {
        "name": "edit_file",
        "description": "替换文件中的精确文本。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "old_text": {"type": "string"},
                "new_text": {"type": "string"},
            },
            "required": ["path", "old_text", "new_text"],
        },
    },

    # v2 中的新功能：TodoWrite
    # 这是启用结构化规划的关键添加
    {
        "name": "TodoWrite",
        "description": "更新任务列表。用于规划和跟踪进度。",
        "input_schema": {
            "type": "object",
            "properties": {
                "items": {
                    "type": "array",
                    "description": "完整的任务列表（替换现有）",
                    "items": {
                        "type": "object",
                        "properties": {
                            "content": {
                                "type": "string",
                                "description": "任务描述"
                            },
                            "status": {
                                "type": "string",
                                "enum": ["pending", "in_progress", "completed"],
                                "description": "任务状态"
                            },
                            "activeForm": {
                                "type": "string",
                                "description": "现在时行动，例如 '正在读取文件'"
                            },
                        },
                        "required": ["content", "status", "activeForm"],
                    },
                }
            },
            "required": ["items"],
        },
    },
]


# =============================================================================
# 工具实现（v1 + TodoWrite）
# =============================================================================

def safe_path(p: str) -> Path:
    """确保路径保持在工作区内。"""
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"路径逃离工作区：{p}")
    return path


def run_bash(cmd: str) -> str:
    """执行带有安全检查的 shell 命令。"""
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot"]
    if any(d in cmd for d in dangerous):
        return "错误：危险命令已阻止"
    try:
        result = subprocess.run(
            cmd, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=60
        )
        output = (result.stdout + result.stderr).strip()
        return output[:50000] if output else "(无输出)"
    except subprocess.TimeoutExpired:
        return "错误：超时"
    except Exception as e:
        return f"错误：{e}"


def run_read(path: str, limit: int = None) -> str:
    """读取文件内容。"""
    try:
        text = safe_path(path).read_text()
        lines = text.splitlines()
        if limit and limit < len(lines):
            lines = lines[:limit] + [f"...（还有 {len(text.splitlines()) - limit} 个）"]
        return "\n".join(lines)[:50000]
    except Exception as e:
        return f"错误：{e}"


def run_write(path: str, content: str) -> str:
    """写入内容到文件。"""
    try:
        fp = safe_path(path)
        fp.parent.mkdir(parents=True, exist_ok=True)
        fp.write_text(content)
        return f"写入 {len(content)} 字节到 {path}"
    except Exception as e:
        return f"错误：{e}"


def run_edit(path: str, old_text: str, new_text: str) -> str:
    """替换文件中的精确文本。"""
    try:
        fp = safe_path(path)
        content = fp.read_text()
        if old_text not in content:
            return f"错误：在 {path} 中未找到文本"
        fp.write_text(content.replace(old_text, new_text, 1))
        return f"已编辑 {path}"
    except Exception as e:
        return f"错误：{e}"


def run_todo(items: list) -> str:
    """
    更新待办事项列表。

    模型发送一个完整的新列表（不是差异）。
    我们验证它并返回渲染视图。
    """
    try:
        return TODO.update(items)
    except Exception as e:
        return f"错误：{e}"


def execute_tool(name: str, args: dict) -> str:
    """将工具调用分派到实现。"""
    if name == "bash":
        return run_bash(args["command"])
    if name == "read_file":
        return run_read(args["path"], args.get("limit"))
    if name == "write_file":
        return run_write(args["path"], args["content"])
    if name == "edit_file":
        return run_edit(args["path"], args["old_text"], args["new_text"])
    if name == "TodoWrite":
        return run_todo(args["items"])
    return f"未知工具：{name}"


# =============================================================================
# 智能体循环（带待办事项跟踪）
# =============================================================================

# 跟踪自上次待办事项更新以来有多少轮
rounds_without_todo = 0


def agent_loop(messages: list) -> list:
    """
    带有待办事项使用跟踪的智能体循环。

    与 v1 相同的核心循环，但现在我们跟踪模型
    是否正在使用待办事项。如果它太长时间没有更新，
    我们会在 main() 函数中注入一个提醒。
    """
    global rounds_without_todo

    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000,
        )

        tool_calls = []
        for block in response.content:
            if hasattr(block, "text"):
                print(block.text)
            if block.type == "tool_use":
                tool_calls.append(block)

        if response.stop_reason != "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            return messages

        results = []
        used_todo = False

        for tc in tool_calls:
            print(f"\n> {tc.name}")
            output = execute_tool(tc.name, tc.input)
            preview = output[:300] + "..." if len(output) > 300 else output
            print(f"  {preview}")

            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": output,
            })

            # 跟踪待办事项使用情况
            if tc.name == "TodoWrite":
                used_todo = True

        # 更新计数器：如果使用了待办事项则重置，否则递增
        if used_todo:
            rounds_without_todo = 0
        else:
            rounds_without_todo += 1

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": results})


# =============================================================================
# 主 REPL
# =============================================================================

def main():
    """
    带有提醒注入的 REPL。

    关键 v2 添加：我们注入"提醒"消息以鼓励
    待办事项使用，而不是强迫它。这是一个软约束。

    提醒被注入为用户消息的一部分，而不是
    单独的系统提示。模型看到它们，但不会
    直接对它们做出回应。
    """
    global rounds_without_todo

    print(f"Mini Claude Code v2（带待办事项）- {WORKDIR}")
    print("键入 'exit' 退出。\n")

    history = []
    first_message = True

    while True:
        try:
            user_input = input("你：").strip()
        except (EOFError, KeyboardInterrupt):
            break

        if not user_input or user_input.lower() in ("exit", "quit", "q"):
            break

        # 构建用户消息内容
        # 可能包含提醒作为上下文提示
        content = []

        if first_message:
            # 开始时的温和提醒
            content.append({"type": "text", "text": INITIAL_REMINDER})
            first_message = False
        elif rounds_without_todo > 10:
            # 如果模型有一段时间没有使用待办事项，催促一下
            content.append({"type": "text", "text": NAG_REMINDER})

        content.append({"type": "text", "text": user_input})
        history.append({"role": "user", "content": content})

        try:
            agent_loop(history)
        except Exception as e:
            print(f"错误：{e}")

        print()


if __name__ == "__main__":
    main()

明确的规划使智能体可靠。

← v1 | 返回 README | v3 →

v2: 结构化计划