agno Team 检查点与崩溃恢复实战checkpoint 策略、原地 /continue 恢复与 HTTP 检查点接口【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno在 agno 中多成员协作的Team与单个Agent共享同一套检查点checkpoint心智模型——同样的动词、同样的参数、同样的 COMPLETED 自动 fork 语义。当您把Team(checkpointtool-batch)交给生产环境时Team 会在每次团队级工具批次一次委派给成员的动作本身就构成一个工具批次之后持久化团队运行状态从而获得运行中途的持久性和崩溃恢复能力。本文基于 cookbook/03_teams/23_checkpointing/README.md 及仓库中的三个完整示例展开讲解 Team 检查点策略的取舍、成员与 Team 检查点的边界划分并逐步演示模拟崩溃 → 从数据库中的RUNNING检查点原地续跑、工具异常与模型调用失败的两种持久化路径、通过 HTTP 检查点接口回溯时间线并驱动 /continue三条实战链路帮助您在多 Agent 服务中构建可靠的崩溃恢复能力。一、核心模型Team 与 Agent 的对齐语义1.1 团队级工具批次即检查点粒度与 Agent 面的检查点实现保持直接对齐parityTeam 检查点遵循同一套规则默认只把运行状态写入终态COMPLETED、ERROR等当显式开启Team(checkpointtool-batch)后Team 在每次团队级工具批次之后写一次检查点且该次写入的状态为RUNNING由于委派给成员在 Team 视角中本身就是一次工具调用因此成员执行产生的消息会以 tool-role 消息的形式进入 Team 会话记录委派过程天然构成被持久化的工具批次。Team.checkpoint字段在类型上被限定为三种取值之一声明于 libs/agno/agno/team/team.py 与 libs/agno/agno/team/_init.pycheckpoint: Optional[Literal[runs, tool-batch, tools]] None策略语义对应 README 中的策略清单取值含义适用场景runs默认仅在终态写入与 Agent 默认行为一致普通对话、无需中途恢复的场景tool-batch每个团队级工具批次后写入一次需要崩溃恢复、运行中途持久化的生产场景tools为 3.0 预留当前使用会抛出NotImplementedError从源码可以确认策略校验发生在 Team 初始化阶段agno/team/_init.py中的set_checkpoint()会把None解析为runs把tools直接抛错并提示checkpointtoolsis reserved for the 3.0 runs-table split and not available yet同时对其它非法取值抛出ValueError(Invalid checkpoint level ...)。单元测试 libs/agno/tests/unit/team/test_team_checkpointing.py 覆盖了以上全部分支默认解析为runs、显式tool-batch被保留、tools抛NotImplementedError、非法值抛ValueError。1.2 成员不在 Team 检查点范围内README 明确指出一个容易踩坑的边界团队成员是 Team 检查点的范围之外对象。从 Team 的角度看成员只是一个被委派的工具它的输出进入 Team 会话时成为一条 tool-role 消息因此fork/regenerate/time-travel等续写能力都作用在Team 自身状态上不会触及成员自身的运行状态。注意这并不等于成员运行的会话不受影响——Team 也会把成员运行member runs持久化到同一会话中只是 Team 的续写/回溯/分支视图始终以团队会话为主干。从 libs/agno/tests/unit/team/test_team_checkpointing.py 的说明看fork 时会为成员运行深层复制出新的run_id并重新挂载到新 Team 下这也印证了成员状态与 Team 检查点分层管理的设计。1.3 三类 /continue 能力的文件夹分布/continue相关的三类重来能力与检查点紧密相关它们分别位于同级目录cookbook/03_teams/24_regenerate/ —— 重做上一条回复regeneratecookbook/03_teams/25_time_travel/ —— 回退到历史检查点continue_from、forkcookbook/03_teams/26_fork_session/ —— 复制整个会话。阅读完本文后建议结合这三个目录把检查点写入 → 崩溃恢复 → 回溯改写串成一条完整链路理解。二、示例总览与运行方式23_checkpointing目录下三个示例分别演示三种故障与恢复场景完整清单见 cookbook/03_teams/23_checkpointing/README.md示例文件演示内容01_crash_recovery.py中断一个正在飞行中的团队运行以模拟崩溃数据库中保留最近一次RUNNING检查点/continue原地恢复02_tool_error_persistence.py工具异常被捕获并记录模型调用失败则逃逸出循环但飞行中的会话被冲刷到ERROR行/continue可重试03_checkpoint_endpoints.py两个 GET 端点/checkpoints时间线与/checkpoints/{message_index}快照并把返回的索引回灌给/continueREADME 中给出的运行命令假设已按 cookbook 规范准备好 demo 虚拟环境.venvs/demo/bin/python cookbook/03_teams/23_checkpointing/01_crash_recovery.py .venvs/demo/bin/python cookbook/03_teams/23_checkpointing/02_tool_error_persistence.py .venvs/demo/bin/python cookbook/03_teams/23_checkpointing/03_checkpoint_endpoints.py三个示例均使用OpenAIResponses模型因此运行时需要OPENAI_API_KEY。根据 cookbook/03_teams/23_checkpointing/TEST_LOG.md它们已通过语法/编译校验实际运行依赖真实模型调用其中示例 01 与 02 的团队侧 flush 与持久化逻辑分别由libs/agno/tests/unit/team/test_team_checkpointing.py中的TestTeamCheckpointConfig、TestTeamTruncate与TestTeamFlushHelper等测试类做单元级覆盖。三、示例一tool-batch检查点 子进程 SIGKILL 模拟崩溃恢复3.1 设计动机为什么用子进程 SIGKILLcookbook/03_teams/23_checkpointing/01_crash_recovery.py 的注释解释了为什么不用asyncio.Task.cancel()来模拟崩溃取消cancel会被框架优雅处理——运行会被标记为CANCELLED并重新持久化而已取消的运行刻意设计为不可续跑。真实的崩溃不会执行任何清理逻辑因此最后一次RUNNING检查点才是唯一幸存者对子进程执行SIGKILL恰好能复现这种硬崩溃例如 OOM-kill。3.2 示例流程拆解示例的完整流程分五步启动 worker 子进程main()用subprocess.Popen([sys.executable, __file__, --worker], ...)拉起子进程子进程内build_team()构造 Team 并执行team.arun(input..., session_idteam-crash-demo-session)。关键配置是父进程与子进程共享同一个DB_FILE指向的 SQLite 数据库team Team( nameresearch-team, modelOpenAIResponses(idgpt-5.4), members[researcher], dbSqliteDb(session_tableteam_checkpoint_demo, db_fileDB_FILE), checkpointtool-batch, # 团队级工具批次后落一次检查点 instructionsDelegate the research to the researcher, then summarize., )成员 Agent 挂载了两个各耗时约 1 秒的模拟工具slow_search与slow_fetch_detail为父进程赢得拦截窗口。轮询数据库父进程以 0.5 秒间隔最多轮询约 80 次直到发现run.status RunStatus.running且run.tools非空——即第一个带工具批次的RUNNING检查点落库。SIGKILL 子进程worker.kill(); worker.wait()制造一次无清理的真实崩溃。若在 worker 自行结束前未抓到检查点脚本会提示model occasionally answers without enough tool batches建议重跑。检查数据库重新读取 session打印崩溃运行的run_id、status、len(tools)工具批次数、len(messages)消息数与last_checkpoint_at_message_idx。此时状态仍是RUNNING——团队运行循环从未到达终态清理。原地续跑调用team.acontinue_run(run_idcrashed_run.run_id, session_idSESSION_ID)恢复。注意恢复后run_id与崩溃前相同——这就是 README 强调的 in-place /continue。3.3 关键语义RUNNING 与 ERROR 对 /continue 等价脚本在恢复前特意打印了这样一句Status is RUNNING — the team loop never reached terminal cleanup. For /continue, RUNNING and ERROR are equivalent: both resume.也就是说只要运行状态不是COMPLETED/continue就会在原 run_id 上原地续跑同一 run_id 上继续追加消息而非另起炉灶。这是团队崩溃恢复能无缝接续的根本前提。3.4 源码佐证tool-batch 检查点到底做了什么checkpointtool-batch的落地逻辑集中在 libs/agno/agno/team/_run.pycheckpoint_team_run()及异步版acheckpoint_team_run()首先判空team.checkpoint ! tool-batch时直接返回——即未开启时为零成本路径写入前会把run_response.status置为RunStatus.running记录run_response.last_checkpoint_at_message_index len(messages)并通过_mark_team_checkpoint_message()在最后一条消息上打上checkpoint_status与checkpoint_created_at标记供前端时间线渲染随后调用_persist_team_run_in_session()libs/agno/agno/team/_run.py走与终态写入同一条路径减去 timer 停止、approval 更新等收尾副作用把消息、工具批次与 media 引用一并落到会话行。回调由build_team_after_tool_results_callback()在每个工具批次结果就绪后触发先把飞行中的model_response状态镜像回run_response再执行一次检查点持久化。该镜像同步有一个团队特有的细节见_sync_team_run_response_with_model_responsechild_run_id委派 → 成员运行 id 的链接是在工具执行期被补丁到已有 tools 条目上的同步时会按tool_call_id搬运避免每次中途检查点把委派链接弄丢。四、示例二工具异常与模型调用失败的两条持久化路径第二个示例 cookbook/03_teams/23_checkpointing/02_tool_error_persistence.py 回答一个更刁钻的问题出错时对话到底能不能活下来它把失败拆成两个看起来像、结局不同的场景4.1 场景 A工具抛出普通 Python 异常可恢复团队成员/团队级工具broken_tool每次都抛ValueError。团队模型循环会在内部捕获它将其转换为一条带tool_call_errorTrue的 tool-role 消息随后照常触发检查点 hook 并继续执行。最终运行正常完成COMPLETED错误以消息形式留在对话里——零数据丢失。示例随后用全新的 Team 读取数据库逐条打印消息并标注[tool_call_error]验证错误消息确实被持久化。4.2 场景 B模型调用本身失败异常逃逸循环场景 B 用故意填入非法 API Keysk-invalid-key-deliberately-broken-to-force-auth-error模拟团队模型调用在任何工具批次发生之前就失败。此时异常逃逸出模型循环每个批次后的检查点 hook 根本没有机会执行如果没有兜底逻辑ERROR行会被以空消息持久化导致引发失败的那段对话永久丢失。仓库在团队侧提供了与 Agent 同名的兜底函数flush_in_flight_messages_on_error_team()libs/agno/agno/team/_run.py。其逻辑如下若飞行中的run_messages为空或run_response.messages已被中途 hook 填充则跳过只填空、不覆盖避免破坏已捕获的中间状态否则把run_messages.messages中add_to_agent_memory为真的消息复制进run_response.messages在终态ERROR写入前完成冲刷flush。该 helper 在 libs/agno/agno/team/_run.py 中被_run的多个错误处理路径调用例如模型循环异常分支中以locals().get(run_messages)方式传入并被libs/agno/tests/unit/team/test_team_checkpointing.py的TestTeamFlushHelper覆盖。源码注释同时诚实标注了一个已知缺口KNOWN GAP / tombstone部分分离的后台包装器detached background wrappers作用域内拿不到run_messages它们的 flush 调用实际是 no-op 并被删除包装器级错误会不带飞行对话地持久化这是后续跟进事项。示例脚本则把空消息与消息已保留两种情况都打印出来让读者可以对照验证 helper 是否生效。4.3 场景 C对ERROR运行执行 /continueERROR不等于COMPLETED因此acontinue_run(run_idfailed_run_id, session_idsess-B)会在同一 run_id 上原地续跑。由于场景 B 中消息已保留模型拥有完整上下文可用于重试。示例最后打印 session 中的运行列表确认 run_id 未变。需要留意的是场景 B 临时改写了OPENAI_API_KEY并在finally块中恢复原值或移除变量cookbook/03_teams/23_checkpointing/02_tool_error_persistence.py这是示例自洽性设计读者不必模仿此做法。五、示例三HTTP 检查点接口时间线 快照 回灌 /continue第三个示例 cookbook/03_teams/23_checkpointing/03_checkpoint_endpoints.py 演示了两个与 Agent 变体对应的团队 GET 端点它们由已持久化的团队运行推导出检查点——并没有一张单独的 checkpoint 表端点返回内容GET /teams/{team_id}/runs/{run_id}/checkpoints?session_id...从团队运行的转录中推导出的消息边界时间线GET /teams/{team_id}/runs/{run_id}/checkpoints/{message_index}?session_id...在指定边界处截断的快照用该边界上的message_index作为续跑时的continue_fromK5.1 服务端实现位置两个端点在 libs/agno/agno/os/routers/teams/router.py 中注册list_team_run_checkpointsoperation_idlist_team_run_checkpoints解析 Team 后读取run_output调用list_run_checkpoints(run_output)返回{run_id, session_id, checkpoints}其 API 描述明确写道检查点由消息级标记与转录终端推导而来不依赖独立表get_team_run_checkpoint_snapshotoperation_idget_team_run_checkpoint_snapshot解析消息索引后调用build_run_checkpoint_snapshot(run_output, message_index)若索引非法则以ValueError→ HTTP 400 返回对RemoteTeam会返回 400 提示不支持远程团队。list_run_checkpoints与build_run_checkpoint_snapshot的实现位于 libs/agno/agno/os/checkpoints.py与 Agent 侧共用同一套推导逻辑。5.2 自包含演示in-process AgentOS示例用fastapi.testclient.TestClient在进程内启动一个AgentOS描述为 team-checkpoint-endpoints demo挂载pop-team因此不需要单独服务器、不绑定端口即可驱动 HTTP 层agent_os AgentOS(descriptionteam-checkpoint-endpoints demo, teams[team]) app agent_os.get_app() client TestClient(app)示例中的 Team 委派一个pop-agent成员回答人口问题并开启checkpointtool-batch然后依次执行四步驱动一次运行team.run(Compare the populations of Paris, Tokyo, and Lagos ...)打印run_id与消息数。拉取时间线GET /teams/{team.id}/runs/{run.run_id}/checkpoints以 JSON 打印checkpoints列表。取内部边界快照从时间线中挑出is_latest为假非终态的首个边界取其message_index请求快照端点响应中的checkpoint携带元数据snapshot.messages为截断后的消息、snapshot.tools只保留被引用到的工具批次。回灌 /continue把上一步的message_index作为continue_from提交到POST /teams/{team.id}/runs/{run.run_id}/continue并附上新的输入如Actually, just Paris.。若执行的是 fork 语义响应体中的forked_from_run_id与forked_from_message_index会指明新 run 从哪个边界分支出来。若时间线中没有内部边界Team 单轮就结束示例会打印(No interior checkpoints found ...)友好降级。六、团队检查点工程的落地清单综合 README 与三份示例在生产中启用 Team 崩溃恢复时建议核对以下要点显式选择策略需要运行中途可恢复就设checkpointtool-batch默认runs只在终态落盘硬崩溃会丢全部中间进度tools在 3.0 前不可用。共享同一数据库崩溃恢复要求续跑方与崩溃方读写同一会话与运行如示例一中父子进程共用同一SqliteDb与DB_FILE才能通过get_session(session_id..., session_typeteam)找到崩溃遗留的运行。理解状态语义RUNNING/ERROR均可被/continue原地续跑同一run_idCOMPLETED则会触发自动 fork 语义被优雅CANCEL的运行刻意不可续跑。记忆委派边界成员运行不在 Team 检查点范围内fork / time-travel 作用于团队自身状态但 Team 会话仍会持久化成员运行行供排查与审计。兜底 flush 是否生效模型调用在首个工具批次前失败时检查ERROR行是否携带飞行消息——它由flush_in_flight_messages_on_error_team保证同时留意源码中标注的 detached-wrapper 已知缺口。HTTP 层可直接编排AgentOS 的/teams/{team_id}/runs/{run_id}/checkpoints系列端点把时间线 → 快照 → continue_from 续跑编排成前端可用的流程团队侧实现与 Agent 侧完全对称可复用同一套前端逻辑。结合 cookbook/03_teams/24_regenerate/、cookbook/03_teams/25_time_travel/ 与 cookbook/03_teams/26_fork_session/您可以进一步把 checkpoint 从崩溃兜底升级为完整的运行可回溯、可改写、可分支的多 Agent 运维体系。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 agno Team 检查点与崩溃恢复实战：checkpoint 策略、原地 /continue 恢复与 HTTP 检查点接口
agno Team 检查点与崩溃恢复实战checkpoint 策略、原地 /continue 恢复与 HTTP 检查点接口【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno在 agno 中多成员协作的Team与单个Agent共享同一套检查点checkpoint心智模型——同样的动词、同样的参数、同样的 COMPLETED 自动 fork 语义。当您把Team(checkpointtool-batch)交给生产环境时Team 会在每次团队级工具批次一次委派给成员的动作本身就构成一个工具批次之后持久化团队运行状态从而获得运行中途的持久性和崩溃恢复能力。本文基于 cookbook/03_teams/23_checkpointing/README.md 及仓库中的三个完整示例展开讲解 Team 检查点策略的取舍、成员与 Team 检查点的边界划分并逐步演示模拟崩溃 → 从数据库中的RUNNING检查点原地续跑、工具异常与模型调用失败的两种持久化路径、通过 HTTP 检查点接口回溯时间线并驱动 /continue三条实战链路帮助您在多 Agent 服务中构建可靠的崩溃恢复能力。一、核心模型Team 与 Agent 的对齐语义1.1 团队级工具批次即检查点粒度与 Agent 面的检查点实现保持直接对齐parityTeam 检查点遵循同一套规则默认只把运行状态写入终态COMPLETED、ERROR等当显式开启Team(checkpointtool-batch)后Team 在每次团队级工具批次之后写一次检查点且该次写入的状态为RUNNING由于委派给成员在 Team 视角中本身就是一次工具调用因此成员执行产生的消息会以 tool-role 消息的形式进入 Team 会话记录委派过程天然构成被持久化的工具批次。Team.checkpoint字段在类型上被限定为三种取值之一声明于 libs/agno/agno/team/team.py 与 libs/agno/agno/team/_init.pycheckpoint: Optional[Literal[runs, tool-batch, tools]] None策略语义对应 README 中的策略清单取值含义适用场景runs默认仅在终态写入与 Agent 默认行为一致普通对话、无需中途恢复的场景tool-batch每个团队级工具批次后写入一次需要崩溃恢复、运行中途持久化的生产场景tools为 3.0 预留当前使用会抛出NotImplementedError从源码可以确认策略校验发生在 Team 初始化阶段agno/team/_init.py中的set_checkpoint()会把None解析为runs把tools直接抛错并提示checkpointtoolsis reserved for the 3.0 runs-table split and not available yet同时对其它非法取值抛出ValueError(Invalid checkpoint level ...)。单元测试 libs/agno/tests/unit/team/test_team_checkpointing.py 覆盖了以上全部分支默认解析为runs、显式tool-batch被保留、tools抛NotImplementedError、非法值抛ValueError。1.2 成员不在 Team 检查点范围内README 明确指出一个容易踩坑的边界团队成员是 Team 检查点的范围之外对象。从 Team 的角度看成员只是一个被委派的工具它的输出进入 Team 会话时成为一条 tool-role 消息因此fork/regenerate/time-travel等续写能力都作用在Team 自身状态上不会触及成员自身的运行状态。注意这并不等于成员运行的会话不受影响——Team 也会把成员运行member runs持久化到同一会话中只是 Team 的续写/回溯/分支视图始终以团队会话为主干。从 libs/agno/tests/unit/team/test_team_checkpointing.py 的说明看fork 时会为成员运行深层复制出新的run_id并重新挂载到新 Team 下这也印证了成员状态与 Team 检查点分层管理的设计。1.3 三类 /continue 能力的文件夹分布/continue相关的三类重来能力与检查点紧密相关它们分别位于同级目录cookbook/03_teams/24_regenerate/ —— 重做上一条回复regeneratecookbook/03_teams/25_time_travel/ —— 回退到历史检查点continue_from、forkcookbook/03_teams/26_fork_session/ —— 复制整个会话。阅读完本文后建议结合这三个目录把检查点写入 → 崩溃恢复 → 回溯改写串成一条完整链路理解。二、示例总览与运行方式23_checkpointing目录下三个示例分别演示三种故障与恢复场景完整清单见 cookbook/03_teams/23_checkpointing/README.md示例文件演示内容01_crash_recovery.py中断一个正在飞行中的团队运行以模拟崩溃数据库中保留最近一次RUNNING检查点/continue原地恢复02_tool_error_persistence.py工具异常被捕获并记录模型调用失败则逃逸出循环但飞行中的会话被冲刷到ERROR行/continue可重试03_checkpoint_endpoints.py两个 GET 端点/checkpoints时间线与/checkpoints/{message_index}快照并把返回的索引回灌给/continueREADME 中给出的运行命令假设已按 cookbook 规范准备好 demo 虚拟环境.venvs/demo/bin/python cookbook/03_teams/23_checkpointing/01_crash_recovery.py .venvs/demo/bin/python cookbook/03_teams/23_checkpointing/02_tool_error_persistence.py .venvs/demo/bin/python cookbook/03_teams/23_checkpointing/03_checkpoint_endpoints.py三个示例均使用OpenAIResponses模型因此运行时需要OPENAI_API_KEY。根据 cookbook/03_teams/23_checkpointing/TEST_LOG.md它们已通过语法/编译校验实际运行依赖真实模型调用其中示例 01 与 02 的团队侧 flush 与持久化逻辑分别由libs/agno/tests/unit/team/test_team_checkpointing.py中的TestTeamCheckpointConfig、TestTeamTruncate与TestTeamFlushHelper等测试类做单元级覆盖。三、示例一tool-batch检查点 子进程 SIGKILL 模拟崩溃恢复3.1 设计动机为什么用子进程 SIGKILLcookbook/03_teams/23_checkpointing/01_crash_recovery.py 的注释解释了为什么不用asyncio.Task.cancel()来模拟崩溃取消cancel会被框架优雅处理——运行会被标记为CANCELLED并重新持久化而已取消的运行刻意设计为不可续跑。真实的崩溃不会执行任何清理逻辑因此最后一次RUNNING检查点才是唯一幸存者对子进程执行SIGKILL恰好能复现这种硬崩溃例如 OOM-kill。3.2 示例流程拆解示例的完整流程分五步启动 worker 子进程main()用subprocess.Popen([sys.executable, __file__, --worker], ...)拉起子进程子进程内build_team()构造 Team 并执行team.arun(input..., session_idteam-crash-demo-session)。关键配置是父进程与子进程共享同一个DB_FILE指向的 SQLite 数据库team Team( nameresearch-team, modelOpenAIResponses(idgpt-5.4), members[researcher], dbSqliteDb(session_tableteam_checkpoint_demo, db_fileDB_FILE), checkpointtool-batch, # 团队级工具批次后落一次检查点 instructionsDelegate the research to the researcher, then summarize., )成员 Agent 挂载了两个各耗时约 1 秒的模拟工具slow_search与slow_fetch_detail为父进程赢得拦截窗口。轮询数据库父进程以 0.5 秒间隔最多轮询约 80 次直到发现run.status RunStatus.running且run.tools非空——即第一个带工具批次的RUNNING检查点落库。SIGKILL 子进程worker.kill(); worker.wait()制造一次无清理的真实崩溃。若在 worker 自行结束前未抓到检查点脚本会提示model occasionally answers without enough tool batches建议重跑。检查数据库重新读取 session打印崩溃运行的run_id、status、len(tools)工具批次数、len(messages)消息数与last_checkpoint_at_message_idx。此时状态仍是RUNNING——团队运行循环从未到达终态清理。原地续跑调用team.acontinue_run(run_idcrashed_run.run_id, session_idSESSION_ID)恢复。注意恢复后run_id与崩溃前相同——这就是 README 强调的 in-place /continue。3.3 关键语义RUNNING 与 ERROR 对 /continue 等价脚本在恢复前特意打印了这样一句Status is RUNNING — the team loop never reached terminal cleanup. For /continue, RUNNING and ERROR are equivalent: both resume.也就是说只要运行状态不是COMPLETED/continue就会在原 run_id 上原地续跑同一 run_id 上继续追加消息而非另起炉灶。这是团队崩溃恢复能无缝接续的根本前提。3.4 源码佐证tool-batch 检查点到底做了什么checkpointtool-batch的落地逻辑集中在 libs/agno/agno/team/_run.pycheckpoint_team_run()及异步版acheckpoint_team_run()首先判空team.checkpoint ! tool-batch时直接返回——即未开启时为零成本路径写入前会把run_response.status置为RunStatus.running记录run_response.last_checkpoint_at_message_index len(messages)并通过_mark_team_checkpoint_message()在最后一条消息上打上checkpoint_status与checkpoint_created_at标记供前端时间线渲染随后调用_persist_team_run_in_session()libs/agno/agno/team/_run.py走与终态写入同一条路径减去 timer 停止、approval 更新等收尾副作用把消息、工具批次与 media 引用一并落到会话行。回调由build_team_after_tool_results_callback()在每个工具批次结果就绪后触发先把飞行中的model_response状态镜像回run_response再执行一次检查点持久化。该镜像同步有一个团队特有的细节见_sync_team_run_response_with_model_responsechild_run_id委派 → 成员运行 id 的链接是在工具执行期被补丁到已有 tools 条目上的同步时会按tool_call_id搬运避免每次中途检查点把委派链接弄丢。四、示例二工具异常与模型调用失败的两条持久化路径第二个示例 cookbook/03_teams/23_checkpointing/02_tool_error_persistence.py 回答一个更刁钻的问题出错时对话到底能不能活下来它把失败拆成两个看起来像、结局不同的场景4.1 场景 A工具抛出普通 Python 异常可恢复团队成员/团队级工具broken_tool每次都抛ValueError。团队模型循环会在内部捕获它将其转换为一条带tool_call_errorTrue的 tool-role 消息随后照常触发检查点 hook 并继续执行。最终运行正常完成COMPLETED错误以消息形式留在对话里——零数据丢失。示例随后用全新的 Team 读取数据库逐条打印消息并标注[tool_call_error]验证错误消息确实被持久化。4.2 场景 B模型调用本身失败异常逃逸循环场景 B 用故意填入非法 API Keysk-invalid-key-deliberately-broken-to-force-auth-error模拟团队模型调用在任何工具批次发生之前就失败。此时异常逃逸出模型循环每个批次后的检查点 hook 根本没有机会执行如果没有兜底逻辑ERROR行会被以空消息持久化导致引发失败的那段对话永久丢失。仓库在团队侧提供了与 Agent 同名的兜底函数flush_in_flight_messages_on_error_team()libs/agno/agno/team/_run.py。其逻辑如下若飞行中的run_messages为空或run_response.messages已被中途 hook 填充则跳过只填空、不覆盖避免破坏已捕获的中间状态否则把run_messages.messages中add_to_agent_memory为真的消息复制进run_response.messages在终态ERROR写入前完成冲刷flush。该 helper 在 libs/agno/agno/team/_run.py 中被_run的多个错误处理路径调用例如模型循环异常分支中以locals().get(run_messages)方式传入并被libs/agno/tests/unit/team/test_team_checkpointing.py的TestTeamFlushHelper覆盖。源码注释同时诚实标注了一个已知缺口KNOWN GAP / tombstone部分分离的后台包装器detached background wrappers作用域内拿不到run_messages它们的 flush 调用实际是 no-op 并被删除包装器级错误会不带飞行对话地持久化这是后续跟进事项。示例脚本则把空消息与消息已保留两种情况都打印出来让读者可以对照验证 helper 是否生效。4.3 场景 C对ERROR运行执行 /continueERROR不等于COMPLETED因此acontinue_run(run_idfailed_run_id, session_idsess-B)会在同一 run_id 上原地续跑。由于场景 B 中消息已保留模型拥有完整上下文可用于重试。示例最后打印 session 中的运行列表确认 run_id 未变。需要留意的是场景 B 临时改写了OPENAI_API_KEY并在finally块中恢复原值或移除变量cookbook/03_teams/23_checkpointing/02_tool_error_persistence.py这是示例自洽性设计读者不必模仿此做法。五、示例三HTTP 检查点接口时间线 快照 回灌 /continue第三个示例 cookbook/03_teams/23_checkpointing/03_checkpoint_endpoints.py 演示了两个与 Agent 变体对应的团队 GET 端点它们由已持久化的团队运行推导出检查点——并没有一张单独的 checkpoint 表端点返回内容GET /teams/{team_id}/runs/{run_id}/checkpoints?session_id...从团队运行的转录中推导出的消息边界时间线GET /teams/{team_id}/runs/{run_id}/checkpoints/{message_index}?session_id...在指定边界处截断的快照用该边界上的message_index作为续跑时的continue_fromK5.1 服务端实现位置两个端点在 libs/agno/agno/os/routers/teams/router.py 中注册list_team_run_checkpointsoperation_idlist_team_run_checkpoints解析 Team 后读取run_output调用list_run_checkpoints(run_output)返回{run_id, session_id, checkpoints}其 API 描述明确写道检查点由消息级标记与转录终端推导而来不依赖独立表get_team_run_checkpoint_snapshotoperation_idget_team_run_checkpoint_snapshot解析消息索引后调用build_run_checkpoint_snapshot(run_output, message_index)若索引非法则以ValueError→ HTTP 400 返回对RemoteTeam会返回 400 提示不支持远程团队。list_run_checkpoints与build_run_checkpoint_snapshot的实现位于 libs/agno/agno/os/checkpoints.py与 Agent 侧共用同一套推导逻辑。5.2 自包含演示in-process AgentOS示例用fastapi.testclient.TestClient在进程内启动一个AgentOS描述为 team-checkpoint-endpoints demo挂载pop-team因此不需要单独服务器、不绑定端口即可驱动 HTTP 层agent_os AgentOS(descriptionteam-checkpoint-endpoints demo, teams[team]) app agent_os.get_app() client TestClient(app)示例中的 Team 委派一个pop-agent成员回答人口问题并开启checkpointtool-batch然后依次执行四步驱动一次运行team.run(Compare the populations of Paris, Tokyo, and Lagos ...)打印run_id与消息数。拉取时间线GET /teams/{team.id}/runs/{run.run_id}/checkpoints以 JSON 打印checkpoints列表。取内部边界快照从时间线中挑出is_latest为假非终态的首个边界取其message_index请求快照端点响应中的checkpoint携带元数据snapshot.messages为截断后的消息、snapshot.tools只保留被引用到的工具批次。回灌 /continue把上一步的message_index作为continue_from提交到POST /teams/{team.id}/runs/{run.run_id}/continue并附上新的输入如Actually, just Paris.。若执行的是 fork 语义响应体中的forked_from_run_id与forked_from_message_index会指明新 run 从哪个边界分支出来。若时间线中没有内部边界Team 单轮就结束示例会打印(No interior checkpoints found ...)友好降级。六、团队检查点工程的落地清单综合 README 与三份示例在生产中启用 Team 崩溃恢复时建议核对以下要点显式选择策略需要运行中途可恢复就设checkpointtool-batch默认runs只在终态落盘硬崩溃会丢全部中间进度tools在 3.0 前不可用。共享同一数据库崩溃恢复要求续跑方与崩溃方读写同一会话与运行如示例一中父子进程共用同一SqliteDb与DB_FILE才能通过get_session(session_id..., session_typeteam)找到崩溃遗留的运行。理解状态语义RUNNING/ERROR均可被/continue原地续跑同一run_idCOMPLETED则会触发自动 fork 语义被优雅CANCEL的运行刻意不可续跑。记忆委派边界成员运行不在 Team 检查点范围内fork / time-travel 作用于团队自身状态但 Team 会话仍会持久化成员运行行供排查与审计。兜底 flush 是否生效模型调用在首个工具批次前失败时检查ERROR行是否携带飞行消息——它由flush_in_flight_messages_on_error_team保证同时留意源码中标注的 detached-wrapper 已知缺口。HTTP 层可直接编排AgentOS 的/teams/{team_id}/runs/{run_id}/checkpoints系列端点把时间线 → 快照 → continue_from 续跑编排成前端可用的流程团队侧实现与 Agent 侧完全对称可复用同一套前端逻辑。结合 cookbook/03_teams/24_regenerate/、cookbook/03_teams/25_time_travel/ 与 cookbook/03_teams/26_fork_session/您可以进一步把 checkpoint 从崩溃兜底升级为完整的运行可回溯、可改写、可分支的多 Agent 运维体系。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考