面试03-TodoManager解决 Agent步骤混乱的问题
一、核心问题:多步骤导致模型「失忆/失控」
在长流程、多步骤任务中(比如10步代码重构、多文件批量处理),AI模型会出现核心问题:
- 上下文过载 :工具执行结果、对话历史不断填充上下文,系统提示的核心要求被「稀释」(
注意力稀疏),模型逐渐忘记整体任务目标; - 步骤混乱 :(1)由于对
Query的意图稀释,导致重复执行已完成的步骤(比如反复创建同一个文件);(2)要么跳过关键步骤(比如没验证文件内容就直接下一步),甚至「走神」即兴发挥(偏离原始任务);((2)和(3)一般需要DAG 进行处理) - 进度无跟踪 :模型没有结构化的方式记录「哪些做了、哪些在做、哪些没做」,全靠上下文记忆,长流程下必然出错。
如何保证模型的多步骤是有意义
通过状态管理器 TodoManager,管理任务的状态,使得模型在同一时间只能执行一个任务,并将任务的执行状态进行显示(知道自己在做什么);
多步骤任务 (items) 是什么样子?
本质: 一个「待办项字典组成的列表」,是用户传入的原始待办数据。如以下所示:
input_items = [
{"id": "1", "text": "买早餐", "status": "pending"},
{"id": "2", "text": "写代码", "status": "in_progress"},
{"id": "3", "text": "整理文档", "status": "completed"}
]
如何处理任务?
任务管理器的设置,其核心设计理念是「 专注单任务 」(类似于对任务上锁),所以对 in_progress 状态做了多重特殊处理,原因如下:
- 避免多任务并行混乱 :人同时处理多个任务容易分心、效率低下,这是待办管理的核心诉求(约束 :最多
1个进行中任务);——>大模型同时处理两个任务带来的两个问题:(1)成本方面,是增加上下文Token,Prefill阶段的时候显存消耗过大,耗时间;(1)效果方面:模型因为上下文的累积导致核心 Query 意图被削弱,进而输出质量变差;
二、核心解决方案:TodoManager + 提醒机制
代码的核心是通过「结构化进度跟踪 + 强制提醒」,让模型始终锚定任务目标,核心分为两部分:
1. TodoManager:给模型一个「可视化的任务清单」
TodoManager 是一个 结构化的任务状态管理器,本质是让模型把 多步骤任务 拆分成 可跟踪的待办项,并强制规范状态,核心逻辑:
- 任务状态约束:每个待办项只能是
pending(待做)、in_progress(进行中)、completed(已完成)三种状态,且同一时间只能有1个「进行中」任务(避免模型同时做多个事导致混乱); - 结构化更新:模型必须通过
todo工具调用TodoManager.update()来更新任务清单,代码会校验任务格式(比如不能为空、状态合法),确保清单规范; - 可视化渲染:任务清单会以固定格式(
[ ] 待做 | [>] 进行中 | [x] 已完成)返回给模型,让模型清晰看到「已完成/总任务数」,锚定整体进度。
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)))
# 约束2:待办文本不能为空
if not text:
raise ValueError(f"Item {item_id}: text required")
# 约束3:状态只能是 pending/in_progress/completed 三者之一
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})
# 约束4:同一时间只能有1个进行中的任务(避免并行混乱)
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()
核心作用: 通过硬约束(状态、数量、文本)强制模型生成规范的任务清单,并用可视化格式返回,让模型清晰看到进度。
2. TodoManager 代码流程阐述:
1. 类定义
class TodoManager:
def __init__(self):
self.items = [] # 存储结构化的待办项
- 代码作用 :定义
TodoManager类(待办事项管理器),init 是构造函数,初始化时创建空列表 self.items。 - 设计意图 :
self.items是整个类的核心数据容器,专门存储「结构化的待办项」,避免数据零散。
2. update 方法(核心:校验 + 更新待办)
(1)
def update(self, items: list) -> str:
# 约束1:最多20个待办,避免清单过长
if len(items) > 20:
raise ValueError("Max 20 todos allowed")
- 代码作用 :先校验待办总数,超过 20 个直接抛异常。
- 设计意图 :防止清单过长导致管理混乱。
(2)
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)))
- 设计意图 :
- 字段提取时做「默认值(空字符串) + 类型转换(小写) + 格式化」,避免原始数据格式混乱(比如状态有大写
In_Progress、ID是数字等); - 单独用
validated存储校验后的项,保证self.items里的每一项都是「干净、合规」的。
(3)
- 字段提取时做「默认值(空字符串) + 类型转换(小写) + 格式化」,避免原始数据格式混乱(比如状态有大写
# 约束2:待办文本不能为空
if not text:
raise ValueError(f"Item {item_id}: text required")
# 约束3:状态只能是 pending/in_progress/completed 三者之一
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})
- 设计意图 :
文本非空&状态枚举:确保待办项有实际内容,避免「空待办」;统计进行中任务:为后续「单任务并行」约束做准备。统一待办项的结构:必须包含 id / text / status 三个键
(4)
# 约束4:同一时间只能有1个进行中的任务(避免并行混乱)
if in_progress_count > 1:
raise ValueError("Only one task can be in_progress at a time")
self.items = validated
return self.render() # 渲染成可视化清单返回给模型
设计意图: 限制「同时只能有 1 个进行中的任务」,符合「专注单任务」的待办管理理念,避免用户同时处理多个任务导致效率低下。
2. 提醒机制:防止模型「忘记更新清单」
模型可能会偷懒 / 忘记调用 todo 工具更新进度,因此代码加入「强制提醒逻辑」:
- 维护
rounds_since_todo计数器:记录模型连续多少轮工具调用没更新todo清单; - 阈值触发提醒:当计数器 ≥
3(连续3轮没更清单),代码会在工具结果中插入<reminder>Update your todos.</reminder>提醒; - 提醒注入上下文:这个提醒会作为工具结果的一部分喂回模型,强制模型回到「更新任务清单→确认进度」的轨道,避免偏离。
Code2 注册 todo 工具(模型能调用的「进度更新入口」):
# 1. 工具映射:把 todo 工具名关联到 TodoManager.update 方法
TOOL_HANDLERS = {
# ... 其他工具(bash/read/write/edit)
"todo": lambda **kw: TODO.update(kw["items"]), # 核心:模型调用todo工具时,执行update
}
# 2. 工具定义:告诉模型怎么调用todo工具(参数规范)
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"]
}},
]
Code3 提醒机制(防止模型忘记更新 todo)
def agent_loop(messages: list):
# 计数器:记录连续多少轮没调用todo工具
rounds_since_todo = 0
# 循环:模型生成 response 和调用 tools 的 loop
while True:
# 1 生成 response
response = client.messages.create(
model=MODEL, system=SYSTEM, messages=messages,
tools=TOOLS, max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
# 2 不需要使用 tool,说明该返回整个 response
if response.stop_reason != "tool_use":
return
results = []
used_todo = False # 标记本轮是否调用了 todo 工具
# 3 遍历 response 中的内容
for block in response.content:
# 3.1 使用工具
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}: {str(output)[:200]}")
# 3.2 添加 tool result
results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
# 3.3 使用工具了,需要更新任务状态
if block.name == "todo":
used_todo = True
# 3.4 更新 todo 工具状态
rounds_since_todo = 0 if used_todo else rounds_since_todo + 1
# 3.5 边界:如果三次没用 todo 工具,则插入提示词,强制模型在下次输出时使用 todo 工具
if rounds_since_todo >= 3:
results.insert(0, {"type": "text", "text": "<reminder>Update your todos.</reminder>"})
# 3.6 把结果(含提醒)喂回模型
messages.append({"role": "user", "content": results})
三、核心执行逻辑(工具调用闭环)
- 任务初始化:用户输入多步骤任务后,模型首先调用
todo工具,拆分任务为结构化待办项(比如「[ ] 创建test.txt | [ ] 写入内容 | [ ] 验证内容」); - 步骤执行+进度更新:每完成一个步骤,模型必须调用
todo工具更新对应任务状态(比如把「创建test.txt」标为completed,把「写入内容」标为in_progress); - 提醒兜底:如果模型连续3轮没更新todo,代码自动插入提醒,强制模型先更新清单再继续;
- 进度可视化:
TodoManager.render()会把当前进度(比如「2/3 completed」)返回给模型,让模型始终知道「做到哪了、还剩啥要做」。
总结
核心关键点:
- 结构化跟踪:用
TodoManager将模糊的「多步骤任务」转化为模型可识别的结构化清单,替代不可靠的上下文记忆; - 强制约束:通过工具调用+格式校验,确保模型必须更新进度,避免「口头规划、实际失控」;
- 提醒兜底:用计数器+强制提醒,解决模型「忘记更新清单」的问题,始终锚定整体任务目标。
简单来说,这套方案相当于给模型配了一个「任务看板」,并在它走神时拍醒它,确保长流程任务的每一步都可控、可追溯。
带来的问题:
- 有些工具执行的结果可能并不能满足任务的需求,可能需要涉及
回滚、回滚等操作; - 有些工具可能需要重复执行;
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)