基于 s03_todo_write.py 源码逐行分析,配合 s03-todo-write.md 设计思路。


一、问题:对话越长,模型越容易跑偏

s02 的 Agent 有了四个工具,简单任务没问题。但复杂任务就出事了。

比如你说:“重构这个文件:加类型提示、加文档字符串、加 main guard、加错误处理、加单元测试”——五个步骤。模型一开始记得很清楚,做完第一步,做第二步。但到第三步、第四步时,messages 里已经累积了大量工具调用和返回结果(几百上千行代码),原始 prompt 里的五个步骤被淹没在最前面。

messages 不清空,信息确实还在。但 LLM 不是数据库——它不会平等对待上下文里的每一个 token。Transformer 的注意力机制天然对最近的内容给更高权重。原始 prompt 沉在上下文底部,几千行工具结果压在上面,模型的有效注意力已经照顾不到那五个步骤了。

结果就是:

用户: "给所有 Python 文件加类型注解"
模型: bash: ls *.py           ← 找到 8 个文件
模型: edit_file: hello.py     ← 改了第 1 个
模型: edit_file: utils.py     ← 改了第 2 个
模型: bash: pip install xxx   ← 跑偏了,去装包了...
模型: edit_file: hello.py     ← 又回头改 hello.py,重复劳动
  • 做完 1-2 步就开始即兴发挥
  • 重复做已经做过的事
  • 跳步、漏步
  • 做到一半跑偏去干别的

这不是代码的问题,是模型架构的特性。 注意力权重分布是训练决定的,你改不了。但你能改的是——让重要信息出现在注意力最强的位置。

两个基础概念,后面会反复用到:

  • "一轮"是什么? 一次 API 调用 → 模型返回 → 执行它要求的工具 → 把结果追加回 messages = 一轮rounds_since_todo 数的就是"模型有多少次机会调 todo 但没调"。
  • messages 是什么? 一个列表,每个元素是 {"role": "user/assistant", "content": ...}。模型每次看到的"上下文"就是这个列表的全部内容。对话越长,列表越长,排在前面的信息离注意力中心越远。

二、解决方案:模型自己管一个带状态的 todo 列表

加一个 todo 工具,让模型自己维护任务列表:

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> | Tools   |
| prompt |      |       |      | + todo  |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                          |
              +-----------+-----------+
              | TodoManager state     |
              | [x] #1: type hints   |
              | [>] #2: docstrings   |
              | [ ] #3: main guard   |
              +-----------------------+
                          |
              if 连续 3 轮没更新 todo:
                inject <reminder>

不是你替它规划,是它自己规划、自己更新、自己追踪进度

核心设计三要素:

  1. TodoManager — 带状态的任务管理器,存储 [ ] [>] [x] 三种状态,强制同时只有一个 in_progress
  2. Nag Reminder — 模型连续 3 轮不更新 todo,就在工具结果中注入 <reminder>Update your todos.</reminder>,追着模型问"你更新计划了吗"
  3. 自纠错回路 — TodoManager 校验失败时抛异常,异常变成 tool_result 返回给模型,模型看到后自行修正

每次模型更新 todo,完整进度表就出现在上下文最新的位置——注意力最强的地方。这就是 todo 解决注意力问题的核心机制:对话会遗忘,但 todo 列表反复刷新到最新位置,永远不会被淹没。


三、和 s02 相比,多了什么?

组件 s02 s03
工具数量 4 (bash/read/write/edit) 5 (+todo)
计划能力 TodoManager 类,存储带状态的任务列表
监督机制 nag reminder:连续 3 轮不更新 todo 就注入提醒
agent_loop 简单分发工具调用 多了 rounds_since_todo 计数器和 nag 注入

核心循环的骨架没变——while loop + 工具分发 + 结果回填,s03 只是在上面加了一层"计划层"。


四、TodoManager 类:模型的外部记忆

这是 s03 的核心新增代码,一个管理任务列表的简单状态机。

4.1 初始化

class TodoManager:
    def __init__(self):
        self.items = []      # 存储任务列表,每个元素是 {"id": str, "text": str, "status": str}

一个全局实例:

TODO = TodoManager()

4.2 update(items) — 校验并更新任务列表

def update(self, items: list) -> str:
    # --- 硬限制:最多 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)))  # id 不提供时自动分配 "1", "2"...

        # --- 字段校验 ---
        if not text:                                    # 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})

    # --- 核心约束:同一时间只能有一个 in_progress ---
    if in_progress_count > 1:
        raise ValueError("Only one task can be in_progress at a time")

    self.items = validated
    return self.render()  # 返回格式化后的任务列表给模型

设计意图

  • max 20:防止模型创建无穷无尽的列表。20 条足够覆盖任何合理任务,超过就是不合理的。
  • id 自动分配:模型可以不传 id,系统自动分配 “1”, “2”, “3”… 降低了模型的使用门槛。
  • status 默认 "pending":模型不指定状态时默认为未开始。
  • 只有一个 in_progress:这是 TodoManager 的核心约束。强制模型一次只专注一件事,避免"同时在做 5 件事,实际一件没做完"的虚假进度。
  • 校验失败抛异常:TodoManager 的 ValueError 会变成 tool_result 返回给模型,模型看到错误后会修正并重试——相当于一个自纠错回路。

4.3 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)

渲染效果示例:

[ ] #1: 添加 hello.py 的类型注解
[>] #2: 添加 utils.py 的类型注解     ← 当前正在做
[x] #3: 添加 main.py 的类型注解      ← 已完成

(1/3 completed)

这个渲染结果会作为 todo 工具的返回值嵌入到对话中,模型在下一轮推理时能看到。关键心理效应:模型看到 [x](已完成)会感到进度,看到 [>](进行中)知道当前焦点,看到 [ ](未开始)知道还剩什么——就像一个外部仪表盘。


五、todo 工具的定义与分发

todo 工具定义的核心信息就三条:

  • items 是一个数组,包含所有任务。模型一次性传入完整的任务列表,全量替换旧列表——不是"把 #2 改成 completed",而是"这是我现在的所有任务和状态"。模型每次更新都会重新审视整体进度,不会产生状态漂移。
  • 每个 item 有三个字段id(任务编号)、text(任务描述)、status"pending" / "in_progress" / "completed" 三选一)。
  • schema 要求必填,代码也做容错:schema 里声明 required: ["id", "text", "status"] 是给 LLM 看的提示;代码里 id 不传自动分配、status 不传默认 "pending",是防止模型偶尔漏传导致崩溃——双保险

工具如何分发到 TodoManager

TOOL_HANDLERS = {
    # ...s02 的 4 个工具...
    "todo": lambda **kw: TODO.update(kw["items"]),
}

这里用了一个 lambda,等价于:

def handle_todo(**kw):
    return TODO.update(kw["items"])

**kw 是 Python 的"关键字参数打包"——调用时传 todo(path="x", items=[...]),函数内部 kw 就变成 {"path": "x", "items": [...]},用 kw["items"] 取出任务列表。之所以用 lambda 而不直接写 TODO.update,是因为 TodoManager 的 update 接收的是 items 列表,而 Anthropic API 传来的是一整个 dict(包含所有参数),需要从中把 items 拆出来。

todo 和 bash/read/write/edit 地位完全平等,只是它的副作用不是操作文件系统,而是更新内存中的 TODO 对象。

你的 todo 去哪了?——完整调用链

模型调 todo({"items": [...]})
        ↓
TOOL_HANDLERS["todo"] → lambda **kw: TODO.update(kw["items"])
        ↓
TodoManager.update([...])    ← 校验 + 存储
        ↓
self.render()                ← 格式化为 "[ ] [>] [x]" 文本
        ↓
返回字符串 → tool_result → messages.append({"role": "user", ...})
        ↓
下一轮 API 调用时,messages 里最新的内容就是进度表

每一步都能在源码里找到对应位置。理解这个链路,就理解了"todo 为什么能解决注意力问题"——render() 返回的进度表出现在 messages 最新位置,每次更新都刷新到注意力最强的地方。

所以 system prompt 为什么变了?

模型从 TOOLS schema 能知道 todo 工具怎么调用(传 items 数组)。但不知道什么时候该用、怎么用才对

# s02:
"Use tools to solve tasks. Act, don't explain."

# s03 新增了三句话:
"Use the todo tool to plan multi-step tasks."  # 什么时候用:多步任务先列计划
"Mark in_progress before starting, completed when done."  # 怎么用:动手前标进行中,做完标完成
"Prefer tools over prose."  # 强化:别啰嗦,动手

工具是工具,使用规范是规范。就像给新员工一个项目管理系统,你还得告诉他"接了任务要标进行中,做完要标已完成"。


六、Agent Loop 的关键升级:Nag 提醒

6.1 计数器逻辑

def agent_loop(messages: list):
    rounds_since_todo = 0    # ← 新增:追踪多少轮没调用 todo
    while True:
        # ...调用 API...
        # ...检查 stop_reason...

        results = []
        used_todo = False    # ← 本轮是否用了 todo

        for block in response.content:
            if block.type == "tool_use":
                # ...执行工具...
                if block.name == "todo":
                    used_todo = True       # 本轮调了 todo,标记

        # --- 计数器更新 ---
        rounds_since_todo = 0 if used_todo else rounds_since_todo + 1
        # 如果这轮调了 todo → 计数器归零
        # 如果这轮没调 todo → 计数器 +1

6.2 Nag 注入

        if rounds_since_todo >= 3:
            results.append({
                "type": "text",
                "text": "<reminder>Update your todos.</reminder>"
            })

当连续 3 轮以上没调用 todo 时,在工具结果列表末尾追加一条提醒。这个提醒会作为 role="user" 消息的一部分发送给模型。

为什么是 3 轮?

  • 1 轮:过于敏感,可能模型下一个 tool_use 就想调 todo
  • 2 轮:还是有点紧
  • 3 轮:给模型足够的空间,但不会让它偏航太久
  • 这是一个调出来的经验值,没有硬道理

为什么放在 tool_result 列表里而不是单独一条消息?
追加到同一个 results 列表里,和真正的工具结果混在一起,作为同一条 role="user" 消息发送。这样做不增加消息轮次,自然融入上下文流。

6.3 完整的工具执行流程

        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}"       # ← s02 没有的 try/except
                print(f"> {block.name}:")
                print(str(output)[:200])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output)
                })
                if block.name == "todo":
                    used_todo = True

        # --- 计数器更新 ---
        rounds_since_todo = 0 if used_todo else rounds_since_todo + 1

        # --- Nag 注入 ---
        if rounds_since_todo >= 3:
            results.append({
                "type": "text",
                "text": "<reminder>Update your todos.</reminder>"
            })

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

注意这个 try/except——s02 没有。因为 TodoManager 的 update() 会主动抛 ValueError(超过 20 条、多个 in_progress 等),如果不 catch,整个 agent 就会崩溃。catch 后把错误信息作为 tool_result 返回给模型,模型看到错误会自我修正:

模型调用: todo(items=[..., {status: "in_progress"}, {status: "in_progress"}])
返回: Error: Only one task can be in_progress at a time
模型: 哦对,只能有一个 in_progress,我修正一下...

这就是自纠错回路:约束 → 违反 → 报错 → 模型修正 → 继续。


七、完整流程走读(以具体任务为例)

假设用户输入:"重构 hello.py:加类型注解、加 docstring、加 main guard"

第 1 轮

  1. 模型收到用户消息,system prompt 说要 “Use the todo tool to plan”
  2. 模型调用 todo 工具:
{
  "items": [
    {"id": "1", "text": "添加类型注解", "status": "pending"},
    {"id": "2", "text": "添加 docstring", "status": "pending"},
    {"id": "3", "text": "添加 __main__ guard", "status": "pending"}
  ]
}
  1. TodoManager 校验通过,存储,返回渲染结果:
[ ] #1: 添加类型注解
[ ] #2: 添加 docstring
[ ] #3: 添加 __main__ guard

(0/3 completed)
  1. rounds_since_todo 归零(因为本轮用了 todo)

第 2 轮

  1. 模型看到 todo 列表,决定开始第 1 个:

    • 调用 todo:把 #1 改为 in_progress
    • 调用 read_file:读取 hello.py 内容
  2. 两个工具都执行完毕,结果追加到 messages

第 3 轮

  1. 模型调用 edit_file:添加类型注解
  2. rounds_since_todo += 1(本轮没调 todo)

第 4 轮

  1. 模型调用 todo:把 #1 改为 completed,把 #2 改为 in_progress
  2. 模型调用 edit_file:添加 docstring
  3. rounds_since_todo 归零

…以此类推,直到 3 个任务全部完成

如果模型忘了(Nag 触发)

假设模型在改代码时过于投入,连续 3 轮没更新 todo:

第 3 轮没 todo → rounds_since_todo = 1
第 4 轮没 todo → rounds_since_todo = 2
第 5 轮没 todo → rounds_since_todo = 3 → 触发 nag!

第 5 轮的结果列表末尾被注入:

<reminder>Update your todos.</reminder>

模型下一轮看到这个提醒 → 停下来更新 todo → 重新聚焦。这就是 "外部问责压力"的工程化实现


八、设计洞察(为什么这样设计?)

8.1 模型的状态机 vs 代码的状态机

TodoManager 的三个状态 (pending → in_progress → completed) 非常简单,但正是这个简单让模型能正确使用它。如果状态太多(如 blockeddeferredreopened…),模型反而会困惑。简单的约束更容易被遵循

8.2 Nag 是软约束,不是硬中断

nag 不打断循环、不抛出异常、不拒绝执行。它只是一段文本,模型可以选择忽略(尽管通常不会)。这就是 “Harness” 层哲学:引导而非控制——给模型搭好轨道,但不替它开车。

8.3 强制单一 in_progress 不是限制,是保护

看起来限制了模型的多任务能力,实际上保护了它——在大模型中,“多任务并行” 往往意味着 “注意力分散,一件都做不好”。单一焦点 = 更可靠

8.4 整体覆盖 vs 增量修改

todo 工具接收的是完整的任务列表,不是 “把 #2 改成 completed”。这意味着:

  • 每次调用都是全量替换,不需要增量合并逻辑
  • 模型每次都必须重新审视整体进度,不会产生"忘了某个任务还在 in_progress 中"的状态漂移
  • 代码极其简单:self.items = validated

8.5 工具定义的双重约束

input_schema 里声明 required 字段和 enum 值——这是给LLM 看的,让模型知道怎么调用。但代码里也做了校验——这是给运行时看的,防止幻觉。两者不耦合:schema 改了不影响运行时校验,运行时校验挂了不影响 schema 语义。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐