tracking-task-progress-with-todos

能力目标

给任意一个多步任务装上可观测的进度：让智能体在开工前写出待办清单、执行中逐条翻转状态，代码这边随时把清单取出来渲染成进度条、写进数据库或推给前端，并在收尾时检测出「清单全勾完却没给最终答案」这一失败形态。

前置

• 已能用 create_deep_agent 构造智能体并读懂 result["messages"]（见 bootstrapping-deepagents-env）。
• 规划能力由框架自动挂载，不需要你手动导入或传入任何中间件；提供的工具叫 write_todos。

实操流程

1. 构造智能体，不必传任何业务工具——规划能力是框架自动挂上的：

   from deepagents import create_deep_agent
   
   agent = create_deep_agent(
       model=model,
       tools=[],
       system_prompt="你是一个能规划任务的 AI 助手。",
   )
   

2. 按稳定触发的写法组装任务提示。规划不是免费的，工具说明里写着「请求不到 3 步就别用这个工具、直接做」，触发与否是模型的判断而非硬阈值。要稳定触发就两件事：提示里明写「先规划再执行」，任务确实有足够步骤：

   task_items = ["收集 A 主题要点", "收集 B 主题要点", "收集 C 主题要点", "对比 A/B/C 并汇总"]
   user_prompt = "请完成以下需要分步进行的任务，先规划再执行：\n" + "\n".join(f"- {t}" for t in task_items)
   
   result = agent.invoke({"messages": [{"role": "user", "content": user_prompt}]})
   

3. 取最终清单。它挂在状态的 todos 字段上，是一个列表、每项含 content 与 status 两个字段。首次调用 write_todos 之前这个字段根本不存在，所以用 .get 取、不要用下标：

   todos = result.get("todos", [])
   for t in todos:
       print(t["content"], t["status"])
   

4. 渲染进度。这段模板与任务领域无关，换任何多步任务都不用改：

   for t in todos:
       icon = "☑" if t["status"] == "completed" else "▶" if t["status"] == "in_progress" else "☐"
       print(f"{icon} {t['content']} [{t['status']}]")
   

5. 要实时进度而不只是终态，就从消息历史里把每一次 write_todos 调用的清单快照按顺序抽出来。注意模型不保证每轮都更新清单，所以按调用次数计序、不要假设「每轮回复对应一次更新」：

   from langchain_core.messages import AIMessage
   
   snapshots = []
   for msg in result["messages"]:
       if isinstance(msg, AIMessage) and msg.tool_calls:
           for tc in msg.tool_calls:
               if tc.get("name") == "write_todos":
                   snapshots.append(tc["args"]["todos"])
   
   for snap in snapshots:
       done = sum(1 for t in snap if t["status"] == "completed")
       total = len(snap)
       pct = int(done / total * 100) if total else 0
       bar = "█" * (done * 5) + "░" * ((total - done) * 5)
       print(f"[{bar}] {pct}%")
   

   两条取数路径别混：工具调用参数里拿到的是「每次写入的快照」，状态字段里拿到的是「最新一版」。

6. 收尾兜底，检测「标完成但无答案」。工具说明明确警告过这一点：write_todos 只追踪进度、不交付答案，把最后一条标成完成本身不等于回答了用户：

   from langchain_core.messages import AIMessage, ToolMessage
   
   msgs = result["messages"]
   last_todos_idx = max(
       i for i, m in enumerate(msgs)
       if isinstance(m, ToolMessage) and "Updated todo" in m.content
   )
   has_answer = any(
       isinstance(m, AIMessage) and m.content and not m.tool_calls
       for m in msgs[last_todos_idx + 1:]
   )
   # has_answer 为 False → 命中失败形态，追一轮让模型给出最终答案
   

校验回路

• result.get("todos", []) 长度 ≥3 且每项都带 status 字段，说明规划真的被触发并落进了状态。
• 消息历史里每一次 write_todos 调用后都跟着一条内容以 Updated todo list to 开头的工具回执消息。
• 进度快照序列从低百分比推进到 100%，且末尾快照里各项状态为 completed。
• 第 6 步的 has_answer 为 True。为 False 就补一轮，否则你拿到的是一份勾满的清单加一个空答案。

常见陷阱

• 任务「看起来像 3 步」却不触发规划：像「解释某协议、给一个示例、总结三条原则」这类中等复杂度任务，模型常常一口气答完、一次都不调 write_todos。稳定触发靠提示里明写「先规划再执行」加真实的步骤数，不要指望步数目测。
• 用下标取 result["todos"]：该字段是可选的，规划未触发时不存在，会抛 KeyError。一律 .get("todos", [])。
• 以为模型是读状态字段来记住计划：这个字段带着「只写出、不读入模型」的屏蔽标记，模型看不到它。模型靠的是消息历史里那条 Updated todo list to ... 的工具回执文本。这意味着：如果你自己改写或裁剪了消息历史，模型对计划的记忆也会跟着丢。
• 想做增量更新只改某一条：清单字段没有累加归并函数，走的是最后写入者获胜的覆盖语义，每次写入都是整张替换。也正因为覆盖会产生时序竞争，框架强制每轮最多调用一次 write_todos，并发调用会被注入一条错误回执要求重试。
• 拿「最终回复变短」当规划变差的证据：带规划的执行是多轮分布式产出，详细内容分散在中间各轮回复里、最后一条往往只是精炼总结。要对比效果就统计全部回复文本的合集或子任务关键词覆盖率，别比最后一条的长度。
• 默认规划总是更好：对落在模型能力范围内的结构化任务，带不带规划的子任务覆盖率可能完全相同。规划的确定收益是「结构化进度可见」，不是「答得更全」。