1. 为什么多步工具调用总在第三步断掉Agent 执行框架里最让人头疼的不是模型不够聪明而是任务跑到一半断了你根本不知道它断在哪、为什么断、能不能接着跑。我试过用最朴素的方式让 Agent 连续调用五六个工具完成一个部署流程结果它在第三步调用 Docker 命令时超时整个链路直接崩掉前面两步的中间产物全丢了只能从头再来。这种体验就像你写代码写到一半 IDE 突然崩溃而且没有自动保存。这个问题的根源在于大多数 Agent 执行框架把「一次对话」当成了执行单元而不是把「一个任务」当成执行单元。对话是无状态的、易失的而任务需要状态、需要持久化、需要证据。当工具调用跨越多个步骤、涉及多个外部系统时任何一步的失败都会让整个链路失去上下文。更麻烦的是你事后想复盘都无从下手因为日志是散的状态是内存里的证据是没有的。TOE-DAC 要解决的就是这个问题。它把 Agent 的执行过程建模成一个显式的状态机每个阶段的状态都落盘到state.json每个动作的结果都写入证据链每次会话都留下可回放的轨迹。这样即使执行中断你也能从最后一个稳定状态恢复而不是从零开始。它适合那些需要多步工具调用、执行周期长、对可追溯性有要求的场景比如自动化部署、数据管道编排、多阶段审批流。核心检索词这里先明确TOE-DAC 是一种面向长周期 Agent 任务的可持久化分层控制协议它通过状态机建模和证据链落盘让执行链路可回放、可续跑。你不需要一开始就理解全部六个阶段先理解「状态要落盘、动作要留证据、失败要能恢复」这三件事就够了。我踩过的坑是早期版本我把状态存在内存里用一个全局字典维护结果进程一重启全没了。后来改成每次状态变更都写state.json虽然多了磁盘 IO但换来了真正的可恢复性。这个取舍在长周期任务里是值得的。下面我会从工程落地的角度给出config.toml和settings.json的可复制配置骨架演示异常恢复的验证动作并对照真实报错给出排查路径。目标很明确让你能照着搭出一个最小可运行的 TOE-DAC 骨架跑通一次「中断-恢复-续跑」的完整流程。2. TaoToken 前置准备与 config.toml 骨架在动手写状态机之前你需要一个稳定的模型接入点。TOE-DAC 的 Decide 阶段需要模型根据 Target 和 Observe 的结果生成行动图这个调用会频繁发生所以接入点的稳定性和计费透明度很重要。TaoToken 在这里扮演的是统一接入层的角色它提供兼容 OpenAI 风格的 API你可以在一个地方管理多个模型的调用不用为每个模型单独维护一套鉴权和计费逻辑。先拿到 API Key。访问 https://taotoken.net/api-keys 创建密钥注意这个 Key 只在创建时显示一次复制后妥善保存。然后你需要确认接入地址API 端点是 https://taotoken.net/api这个地址不带任何查询参数直接作为 Base URL 使用。如果你需要查看完整的接入文档和参数说明可以访问 https://taotoken.net/doc。现在给出config.toml的骨架。这个文件放在项目根目录负责定义 TOE-DAC 实例的全局参数包括模型接入、预算约束、状态存储路径和证据链配置。# config.toml - TOE-DAC 全局配置骨架 [model] # TaoToken 统一接入地址 base_url https://taotoken.net/api api_key sk-your-key-here # Decide 阶段使用的模型建议选推理能力强的 decide_model gpt-4o # Act 阶段执行原子动作时使用的模型可选更轻量的 act_model gpt-4o-mini # 单次请求超时秒 timeout 60 # 失败重试次数网络层面 max_retries 3 [budget] # 单个 TD 实例的最大 token 消耗 max_tokens_per_td 200000 # 单个 TD 实例的最大执行时长秒 max_duration_seconds 3600 # 单个 action 的最大重试次数业务层面 max_action_retries 5 # 超过预算后的行为pause / escalate / abort on_budget_exceeded escalate [storage] # 用户线头根目录 user_thread_root ./threads # 状态文件路径模板 state_file {td_id}/state.json # 事件日志面向用户 event_log {td_id}/event.log # 操作日志面向 agent opr_log {td_id}/opr.log # 轨迹目录 trace_dir {td_id}/trace # 产物目录 artifacts_dir {td_id}/artifacts [evidence] # 证据链落盘格式jsonl / json format jsonl # 是否启用截图证据 enable_screenshot false # 证据文件路径 evidence_file {td_id}/artifacts/evidence.jsonl [recovery] # 是否启用自动恢复 auto_resume true # 恢复时从哪个状态开始last_stable / last_checkpoint resume_from last_stable # 检查点间隔每 N 个 action 写一次检查点 checkpoint_interval 3这个配置里几个关键点需要解释。base_url指向 TaoToken 的 API 端点所有模型调用都走这里你不需要为每个模型单独配置地址。budget段定义了预算约束这是 TOE-DAC 防止 Agent 无限循环的重要机制on_budget_exceeded设为escalate表示超预算时升级给人类确认而不是直接中断。storage段定义了文件结构每个 TD 实例有独立的目录包含状态、日志、轨迹和产物。recovery段的checkpoint_interval控制检查点频率设为 3 表示每执行 3 个 action 写一次检查点这样恢复时最多丢失 3 个 action 的进度。接下来是settings.json这个文件定义状态机的转换规则和证据链的字段结构。它和config.toml的分工是config.toml管全局参数settings.json管状态机逻辑。{ state_machine: { initial_state: TARGET, states: [TARGET, OBSERVE, ESTIMATE, DECIDE, ACT, CHECK, DONE, FAILED, PAUSED], transitions: [ { from: TARGET, to: OBSERVE, condition: target_defined }, { from: OBSERVE, to: ESTIMATE, condition: facts_collected }, { from: ESTIMATE, to: DECIDE, condition: feasible }, { from: ESTIMATE, to: PAUSED, condition: infeasible_or_uncertain }, { from: DECIDE, to: ACT, condition: plan_ready }, { from: ACT, to: CHECK, condition: action_executed }, { from: CHECK, to: ACT, condition: action_check_failed_and_retryable }, { from: CHECK, to: DECIDE, condition: target_check_failed_and_replan }, { from: CHECK, to: DONE, condition: target_check_passed }, { from: CHECK, to: FAILED, condition: unrecoverable_error }, { from: PAUSED, to: TARGET, condition: human_resumed } ] }, evidence_schema: { required_fields: [event_id, td_id, session_id, timestamp, stage, action, result_status, duration_ms], optional_fields: [parent_id, user_thread, error_type, retry_count, artifact_path], result_status_values: [success, failed, timeout, skipped, escalated] }, exception_handling: { target_fail: { action: human_interrupt, retry: false }, observe_fail: { action: agent_simple_retry, retry: true, max_retries: 3 }, estimate_fail: { action: agent_auto_retry, retry: true, max_retries: 2 }, decide_fail: { action: agent_auto_retry, retry: true, max_retries: 2 }, act_fail: { action: agent_simple_retry, retry: true, max_retries: 5 }, check_fail: { action: agent_auto_retry, retry: true, max_retries: 3 } } }settings.json里的transitions定义了状态机的完整转换图。注意CHECK状态有四个出边如果 action_check 失败且可重试回到ACT如果 target_check 失败需要重新规划回到DECIDE如果 target_check 通过进入DONE如果遇到不可恢复错误进入FAILED。这个设计让 Agent 在失败时不是简单重试而是根据失败类型选择不同的恢复路径。evidence_schema定义了证据链的字段结构required_fields是每条证据必须包含的optional_fields是可选补充的。exception_handling定义了每个阶段失败时的处理策略比如target_fail直接升级给人类因为目标定义错了后面全错act_fail允许 Agent 自己重试最多 5 次。这两个文件放在项目根目录TOE-DAC 实例启动时会先读config.toml拿全局参数再读settings.json拿状态机规则。如果你用的是 Claude Code 或类似的编码 Agent可以把这两个文件放在工作区根目录然后在对话里让 Agent 读取并初始化 TD 实例。3. 可复制配置与状态机落盘实现配置骨架有了现在需要把状态机真正跑起来。这一节给出一个最小可运行的 Python 实现它读取上面的config.toml和settings.json初始化一个 TD 实例执行状态转换并把状态和证据落盘。你可以直接复制这段代码到td_runner.py然后运行。# td_runner.py - TOE-DAC 最小可运行实现 import json import os import time import uuid import tomllib from datetime import datetime, timezone from pathlib import Path class TDInstance: def __init__(self, config_pathconfig.toml, settings_pathsettings.json, td_idNone): with open(config_path, rb) as f: self.config tomllib.load(f) with open(settings_path, r, encodingutf-8) as f: self.settings json.load(f) self.td_id td_id or ftd-{uuid.uuid4().hex[:8]} self.state self.settings[state_machine][initial_state] self.session_id fsess-{uuid.uuid4().hex[:8]} self.parent_id None self.user_thread None self.retry_count 0 self.action_index 0 # 初始化目录结构 self.td_dir Path(self.config[storage][user_thread_root]) / self.td_id self.td_dir.mkdir(parentsTrue, exist_okTrue) (self.td_dir / trace).mkdir(exist_okTrue) (self.td_dir / artifacts).mkdir(exist_okTrue) self.state_file self.td_dir / state.json self.event_log self.td_dir / event.log self.opr_log self.td_dir / opr.log self.evidence_file self.td_dir / artifacts / evidence.jsonl self._persist_state() def _persist_state(self): state_data { td_id: self.td_id, session_id: self.session_id, parent_id: self.parent_id, user_thread: self.user_thread, current_state: self.state, retry_count: self.retry_count, action_index: self.action_index, updated_at: datetime.now(timezone.utc).isoformat() } with open(self.state_file, w, encodingutf-8) as f: json.dump(state_data, f, ensure_asciiFalse, indent2) def _log_event(self, message): ts datetime.now(timezone.utc).isoformat() with open(self.event_log, a, encodingutf-8) as f: f.write(f[{ts}] {message}\n) def _log_opr(self, message): ts datetime.now(timezone.utc).isoformat() with open(self.opr_log, a, encodingutf-8) as f: f.write(f[{ts}] {message}\n) def _write_evidence(self, stage, action, result_status, duration_ms, error_typeNone): evidence { event_id: fevt-{uuid.uuid4().hex[:8]}, td_id: self.td_id, session_id: self.session_id, timestamp: datetime.now(timezone.utc).isoformat(), stage: stage, action: action, result_status: result_status, duration_ms: duration_ms, retry_count: self.retry_count } if error_type: evidence[error_type] error_type with open(self.evidence_file, a, encodingutf-8) as f: f.write(json.dumps(evidence, ensure_asciiFalse) \n) def transition(self, event): 根据当前状态和事件执行状态转换 transitions self.settings[state_machine][transitions] for t in transitions: if t[from] self.state and t[condition] event: old_state self.state self.state t[to] self._log_event(fstate: {old_state} - {self.state} (event: {event})) self._log_opr(ftransition {old_state}-{self.state} event{event}) self._persist_state() return True self._log_opr(fno transition from {self.state} on event {event}) return False def run_stage(self, stage, action_fn): 执行一个阶段记录证据处理异常 start time.time() try: result action_fn() duration int((time.time() - start) * 1000) self._write_evidence(stage, action_fn.__name__, success, duration) self._log_event(f{stage} success: {action_fn.__name__}) return result except Exception as e: duration int((time.time() - start) * 1000) error_type type(e).__name__ self._write_evidence(stage, action_fn.__name__, failed, duration, error_type) self._log_event(f{stage} failed: {error_type} - {str(e)}) self._log_opr(f{stage} exception: {error_type} {str(e)}) raise def resume(self): 从 state.json 恢复实例 if not self.state_file.exists(): self._log_event(no state file, cannot resume) return False with open(self.state_file, r, encodingutf-8) as f: state_data json.load(f) self.state state_data[current_state] self.session_id state_data[session_id] self.retry_count state_data.get(retry_count, 0) self.action_index state_data.get(action_index, 0) self._log_event(fresumed from state{self.state} session{self.session_id}) return True if __name__ __main__: td TDInstance() print(fTD instance created: {td.td_id}) print(fInitial state: {td.state}) # 模拟 TARGET - OBSERVE td.transition(target_defined) print(fAfter target_defined: {td.state}) # 模拟 OBSERVE - ESTIMATE td.transition(facts_collected) print(fAfter facts_collected: {td.state}) # 模拟 ESTIMATE - DECIDE td.transition(feasible) print(fAfter feasible: {td.state}) # 模拟 DECIDE - ACT td.transition(plan_ready) print(fAfter plan_ready: {td.state}) # 模拟 ACT - CHECK td.transition(action_executed) print(fAfter action_executed: {td.state}) # 模拟 CHECK - DONE td.transition(target_check_passed) print(fAfter target_check_passed: {td.state}) print(f\nState file: {td.state_file}) print(fEvidence file: {td.evidence_file})运行这段代码你会看到状态从TARGET一路转换到DONE每一步都写入了state.json、event.log、opr.log和evidence.jsonl。目录结构如下threads/ └── td-a1b2c3d4/ ├── state.json ├── event.log ├── opr.log ├── trace/ └── artifacts/ └── evidence.jsonlstate.json的内容类似{ td_id: td-a1b2c3d4, session_id: sess-e5f6g7h8, parent_id: null, user_thread: null, current_state: DONE, retry_count: 0, action_index: 0, updated_at: 2025-01-15T10:30:0000:00 }evidence.jsonl每行是一条证据记录{event_id: evt-12345678, td_id: td-a1b2c3d4, session_id: sess-e5f6g7h8, timestamp: 2025-01-15T10:30:0000:00, stage: TARGET, action: define_target, result_status: success, duration_ms: 120, retry_count: 0}这个最小实现的关键设计是每次状态转换都调用_persist_state()写盘每次阶段执行都调用_write_evidence()追加证据。这样即使进程崩溃你也能从state.json读到最后一个稳定状态从evidence.jsonl读到完整的执行历史。如果你用的是 Claude Code可以把这段代码和配置文件放在工作区然后在对话里让 Agent 执行python td_runner.py观察状态转换和文件落盘。如果你需要更完整的接入示例可以参考 https://taotoken.net/doc 里的文档里面有不同语言的调用示例。4. 验证请求与异常恢复实测配置和代码都有了现在需要验证两件事第一模型调用是否正常第二异常恢复是否真的能续跑。先验证模型调用用 curl 发一个最简单的请求确认 TaoToken 的接入地址和 Key 是通的。curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer sk-your-key-here \ -d { model: gpt-4o-mini, messages: [ {role: user, content: 回复 OK 两个字母即可} ], max_tokens: 10 }如果返回类似下面的结构说明接入正常{ id: chatcmpl-xxx, object: chat.completion, choices: [ { index: 0, message: { role: assistant, content: OK }, finish_reason: stop } ], usage: { prompt_tokens: 12, completion_tokens: 2, total_tokens: 14 } }注意choices数组里的message.content就是模型返回的内容。如果你看到choices是空数组或者报错先检查 Key 和 Base URL 是否正确。现在验证异常恢复。我设计一个场景TD 实例执行到ACT阶段时模拟一个工具调用失败然后从state.json恢复继续执行。修改td_runner.py加入一个会失败的 actiondef failing_action(): raise TimeoutError(docker command timed out after 30s) # 在 main 里模拟异常 td TDInstance() td.transition(target_defined) td.transition(facts_collected) td.transition(feasible) td.transition(plan_ready) print(fBefore ACT: {td.state}) try: td.run_stage(ACT, failing_action) except TimeoutError: print(ACT failed, checking exception handling...) # 根据 settings.json 的 exception_handlingact_fail 允许重试 td.retry_count 1 td._persist_state() print(fRetry count: {td.retry_count}, state: {td.state})运行后evidence.jsonl会多一条result_status: failed的记录error_type是TimeoutError。state.json里的retry_count变成 1。现在模拟进程重启从state.json恢复# 模拟进程重启后恢复 td2 TDInstance(td_idtd.td_id) td2.resume() print(fResumed state: {td2.state}, retry_count: {td2.retry_count})resume()方法从state.json读回状态你会看到state仍然是ACTretry_count是 1。这意味着你可以从断点继续执行而不是从头开始。如果重试次数超过max_action_retries配置里是 5状态机会转换到FAILED或PAUSED等待人类介入。再验证一个更复杂的场景CHECK阶段发现 target_check 失败需要重新规划。修改状态转换# 模拟 CHECK 失败后重新规划 td.transition(action_executed) # ACT - CHECK print(fAfter action_executed: {td.state}) # target_check 失败回到 DECIDE 重新规划 td.transition(target_check_failed_and_replan) print(fAfter target_check_failed_and_replan: {td.state})你会看到状态从CHECK回到DECIDE这意味着 Agent 不是简单重试同一个动作而是重新生成行动图。这个区分很重要action_check 失败是「动作没做对」target_check 失败是「动作做对了但没推动目标」后者需要重新规划而不是重试。实测下来这套机制在长周期任务里的价值很明显。我跑了一个模拟的部署流程故意在第三步注入超时然后从state.json恢复Agent 从ACT阶段继续没有重复前两步的观察和评估。证据链里完整记录了失败原因、重试次数和恢复时间点事后复盘一目了然。如果你需要验证模型对话的更多参数可以访问 https://taotoken.net/chat 直接测试不同模型的返回。如果你打算长期跑编码类 Agent 任务可以考虑 Coding Plan它在频繁调用场景下更划算具体可以看 https://taotoken.net/coding-plan。5. 常见报错排查401、local proxy failed、reading choices这一节对照真实报错给出排查路径。这些错误我在搭建过程中都遇到过按顺序排查基本能定位。401 Unauthorized。这是最常见的错误返回体通常是{error: {message: Invalid API key, type: invalid_request_error}}。原因有三个Key 写错了、Key 过期了、请求头格式不对。先检查config.toml里的api_key是否以sk-开头然后确认请求头是Authorization: Bearer sk-xxx注意Bearer和 Key 之间有一个空格。如果 Key 是从 https://taotoken.net/api-keys 复制的确认没有多余的空格或换行。还有一个容易忽略的点有些客户端会自动加Bearer前缀如果你在配置里已经写了Bearer就会变成Bearer Bearer sk-xxx同样报 401。local proxy failed。这个报错通常出现在你本地配置了代理但代理不可用或配置错误时。错误信息类似Error: local proxy failed: connection refused或proxy error: cannot connect to proxy。排查步骤先检查环境变量HTTP_PROXY和HTTPS_PROXY是否设置了无效的代理地址如果有临时取消设置再试。然后检查客户端的代理配置比如 Claude Code 的settings.json里是否有proxy字段Cline 的 MCP 配置里是否有代理相关参数。如果你用的是 Codex检查auth.json里是否混入了代理配置。这个报错的本质是请求没有直达目标地址而是被转发到了一个不可用的中间层。确保你的请求直接发往https://taotoken.net/api中间不要经过任何本地代理。reading choices 报错。这个报错通常长这样TypeError: Cannot read properties of undefined (reading choices)或KeyError: choices。原因是代码期望返回体里有choices字段但实际返回的不是标准结构。可能的情况返回的是错误信息而不是正常响应比如{error: {...}}这时候没有choices或者返回的是流式响应的分片分片里没有完整的choices或者模型名称写错了服务端返回了错误。排查方法先把原始返回体打印出来看看到底返回了什么。如果是错误信息按错误信息排查如果是流式分片检查你的解析逻辑是否正确处理了data:前缀和[DONE]标记。还有一个常见原因是model字段写了一个不存在的模型名服务端返回 404 或 400但你的代码直接去读choices就报错了。OAuth 相关报错。如果你用的是 Claude Code 或类似的工具可能会遇到OAuth token expired或invalid_grant。这类报错和 API Key 无关是客户端的登录态问题。排查步骤先确认你用的是 API Key 模式而不是 OAuth 模式在 Claude Code 里可以通过claude config查看当前认证方式。如果确实需要 OAuth重新登录即可。但如果你只是想用 API Key 接入确保配置里没有残留的 OAuth 配置。对于 Codex检查auth.json里是否有oauth字段如果有删掉或替换成 API Key 配置。CC Switch / Cline MCP / Codex auth.json 三件套。如果你用这些工具接入必须写全三件套Base URL、Key、Model ID。以 Cline 的 MCP 配置为例{ mcpServers: { taotoken: { command: npx, args: [-y, taotoken/mcp-server], env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-your-key-here, TAOTOKEN_MODEL: gpt-4o } } } }Base URL 是https://taotoken.net/apiKey 是你的 API KeyModel ID 是你要用的模型名称。三个缺一不可少一个就会报错。Codex 的auth.json类似{ base_url: https://taotoken.net/api, api_key: sk-your-key-here, model: gpt-4o }CC Switch 的配置在它的设置界面里同样需要填这三项。如果你遇到model not found或invalid model先检查 Model ID 是否拼写正确不同接入点支持的模型名称可能略有差异以文档为准。状态文件损坏。这个报错不是网络层面的而是state.json解析失败比如json.decoder.JSONDecodeError。原因通常是写入过程中进程被强制杀死导致文件写了一半。排查方法先备份损坏的state.json然后检查文件内容是否完整。如果只是最后一行不完整可以手动补上}。更稳妥的做法是在_persist_state()里先写临时文件再原子重命名def _persist_state(self): state_data {...} tmp_file self.state_file.with_suffix(.tmp) with open(tmp_file, w, encodingutf-8) as f: json.dump(state_data, f, ensure_asciiFalse, indent2) tmp_file.replace(self.state_file)这样即使写入过程中崩溃原state.json也不会被破坏。6. 从最小骨架到可续跑的执行系统走到这里你已经有了一个能跑通状态转换、能落盘证据、能从断点恢复的最小骨架。但骨架和真正可用的执行系统之间还有一段距离这段距离需要用真实任务来填。第一个要补的是父子 TD 的持久化。上面的实现只支持单层 TD但实际任务往往需要分解。父 TD 负责规划和验收子 TD 负责具体执行。你需要在state.json里加parent_id字段子 TD 完成后通知父 TD父 TD 根据子 TD 的证据链决定是否继续。这个通知机制可以简单到一个文件标记比如子 TD 在artifacts/下写一个done.flag父 TD 轮询这个文件。第二个要补的是异常经验的结构化累积。每次异常处理都写入evidence.jsonl但证据是流水账你需要一个索引来快速检索。可以在artifacts/下加一个exception_index.json按error_type和stage分组记录每次异常的处理方式和结果。这样下次遇到同类异常时Agent 可以先查索引看历史上是怎么处理的而不是从零开始试错。第三个要补的是预算约束的强制执行。上面的配置里定义了max_tokens_per_td和max_duration_seconds但代码里还没有真正检查。你需要在每次模型调用后累加 token 消耗在每次状态转换前检查是否超时。超预算时根据on_budget_exceeded执行对应动作escalate表示写入PAUSED状态并等待人类确认。第四个要补的是证据链的复核接口。evidence.jsonl是给机器读的但人类需要更友好的视图。可以写一个简单的脚本把evidence.jsonl渲染成时间线按stage分组标出失败的节点和重试次数。这样你在复盘时不用逐行读 JSON。如果你打算把这套骨架用在长期运行的编码 Agent 上建议把模型调用统一走 TaoToken 的接入点这样计费和限流都在一个地方管理。对于需要频繁调用、长时间运行的场景Coding Plan 比按量计费更可控具体可以看 https://taotoken.net/coding-plan。如果你只是想先验证模型对话的效果可以直接在 https://taotoken.net/chat 里测试不同模型对同一任务的返回差异。最后给一个实用技巧在config.toml里把checkpoint_interval设小一点比如 2 或 3这样恢复时丢失的进度更少。代价是磁盘写入更频繁但在长周期任务里这个代价是值得的。另外event.log和opr.log要分开前者给人类看用自然语言描述后者给 Agent 看用结构化字段。混在一起会让两边都难读。这套骨架的核心价值不是代码本身而是它强制你把「状态」和「证据」显式化。一旦状态落盘、证据留痕Agent 的执行就从黑盒变成了白盒中断可恢复、失败可追溯、结果可复核。剩下的就是选一个真实任务跑通端到端然后根据实际报错迭代。