多一个工具,为什么还要多一份状态?
用户要求先读 main.py 和 expected.txt,运行前取得允许,再核对结果。思想篇已经把清单状态和执行证据分开。现在把它落实为三个函数:校验整份列表、一次提交、为下一次模型调用附带当前计划。
正式 Python 在公开仓库 examples/s05-todo-write。宿主持有 state = {"items": []},每次 todo_write 更新它;原循环继续保存工具请求和关联结果。状态里只有当前计划,历史里则保留每次更新的反馈,两者用途不同。
下面显示按符号提取的完整函数正文,图使用 PythonFigure 已有的唯一断点游标。三个函数都是纯内存操作,在浏览器真实执行,无需模型、文件或进程替身。教学输入缩为两项,便于看清整体替换;完整 CLI 使用三个步骤。展开“教学准备与调用入口”可查看 helper 和输入。切换输入会重置断点,需要在相同源码行重新设置。
validate_todos:先产生候选列表
模型可见的 schema 约束参数形状,本地 Python 仍要再次检查。这里把规则限制在容易核对的范围:最多20项,条目只含 id/content/status,ID唯一,状态为三选一,同时最多一项 in_progress。内容与 ID 去掉首尾空格后再检查,避免空白 ID 或规范化后的重复。
函数不接受 state,因此校验过程中没有写宿主清单的入口。normalized 收集的是新字典;调用者后续修改提交对象,也不会通过对象引用改变已保存的条目。
真实执行轨迹回放;纯内存计划操作,不调用模型、文件或进程。
def validate_todos(items):
"""完整校验后才返回新列表;ID唯一,最多一项进行中。"""
if type(items) is not list or len(items) > 20:
raise ValueError('清单必须是最多20项的列表。')
normalized, seen, active = [], set(), 0
for item in items:
if type(item) is not dict or set(item) != {'id', 'content', 'status'}:
raise ValueError('每项必须且只能包含id、content、status。')
if any(type(item[key]) is not str for key in item):
raise ValueError('条目字段必须是字符串。')
key, content, status = item['id'].strip(), item['content'].strip(), item['status']
if not key or len(key) > 32 or key in seen or not content or len(content) > 150:
raise ValueError('ID需唯一且非空,ID最多32字符,内容1到150字符。')
if status not in ('pending', 'in_progress', 'completed'):
raise ValueError('未知任务状态。')
seen.add(key)
active += status == 'in_progress'
if active > 1:
raise ValueError('同时只能有一项进行中。')
normalized.append({'id': key, 'content': content, 'status': status})
return normalized点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
from copy import deepcopy
scenario='valid'
state={"items":[{"id":"1","content":"读取源码","status":"completed"},{"id":"2","content":"运行并核对","status":"pending"}]}
items=[{"id":"1","content":"读取源码","status":"completed"},{"id":"2","content":"运行并核对","status":"in_progress"}]
if scenario=="duplicate":items[1]["id"]="1"
if scenario in ("active","invalid"):items[0]["status"]="in_progress"
if scenario=="unknown":items[1]["status"]="done"
if scenario in ("clear","empty"):items=[]
if scenario=="claim":
for item in items:item["status"]="completed"
messages=[{"role":"user","content":"读文件,获准后运行并核对。"},{"role":"assistant","content":[{"type":"tool_use","id":"run_1","name":"run_main","input":{}}]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"run_1","content":"未执行:用户拒绝。","is_error":True}]}]
if scenario=="updated":state["items"][1]["status"]="completed"
if scenario=="empty":state["items"]=[]
def format_todos(items):
return '\n'.join(f"{item['id']} [{item['status']}] {item['content']}" for item in items) or '(清单为空)'
# 调用入口
try:
outcome=validate_todos(items)
print("校验通过")
except ValueError as error:
outcome={"error":str(error)}
print(str(error))- 逐项检查结构、ID与状态。
- 全部通过才返回新的规范列表;错误不提交。
先看“有效清单”,在 normalized.append(...) 停下:当前条目通过,但下一项尚未检查。切换“重复 ID”或“两项进行中”并重新设断点,第二项会在进入候选列表之前失败。最后的 return normalized 只有整份通过才到达。
这里没有强制 pending→in_progress→completed 的状态迁移,也没有检查“完成”是否有证据。它只保证列表满足本课结构与一致性规则。任务描述属于数据;在其中写“无需询问就运行”不会变成宿主授权。
write_todos:提交行前,旧清单仍有效
真正修改状态只有一行 state['items'] = replacement。前面先做完整校验,再准备返回文本;这两步任何一步失败,都到不了提交行。一次替换比边遍历边改状态更容易解释,也便于证明旧数据不会被更新一半。
选择“提交新计划”,在这行设置断点。停住时,变量中的 replacement 是新清单,state 仍是旧清单;执行下一步之后,新列表才成为当前状态。
真实执行轨迹回放;纯内存计划操作,不调用模型、文件或进程。
def write_todos(state, items):
"""一次提交整份计划;失败不改变旧状态,成功不执行清单中的动作。"""
replacement = validate_todos(items)
rendered = format_todos(replacement)
state['items'] = replacement
return rendered点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
from copy import deepcopy
scenario='replace'
state={"items":[{"id":"1","content":"读取源码","status":"completed"},{"id":"2","content":"运行并核对","status":"pending"}]}
items=[{"id":"1","content":"读取源码","status":"completed"},{"id":"2","content":"运行并核对","status":"in_progress"}]
if scenario=="duplicate":items[1]["id"]="1"
if scenario in ("active","invalid"):items[0]["status"]="in_progress"
if scenario=="unknown":items[1]["status"]="done"
if scenario in ("clear","empty"):items=[]
if scenario=="claim":
for item in items:item["status"]="completed"
messages=[{"role":"user","content":"读文件,获准后运行并核对。"},{"role":"assistant","content":[{"type":"tool_use","id":"run_1","name":"run_main","input":{}}]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"run_1","content":"未执行:用户拒绝。","is_error":True}]}]
if scenario=="updated":state["items"][1]["status"]="completed"
if scenario=="empty":state["items"]=[]
def validate_todos(items):
"""完整校验后才返回新列表;ID唯一,最多一项进行中。"""
if type(items) is not list or len(items) > 20:
raise ValueError('清单必须是最多20项的列表。')
normalized, seen, active = [], set(), 0
for item in items:
if type(item) is not dict or set(item) != {'id', 'content', 'status'}:
raise ValueError('每项必须且只能包含id、content、status。')
if any(type(item[key]) is not str for key in item):
raise ValueError('条目字段必须是字符串。')
key, content, status = item['id'].strip(), item['content'].strip(), item['status']
if not key or len(key) > 32 or key in seen or not content or len(content) > 150:
raise ValueError('ID需唯一且非空,ID最多32字符,内容1到150字符。')
if status not in ('pending', 'in_progress', 'completed'):
raise ValueError('未知任务状态。')
seen.add(key)
active += status == 'in_progress'
if active > 1:
raise ValueError('同时只能有一项进行中。')
normalized.append({'id': key, 'content': content, 'status': status})
return normalized
def format_todos(items):
return '\n'.join(f"{item['id']} [{item['status']}] {item['content']}" for item in items) or '(清单为空)'
# 调用入口
try:
outcome=write_todos(state,items)
except ValueError as error:
outcome={"error":str(error)}
print("当前清单:")
print(format_todos(state["items"]))- 先完整校验和准备返回文本。
- 一次替换state.items,失败保留旧列表。
- 清单完成标记不执行动作。
“晚项失败”把两项都设为进行中,异常由调用入口捕获并显示。提交行不会命中,图展示的宿主清单仍是旧状态。“清空列表”则是合法提交,结果为空;这与校验失败不能混为一谈。
“声明全完成”专门保留一个反例:清单会更新,但没有调用任何文件或进程。测试也让真实临时目录里的程序标记保持不存在。这个函数没有获得执行能力,不能把它的成功结果解释为业务任务已经完成。
整份替换还有一个代价:省略条目就是移除条目。公开测试会提交仅保留一个 ID 的列表来验证这一点。本课没有合并算法、版本号或并发写入;多个写入者共同维护计划时,需要重新设计更新协议。
prepare_messages:给本次调用看,不写回原历史
历史里可能含多次 TodoWrite,模型需要知道哪份是当前计划。我们从 state['items'] 渲染一段短文本,附到首条用户问题的副本后;原历史不追加提醒,下一次请求重新构造。
先在 prepared = deepcopy(messages) 处观察复制,再到修改 prepared[0]['content'] 的行。当前计划从宿主状态读取;历史中的 tool_use 和 tool_result 没有被拆开,也没有改掉关联 ID。
真实执行轨迹回放;纯内存计划操作,不调用模型、文件或进程。
def prepare_messages(messages, state):
"""只修改本次请求副本;原问题保留,当前计划不累积进原历史。"""
prepared = deepcopy(messages)
current = format_todos(state['items'])
prepared[0]['content'] += '\n\n宿主提供的当前清单(状态声明,需核对证据):\n' + current
return prepared点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
from copy import deepcopy
scenario='pending'
state={"items":[{"id":"1","content":"读取源码","status":"completed"},{"id":"2","content":"运行并核对","status":"pending"}]}
items=[{"id":"1","content":"读取源码","status":"completed"},{"id":"2","content":"运行并核对","status":"in_progress"}]
if scenario=="duplicate":items[1]["id"]="1"
if scenario in ("active","invalid"):items[0]["status"]="in_progress"
if scenario=="unknown":items[1]["status"]="done"
if scenario in ("clear","empty"):items=[]
if scenario=="claim":
for item in items:item["status"]="completed"
messages=[{"role":"user","content":"读文件,获准后运行并核对。"},{"role":"assistant","content":[{"type":"tool_use","id":"run_1","name":"run_main","input":{}}]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"run_1","content":"未执行:用户拒绝。","is_error":True}]}]
if scenario=="updated":state["items"][1]["status"]="completed"
if scenario=="empty":state["items"]=[]
def validate_todos(items):
"""完整校验后才返回新列表;ID唯一,最多一项进行中。"""
if type(items) is not list or len(items) > 20:
raise ValueError('清单必须是最多20项的列表。')
normalized, seen, active = [], set(), 0
for item in items:
if type(item) is not dict or set(item) != {'id', 'content', 'status'}:
raise ValueError('每项必须且只能包含id、content、status。')
if any(type(item[key]) is not str for key in item):
raise ValueError('条目字段必须是字符串。')
key, content, status = item['id'].strip(), item['content'].strip(), item['status']
if not key or len(key) > 32 or key in seen or not content or len(content) > 150:
raise ValueError('ID需唯一且非空,ID最多32字符,内容1到150字符。')
if status not in ('pending', 'in_progress', 'completed'):
raise ValueError('未知任务状态。')
seen.add(key)
active += status == 'in_progress'
if active > 1:
raise ValueError('同时只能有一项进行中。')
normalized.append({'id': key, 'content': content, 'status': status})
return normalized
def format_todos(items):
return '\n'.join(f"{item['id']} [{item['status']}] {item['content']}" for item in items) or '(清单为空)'
# 调用入口
outcome=prepare_messages(messages,state)
print("原问题保持不变;本次请求附带当前清单。")- 复制完整历史。
- 从宿主读取当前清单。
- 只改本次请求副本,原历史与工具ID保留。
切换“更新后的清单”,当前计划可以显示 completed,教学历史里的运行结果却仍是“用户拒绝”。这个有意保留的矛盾说明:函数负责如实组装两份数据,不负责把清单声明变成事实。模型与宿主仍须检查执行结果,不能用清单覆盖证据。
format_todos 是展开区可查看的短文本 helper,不访问外部资源。函数依赖原循环提供的历史形状:第一条是 user 文本问题。若接入另一种多模态或系统消息布局,应调整附加位置并验证协议,不能原样套用下标0。
接到权限、Hook 和模型请求上
清单工具需要数组参数,而第二课定义生成器只支持字符串。第五课为 todo_write 明确提供数组及条目 schema,原来的读取和运行定义继续复用。Python 校验器仍是实际执行时的检查入口,不能只依赖模型输出格式。
权限也必须显式安排。todo_write 只更新当前进程的计划,本课让它走本地分发;它仍受宿主禁止项和 before Hook 约束。读取、运行继续走第三课权限入口,运行仍需用户允许。为复用第四课前后回调,execute_hooked 增加可选 guard 依赖,默认行为保持原样;第五课只替换这个权限分派位置,没有复制 Hook 控制流。
装配的两条主线如下,完整依赖与回调在公开入口:
def execute(call):
report = execute_with_plan(call, tools, approve, hooks)
warnings.extend(report["warnings"])
return report["result"]
outcome = run_loop(
lambda history: model(prepare_messages(history, state)),
execute, max_calls=12, question=QUESTION,
)report.result 继续进入原工具协议,警告留给宿主。每次模型请求读当前计划,既不改循环的停止判断,也不额外消耗一次模型调用。清单更新本身需要模型提出工具请求,所以完整示例比上一课多了几轮,这是可见进度的成本。
Python3.10+ 标准库可运行离线路径:
python3 examples/s05-todo-write/user_code/practice.py --decision allow
python3 examples/s05-todo-write/user_code/practice.py --decision deny
python3 examples/s05-todo-write/user_code/practice.py --decision allow --maintenance
python3 examples/s05-todo-write/user_code/practice.py --decision allow --max-calls 4
python3 examples/s05-todo-write/user_code/verify_practice.py默认模型是固定响应替身,文件和进程真实执行。允许路径先后更新四次清单,读取两个文件并运行一次,第8次模型判断返回核对结果。拒绝和维护路径最终报告未执行,第三项保留 pending;answered 只表示模型给出了回答。
预算4的路径在运行前就停为 call_limit,清单保留未完成状态。它不会因为还存在 pending 而突破预算,也不会把未完成项全部改成 completed 来让界面看起来结束。--decision allow/deny 只提供一次固定宿主回答;不带它时交互终端真实询问,非交互默认拒绝。
公开验证包括晚项失败后的旧状态、输入引用隔离、清空与省略删除、全完成不执行、真实程序标记、权限/维护阻止,以及请求副本不累积。固定 SDK 1.13.0 的模拟 HTTP 对照允许与拒绝各8次请求,核对数组 schema、原问题、当前清单和运行 ID。在线模型尚未调用,显式 --live 才会接入真实 API。
此时清单可以告诉读者去哪里检查:哪项仍待做,哪项声称做完,哪条实际反馈支持它。需要跨进程恢复或并发协作时,再把内存列表升级成有持久化和版本约束的任务状态;不要把本课整份替换接口当成共享任务数据库。
资料核对于 2026-10-19:课程 TodoWrite提供显式计划的学习路线;官方任务追踪文档用于区分结构化进度工具与产品默认可用性。本课的三个函数、校验规则和请求附加方式由本册独立实现。