这次我们来看一个最近在 Show HN 上出现的方向让普通的 subagent 工作流变得持久化persistent且可追踪trackable。Show HN 是 Hacker News 上作者展示自己开源项目的栏目所以这个标题翻译过来就是“给我那个普通子智能体工作流加上保存和追踪能力”。用大白话说它解决的是 AI Agent 开发里一个特别常见的痛点主 Agent 派了一堆 subagent 干活任务执行过程中上下文、中间结果、调用记录全在内存里进程一崩就全没了出了问题也没法复盘是哪个子任务花了多少 token、哪一步失败的。这个项目方向的三个关键词已经说明了核心卖点ordinary不要求你推倒现有 Agent 框架重来persistent任务状态、上下文、中间产物可以落盘恢复trackable每次 subagent 调用的输入输出、耗时、token 消耗都可以查询和审计。下面我会先从核心能力、适用场景开始然后给出一个可落地的持久化模型设计再带你把任务执行器、恢复机制、追踪查询 API、批量任务队列完整走一遍最后补上常见的坑和工程化建议。1. 核心能力速览在直接看代码之前先把这个项目方向能做什么、不能做什么放在一张表里。因为标题只给出了能力边界没有具体的仓库地址和版本所以表格里的参数我尽量写成通用范围实际以你拿到源码后的文档为准。能力项说明项目类型AI Agent 编排增强层 / subagent 工作流管理工具核心功能把普通 subagent 任务的状态持久化到数据库并提供完整的调用追踪能力持久化范围任务元数据、输入输出、上下文消息、错误信息、执行事件追踪维度调用链trace、状态变更历史、耗时、token 用量、模型名称恢复能力进程重启后自动找回中断任务支持重新执行或标记失败重试存储依赖推荐 SQLite 起步生产环境使用 PostgreSQL / MySQLRedis 用于队列和心跳硬件门槛极低普通开发机即可主要消耗 CPU、内存、磁盘GPU 不是必须的启动方式源码运行提供任务执行器、Worker 进程、API 服务三部分是否支持 API支持通用 REST 接口可用于查询任务状态和追踪链是否支持批量任务支持通过任务队列批量提交 subagent 工作流适合场景本地调试、生产 Agent 系统、自动化流水线、审计合规这里需要强调一点标题里的 persistent 和“persistent homology”没有任何关系后者是拓扑数据分析里的持续同调概念属于数学方向。这个项目里的 persistent 指的是任务状态在进程退出、崩溃、重启之后仍然存在可以继续执行或回溯。2. 适用场景与使用边界这个方案适合谁最典型的是已经在用 LangGraph、AutoGen、CrewAI 这类框架或者自己手写 Agent 循环的团队。你现在的 subagent 流程大概率是这样的主 Agent 拆解任务创建 subagentsubagent 调用 LLM把结果返回主 Agent流程结束。看起来简单但一旦进入生产环境你会发现三件事没法回答任务执行到一半服务重启了这个任务是从头再来还是接着跑某个 subagent 失败是因为 prompt 问题还是模型返回异常这个月所有 subagent 调用合计花了多少 token、多少时间持久化和追踪就是来解决这三个问题的。它特别适合以下场景长时间运行的多步骤任务比如研究报告生成、批量数据处理、多轮代码审查这类任务分钟级起步中断成本高。需要审计和复盘的生产环境比如客服工单处理、内容审核流程、金融文档分析你需要知道每一步是谁、什么模型、什么输入、什么输出。批量并发的 subagent 调度几十个文件、几十个 URL 需要同时处理必须有一个可靠的任务队列和失败重试机制。也存在不适合的情况。如果你的调用链极短比如只是单次 LLM 包装调用做完即返回那引入持久化和追踪是过度设计多一次数据库写入反而增加延迟。如果业务对响应时间极度敏感每个请求都要在 100ms 内返回那么每次 subagent 状态都落库也不合适这时候应该改成异步事件写入。安全边界必须说清楚。持久化意味着把 Agent 的输入输出、prompt 内容、工具调用参数都写进数据库这些数据里很可能包含个人信息、商业机密、内部代码。所以这个方案落地时至少要配套三样东西数据库访问权限控制、敏感字段脱敏、追踪日志的访问审计。另外如果 subagent 的用途涉及人脸、声音、版权素材、个人信息处理必须确认你已经获得合法授权并在文档里写明数据保留周期。3. subagent 工作流的持久化模型设计要让 subagent 工作流可持久、可追踪第一步不是写代码而是把模型设计清楚。直接对着内存里的对象做持久化是没有意义的你需要先拆出任务、事件、追踪三个层次。3.1 任务状态机一个 subagent 任务从创建到结束状态应该明确收敛。参考实现里我建议使用这样的状态机PENDING任务已创建等待 Worker 消费。RUNNING任务正在执行。WAITING任务等待子任务或外部工具结果不消耗计算资源。COMPLETED任务成功结束输出已保存。FAILED任务执行失败保存错误信息。CANCELLED任务被用户或主 Agent 主动取消。INTERRUPTED进程异常退出后恢复时发现任务还在 RUNNING但已无心跳。这里最关键的是 RUNNING 和 INTERRUPTED 的判断逻辑。你没法证明一个进程是死了还是正在慢查询所以通常的做法是给 RUNNING 任务加上心跳时间戳恢复时把超过阈值的 RUNNING 任务标记为 INTERRUPTED再决定是重新入队还是进入失败重试。状态转换要收敛不能随意跳转。合法转换一般是PENDING - RUNNINGRUNNING - WAITING / COMPLETED / FAILED / INTERRUPTEDWAITING - RUNNINGINTERRUPTED - PENDING重试或 FAILED放弃CANCELLED 是终态不参与后续流转3.2 需要持久化的数据很多人以为持久化只需要保存任务状态实际上远远不够。为了能恢复执行至少要把以下四类数据落库第一类是任务元数据。包括 task_id、parent_task_id、agent_name、status、priority、payload、created_at、updated_at、heartbeat_at。payload 可以是一个 JSON 字段存放创建这个 subagent 时的参数。第二类是输入输出数据。每个任务要保存它收到的完整输入、最终输出、以及错误堆栈。LLM 调用是无状态的恢复执行时主 Agent 要根据保存的输入重新发起调用而输出则是后续任务引用它的依据。第三类是上下文消息列表。这一步最容易漏。如果你的 subagent 是多轮对话式的比如先让模型读文档、再根据用户问题回答那中间每一轮的消息对象都要保存。恢复时如果不能把 context_messages 完整还原任务即使状态恢复到了 RUNNING本质上也跑不对。第四类是执行事件和追踪数据。事件表记录状态变更、模型调用、工具调用、重试原因追踪数据记录 trace_id、span_id、parent_span_id、model_name、prompt_tokens、completion_tokens、duration_ms。持久化对象不要写成一张大表建议拆成 tasks、task_events、llm_spans 三张表。tasks 表负责当前状态task_events 表负责完整历史llm_spans 表负责追踪明细。查询当前状态只看 tasks 表分析历史或者做审计再查事件和 span 表这样读写都能保持简单。4. 本地环境准备与依赖选型这个项目方向不挑硬件重点是软件依赖。按最小可运行方案推荐以下环境Python 3.10 或更高版本。SQLite 作为初始存储后续切 PostgreSQL 或 MySQL。Redis 可选用于 Worker 队列、任务心跳、分布式锁。单机小规模任务可以先不引入。Agent 运行库比如 openai、litellm 或你自己封装的 LLM 客户端。Web 框架推荐 FastAPI提供查询 API 和简单看板。如果你基于现有 Agent 框架比如 LangGraph 或 CrewAI这个方案可以包在外层不必侵入框架内部。也就是在创建 subagent 的入口统一记录任务在任务结束的出口统一更新状态。标题里说 ordinary就是这个意思不改框架只加钩子。通用检查清单如下检查项要求操作系统Windows / macOS / Linux 均可Python 版本3.10数据库SQLite 开发PostgreSQL 生产队列可选Redis 5.0磁盘空间根据持久化数据量预估预留日志和事件表增长网络能访问 LLM API或本地模型服务GPU不需要除非 subagent 本身要跑本地模型注意这里没有写死任何第三方库的精确版本号。实际项目里依赖版本以项目 requirements.txt 或 pyproject.toml 为准避免直接用最新版导致的不兼容。5. 部署启动从数据库模型到任务执行器因为没有具体仓库地址这里给出一套通用的实现模板。你可以把它当作参考骨架根据自己项目的目录结构替换路径和配置。5.1 初始化数据库模型使用 SQLAlchemy 定义三张核心表以下是一个可运行的参考模型。# models.py from datetime import datetime from sqlalchemy import String, Integer, JSON, DateTime, ForeignKey from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class Task(Base): __tablename__ tasks id: Mapped[str] mapped_column(String(64), primary_keyTrue) parent_task_id: Mapped[str | None] mapped_column(String(64), indexTrue) agent_name: Mapped[str] mapped_column(String(128), indexTrue) status: Mapped[str] mapped_column(String(32), indexTrue) payload: Mapped[dict] mapped_column(JSON, defaultdict) result: Mapped[dict | None] mapped_column(JSON, nullableTrue) error: Mapped[str | None] mapped_column(String, nullableTrue) heartbeat_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) updated_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) class TaskEvent(Base): __tablename__ task_events id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) task_id: Mapped[str] mapped_column(String(64), indexTrue) event_type: Mapped[str] mapped_column(String(64)) payload: Mapped[dict] mapped_column(JSON, defaultdict) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class LLMSpan(Base): __tablename__ llm_spans id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) trace_id: Mapped[str] mapped_column(String(64), indexTrue) span_id: Mapped[str] mapped_column(String(64), indexTrue) parent_span_id: Mapped[str | None] mapped_column(String(64), nullableTrue) task_id: Mapped[str] mapped_column(String(64), indexTrue) model_name: Mapped[str] mapped_column(String(128)) prompt: Mapped[dict] mapped_column(JSON, defaultdict) response: Mapped[dict | None] mapped_column(JSON, nullableTrue) prompt_tokens: Mapped[int] mapped_column(Integer, default0) completion_tokens: Mapped[int] mapped_column(Integer, default0) duration_ms: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow)这张表的重点有两个tasks 表的 heartbeat_at 字段用于恢复时判断任务是否已中断llm_spans 表的 trace_id 和 parent_span_id 字段用于还原完整的调用树。没有这两个字段后面的恢复和追踪都无从谈起。5.2 任务状态变更封装所有状态变更统一走一个接口不要到处直接 update status。这样可以在每次变更时同时写事件表保证状态和事件一致。# task_store.py from datetime import datetime from sqlalchemy.orm import Session from models import Task, TaskEvent def update_task_status( db: Session, task_id: str, status: str, result: dict | None None, error: str | None None, ): task db.get(Task, task_id) if task is None: raise KeyError(ftask {task_id} not found) task.status status task.updated_at datetime.utcnow() if result is not None: task.result result if error is not None: task.error error db.add(TaskEvent( task_idtask_id, event_typestatus_changed, payload{ from: task.status, to: status, at: datetime.utcnow().isoformat(), }, )) db.commit()很多 Agent 框架里 subagent 的返回值就是普通函数返回值你可以把这个 update_task_status 放在每个 subagent 函数的 finally 块里确保无论成功还是异常都走状态更新。5.3 保存与恢复机制持久化做完之后恢复机制是这个项目真正值钱的部分。恢复逻辑要做两件事第一启动时扫描所有 RUNNING 但心跳过期的任务第二根据任务类型决定重新入队还是标记失败。# recovery.py from datetime import datetime, timedelta from sqlalchemy import select from sqlalchemy.orm import Session from models import Task def recover_interrupted_tasks(db: Session, timeout_seconds: int 300): cutoff datetime.utcnow() - timedelta(secondstimeout_seconds) running_tasks db.scalars( select(Task).where( Task.status RUNNING, Task.heartbeat_at cutoff, ) ).all() recovered [] for task in running_tasks: # 如果当前 Agent 使用幂等重试可以直接回到 PENDING task.status PENDING task.updated_at datetime.utcnow() db.add(TaskEvent( task_idtask.id, event_typerecovered, payload{reason: heartbeat_timeout, timeout_seconds: timeout_seconds}, )) recovered.append(task.id) db.commit() return recovered这段代码对应的是标题里的 persistent。所谓持久化不只是一个概念而是明确回答了“如果系统刚才挂了这个 subagent 任务怎么办”的问题。恢复时是否重新执行取决于你的 subagent 函数是否具备幂等性。如果 subagent 执行的是不可幂等的操作比如发送邮件、扣减库存那恢复时不能直接重试必须先把任务标记为 INTERRUPTED再做人工或补偿处理。6. 功能测试与效果验证环境跑起来以后建议按下面的顺序验证功能。不要一上来就堆几十个任务先把单任务、恢复、追踪链路分别测过再上批量。6.1 验证任务持久化测试目的确认 subagent 任务从 PENDING 到 COMPLETED 的全过程都写入了数据库。操作步骤启动数据库和 Worker。创建一个最简单 subagent 任务输入一条 prompt。观察 tasks 表和 task_events 表的数据变化。预期结果tasks 表出现一条记录状态从 PENDING 变为 RUNNING最后变为 COMPLETED。task_events 表至少有三条 status_changed 事件。判断成功标准进程重启后tasks 表里的 COMPLETED 记录仍然存在查询接口可以正常返回结果。6.2 验证崩溃恢复测试目的确认任务执行到一半进程被 kill重启后能找回并重新入队。操作步骤创建一个 subagent 任务在 subagent 函数里加一个 sleep人为拉长执行时间。在 RUNNING 状态下直接 kill 掉 Worker 进程。修改心跳超时阈值为 10 秒重启 Worker。查看恢复模块扫描结果。预期结果重启后这个任务的状态由 RUNNING 变成 PENDING。日志里出现 recovered 事件。Worker 重新消费任务最终进入 COMPLETED 或 FAILED。常见失败原因如果恢复后重新执行需要完整的上下文但 context_messages 没有持久化任务会在第二次调用时抛出“上下文不存在”的错误。这就是前面强调上下文必须持久化的原因。6.3 验证调用链追踪测试目的确认每次 LLM 调用都记录了 trace_id、token 和耗时。操作步骤创建一个主任务内部派发两个 subagent。每个 subagent 至少调用一次 LLM。查询 llm_spans 表按 trace_id 分组。预期结果主任务生成一个 trace_id。两个 subagent 的调用都有自己的 span_id并通过 parent_span_id 指向主任务。prompt_tokens、completion_tokens、duration_ms 都有实际数值。判断成功标准能从 API 接口返回一棵完整调用树结构类似主 Agent - subagent A - LLM 调用主 Agent - subagent B - LLM 调用。6.4 验证批量任务测试目的确认多个 subagent 任务可以并发消费失败任务不阻塞其他任务。操作步骤构造一个包含 10 个任务的批量提交。其中一条故意使用一个会报错的 prompt模拟模型调用失败。观察队列消费结果。预期结果9 个任务进入 COMPLETED1 个进入 FAILED。失败任务的 error 字段有错误信息。成功任务不受失败任务影响。7. 接口 API 与批量任务接入持久化和追踪做出来之后一定要暴露成 API否则别人没法用。下面是一套通用接口设计你拿到具体项目后按它的路由调整即可。7.1 任务 API任务 API 最少需要三个接口创建任务、查询任务状态、取消任务。# api.py 片段使用 FastAPI from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class TaskCreate(BaseModel): agent_name: str payload: dict parent_task_id: str | None None class TaskOut(BaseModel): id: str status: str result: dict | None error: str | None app.post(/api/tasks, response_modelTaskOut) def create_task(body: TaskCreate): # 生成 task_id写入 tasks 表状态为 PENDING投递到队列 ... app.get(/api/tasks/{task_id}, response_modelTaskOut) def get_task(task_id: str): # 查询 tasks 表返回当前状态、result、error ... app.post(/api/tasks/{task_id}/cancel) def cancel_task(task_id: str): # 只有 PENDING / RUNNING 状态可以取消终态不接受 ...调用示例curl -X POST http://127.0.0.1:8000/api/tasks \ -H Content-Type: application/json \ -d { agent_name: research_agent, payload: {query: how to design a persistent agent workflow} }curl http://127.0.0.1:8000/api/tasks/8f2c1a3e-4b5d-4c7e-9f0a-1b2c3d4e5f6a这个接口是 Worker 的入口也是你把自己现有 Agent 系统接进来的桥梁。不需要改 Agent 内部逻辑只需要在创建 subagent 的地方调用 create_task在完成时把结果传回来。7.2 追踪查询 API追踪查询 API 是排查问题的关键。输入一个 task_id 或 trace_id返回完整的调用链。app.get(/api/traces/{trace_id}) def get_trace(trace_id: str): # 查询 llm_spans 表按 parent_span_id 递归组装成树 ...返回结构示例{ trace_id: trace_001, root_task_id: task_main_001, spans: [ { span_id: span_root, parent_span_id: null, model_name: gpt-4o, duration_ms: 2500, prompt_tokens: 800, completion_tokens: 200, status: ok }, { span_id: span_sub_1, parent_span_id: span_root, model_name: gpt-4o-mini, duration_ms: 800, prompt_tokens: 300, completion_tokens: 80, status: ok } ] }有了这个接口主 Agent 调了哪些 subagent、每个 subagent 调了什么模型、花了多少 token、每步耗时多少就变成了可查询的数据而不只是日志里的一行模糊输出。7.3 批量提交批量提交不是简单地在 for 循环里调 create_task。更稳妥的做法是把批量任务作为一条父任务记录子任务各自入队。import requests base_url http://127.0.0.1:8000 # 1. 创建批量父任务 parent requests.post( f{base_url}/api/tasks, json{ agent_name: batch_processor, payload: {items: [ffile_{i}.md for i in range(10)]}, }, ).json() # 2. 针对每个 item 创建子任务 subtask_ids [] for item in parent[payload][items]: subtask requests.post( f{base_url}/api/tasks, json{ parent_task_id: parent[id], agent_name: file_summarizer, payload: {file: item}, }, ).json() subtask_ids.append(subtask[id])批量任务需要注意三件事子任务之间是否相互独立失败重试策略是整体重试还是单条重试是否需要等待所有子任务完成再聚合结果。推荐的做法是单条重试子任务失败不阻塞其他子任务父任务在所有子任务完成后自动汇总。如果你想自己实现不依赖外部队列也可以用 asyncio.Queue 做一个进程内队列加一个消费循环。但这种方案适合单机小规模任务遇到多 Worker 或需要任务持久化排队的情况还是要引入 Redis 或 RabbitMQ确保任务在 Worker 崩溃后不丢。8. 资源占用与性能观察这个方案的主要资源消耗不在模型推理而在数据库写入和队列消费。你需要重点观察四个指标每次任务状态变更产生 1 次 tasks 表更新和 1 次 task_events 插入。任务量很大的时候这个写入频率会直接决定数据库压力。每次 LLM 调用产生 1 条 llm_spans 记录。如果 subagent 内部有多次工具调用还要在事件表里记录工具调用事件量级会成倍增长。进程重启后的恢复扫描如果任务表数据量很大建议给 status 和 heartbeat_at 加联合索引避免全表扫描。队列积压情况Worker 消费速度跟不上生产速度时任务会长期停留在 PENDING。在开发阶段SQLite 足够你用数据量通常也就几千条。到了生产环境必须切 PostgreSQL因为并发写入和索引查询的差距会非常明显。另外llm_spans 表的数据增长往往比 tasks 表快得多因为它记录了每次调用的完整 prompt 和 response建议做定期归档或者只保留 prompt 摘要和 token 统计完整请求体写到对象存储里。如果发现响应变慢优先检查是不是每次状态变更都同步 commit。批量场景下可以把多个状态变更合并成一次事务提交或者使用异步方式写事件表让主流程不被磁盘写入拖慢。9. 常见问题与排查方法这里整理几个最容易踩的坑和排查路径。问题现象可能原因排查方式解决方案任务一直处于 RUNNING 状态Worker 崩溃后没有心跳更新且恢复机制未启动查询任务的 heartbeat_at 是否早于当前时间增加心跳更新循环启动时执行 recover_interrupted_tasks恢复后任务重新执行但上下文不完整只保存了任务状态没有保存 context_messages检查 tasks 表 payload 或关联表是否包含历史消息把多轮消息数组完整持久化调用链里没有 LLM 调用记录LLM 调用没有走统一的记录包装器查看代码是否直接在业务函数里调用模型封装一个 llm_call 函数统一写 llm_spans事件表只有状态变更没有错误原因异常被吞掉没有写入 error 字段检查 subagent 函数是否有 try/except异常时调用 update_task_status(statusFAILED, error堆栈)批量任务中一个失败影响其他所有任务使用了全局异常终止逻辑看队列消费代码是否在异常时 break单个任务异常只标记该任务失败继续消费队列存储增长过快每次调用都保存了完整 prompt 和 response查看 llm_spans 表数据量开启归档或者只保留摘要重复执行产生重复结果任务不是幂等的恢复逻辑直接重试检查任务是否写了外部系统非幂等任务恢复时标记 INTERRUPTED人工介入进程重启后任务丢失任务没有入持久化队列或者任务表未写入就投递检查创建任务的顺序先写 tasks 表再投递队列保证二者一致这里面最值得关注的是最后两个。持久化虽然能保存状态但如果你不处理幂等性恢复机制反而可能造成重复操作。非幂等任务要么不自动重试要么在重试前做幂等检查这是生产环境必须考虑的问题。10. 最佳实践与使用建议基于这类系统常见的落地经验给你几条具体建议。第一第一次跑通先用最小配置。不要一上来就接 Redis、PostgreSQL、Agent 框架全家桶。先用 SQLite 一个最简单的 subagent 函数验证任务能从 PENDING 走到 COMPLETED再逐步加恢复、追踪、队列。环境越简单定位问题越快。第二保留一套最小可运行配置。把模型定义、任务状态机、恢复逻辑、一个示例 subagent 放在一起作为项目骨架。后续加新 Agent 类型只需要在统一入口注册新的 agent_name不需要重写持久化逻辑。第三任务数据分目录管理。tasks 表、事件表、模型调用表分开存放输入素材、中间产物、输出结果在文件系统里分目录。不要在 payload 里塞大量文件内容存文件路径即可。第四所有 LLM 调用必须统一封装。不要每个 subagent 各自直接调用 OpenAI SDK。封装一个 llm_call 函数内部完成调用、计时、token 统计、llm_spans 写入。只有统一封装追踪数据才会完整。第五批量任务要加日志、心跳和失败重试。每个任务都要有 trace_id日志里打印 task_id、subagent 名称、状态变更。不要让 Worker 默默消费队列要能看到每个任务的执行轨迹。第六接口服务要限制访问范围。任务 API 会暴露 subagent 的输入输出默认不要监听 0.0.0.0最好绑定 127.0.0.1如果有跨机器调用需求加认证和访问控制。第七涉及隐私和数据合规时先确认授权。subagent 处理的数据可能包含个人信息、受版权保护的素材或内部业务数据。持久化之前评估哪些字段需要加密追踪日志里是否可能包含敏感 prompt 内容发布或商用前做好效果复核。11. 总结与下一步这个项目方向最值得尝试的点在于它不需要改造你现有的 Agent 框架只要在 subagent 创建和结束的地方加持久化和追踪钩子就能解决任务中断恢复、调用链审计、token 成本统计这些生产问题。你最先应该验证的是单任务从创建到完成的持久化流程然后人为 kill 进程验证恢复最后再做批量任务。最容易踩的坑有两个一是只保存任务状态却没有保存上下文消息导致恢复后的任务技术上“恢复”了、实际上一执行就错二是所有 LLM 调用没有统一封装追踪数据缺失最后又回到对着日志猜问题的状态。接下来可以扩展的方向包括把追踪数据接到 OpenTelemetry做可视化调用链增加 tag 和 label 支持按业务线、模型、时间范围做成本分析在任务事件上做更细粒度的审计报表把恢复策略从简单的重试升级为检查和补偿机制让非幂等任务也能安全恢复。如果你正在做 Agent 系统目前卡在“任务跑起来之后没法复盘、进程一停就前功尽弃”这个阶段建议直接按上面的模型落地一版数据库选 SQLite 先跑通验证完恢复和追踪的价值再上生产配置。 subagent工作流持久化与追踪：让AI Agent任务可恢复可审计
这次我们来看一个最近在 Show HN 上出现的方向让普通的 subagent 工作流变得持久化persistent且可追踪trackable。Show HN 是 Hacker News 上作者展示自己开源项目的栏目所以这个标题翻译过来就是“给我那个普通子智能体工作流加上保存和追踪能力”。用大白话说它解决的是 AI Agent 开发里一个特别常见的痛点主 Agent 派了一堆 subagent 干活任务执行过程中上下文、中间结果、调用记录全在内存里进程一崩就全没了出了问题也没法复盘是哪个子任务花了多少 token、哪一步失败的。这个项目方向的三个关键词已经说明了核心卖点ordinary不要求你推倒现有 Agent 框架重来persistent任务状态、上下文、中间产物可以落盘恢复trackable每次 subagent 调用的输入输出、耗时、token 消耗都可以查询和审计。下面我会先从核心能力、适用场景开始然后给出一个可落地的持久化模型设计再带你把任务执行器、恢复机制、追踪查询 API、批量任务队列完整走一遍最后补上常见的坑和工程化建议。1. 核心能力速览在直接看代码之前先把这个项目方向能做什么、不能做什么放在一张表里。因为标题只给出了能力边界没有具体的仓库地址和版本所以表格里的参数我尽量写成通用范围实际以你拿到源码后的文档为准。能力项说明项目类型AI Agent 编排增强层 / subagent 工作流管理工具核心功能把普通 subagent 任务的状态持久化到数据库并提供完整的调用追踪能力持久化范围任务元数据、输入输出、上下文消息、错误信息、执行事件追踪维度调用链trace、状态变更历史、耗时、token 用量、模型名称恢复能力进程重启后自动找回中断任务支持重新执行或标记失败重试存储依赖推荐 SQLite 起步生产环境使用 PostgreSQL / MySQLRedis 用于队列和心跳硬件门槛极低普通开发机即可主要消耗 CPU、内存、磁盘GPU 不是必须的启动方式源码运行提供任务执行器、Worker 进程、API 服务三部分是否支持 API支持通用 REST 接口可用于查询任务状态和追踪链是否支持批量任务支持通过任务队列批量提交 subagent 工作流适合场景本地调试、生产 Agent 系统、自动化流水线、审计合规这里需要强调一点标题里的 persistent 和“persistent homology”没有任何关系后者是拓扑数据分析里的持续同调概念属于数学方向。这个项目里的 persistent 指的是任务状态在进程退出、崩溃、重启之后仍然存在可以继续执行或回溯。2. 适用场景与使用边界这个方案适合谁最典型的是已经在用 LangGraph、AutoGen、CrewAI 这类框架或者自己手写 Agent 循环的团队。你现在的 subagent 流程大概率是这样的主 Agent 拆解任务创建 subagentsubagent 调用 LLM把结果返回主 Agent流程结束。看起来简单但一旦进入生产环境你会发现三件事没法回答任务执行到一半服务重启了这个任务是从头再来还是接着跑某个 subagent 失败是因为 prompt 问题还是模型返回异常这个月所有 subagent 调用合计花了多少 token、多少时间持久化和追踪就是来解决这三个问题的。它特别适合以下场景长时间运行的多步骤任务比如研究报告生成、批量数据处理、多轮代码审查这类任务分钟级起步中断成本高。需要审计和复盘的生产环境比如客服工单处理、内容审核流程、金融文档分析你需要知道每一步是谁、什么模型、什么输入、什么输出。批量并发的 subagent 调度几十个文件、几十个 URL 需要同时处理必须有一个可靠的任务队列和失败重试机制。也存在不适合的情况。如果你的调用链极短比如只是单次 LLM 包装调用做完即返回那引入持久化和追踪是过度设计多一次数据库写入反而增加延迟。如果业务对响应时间极度敏感每个请求都要在 100ms 内返回那么每次 subagent 状态都落库也不合适这时候应该改成异步事件写入。安全边界必须说清楚。持久化意味着把 Agent 的输入输出、prompt 内容、工具调用参数都写进数据库这些数据里很可能包含个人信息、商业机密、内部代码。所以这个方案落地时至少要配套三样东西数据库访问权限控制、敏感字段脱敏、追踪日志的访问审计。另外如果 subagent 的用途涉及人脸、声音、版权素材、个人信息处理必须确认你已经获得合法授权并在文档里写明数据保留周期。3. subagent 工作流的持久化模型设计要让 subagent 工作流可持久、可追踪第一步不是写代码而是把模型设计清楚。直接对着内存里的对象做持久化是没有意义的你需要先拆出任务、事件、追踪三个层次。3.1 任务状态机一个 subagent 任务从创建到结束状态应该明确收敛。参考实现里我建议使用这样的状态机PENDING任务已创建等待 Worker 消费。RUNNING任务正在执行。WAITING任务等待子任务或外部工具结果不消耗计算资源。COMPLETED任务成功结束输出已保存。FAILED任务执行失败保存错误信息。CANCELLED任务被用户或主 Agent 主动取消。INTERRUPTED进程异常退出后恢复时发现任务还在 RUNNING但已无心跳。这里最关键的是 RUNNING 和 INTERRUPTED 的判断逻辑。你没法证明一个进程是死了还是正在慢查询所以通常的做法是给 RUNNING 任务加上心跳时间戳恢复时把超过阈值的 RUNNING 任务标记为 INTERRUPTED再决定是重新入队还是进入失败重试。状态转换要收敛不能随意跳转。合法转换一般是PENDING - RUNNINGRUNNING - WAITING / COMPLETED / FAILED / INTERRUPTEDWAITING - RUNNINGINTERRUPTED - PENDING重试或 FAILED放弃CANCELLED 是终态不参与后续流转3.2 需要持久化的数据很多人以为持久化只需要保存任务状态实际上远远不够。为了能恢复执行至少要把以下四类数据落库第一类是任务元数据。包括 task_id、parent_task_id、agent_name、status、priority、payload、created_at、updated_at、heartbeat_at。payload 可以是一个 JSON 字段存放创建这个 subagent 时的参数。第二类是输入输出数据。每个任务要保存它收到的完整输入、最终输出、以及错误堆栈。LLM 调用是无状态的恢复执行时主 Agent 要根据保存的输入重新发起调用而输出则是后续任务引用它的依据。第三类是上下文消息列表。这一步最容易漏。如果你的 subagent 是多轮对话式的比如先让模型读文档、再根据用户问题回答那中间每一轮的消息对象都要保存。恢复时如果不能把 context_messages 完整还原任务即使状态恢复到了 RUNNING本质上也跑不对。第四类是执行事件和追踪数据。事件表记录状态变更、模型调用、工具调用、重试原因追踪数据记录 trace_id、span_id、parent_span_id、model_name、prompt_tokens、completion_tokens、duration_ms。持久化对象不要写成一张大表建议拆成 tasks、task_events、llm_spans 三张表。tasks 表负责当前状态task_events 表负责完整历史llm_spans 表负责追踪明细。查询当前状态只看 tasks 表分析历史或者做审计再查事件和 span 表这样读写都能保持简单。4. 本地环境准备与依赖选型这个项目方向不挑硬件重点是软件依赖。按最小可运行方案推荐以下环境Python 3.10 或更高版本。SQLite 作为初始存储后续切 PostgreSQL 或 MySQL。Redis 可选用于 Worker 队列、任务心跳、分布式锁。单机小规模任务可以先不引入。Agent 运行库比如 openai、litellm 或你自己封装的 LLM 客户端。Web 框架推荐 FastAPI提供查询 API 和简单看板。如果你基于现有 Agent 框架比如 LangGraph 或 CrewAI这个方案可以包在外层不必侵入框架内部。也就是在创建 subagent 的入口统一记录任务在任务结束的出口统一更新状态。标题里说 ordinary就是这个意思不改框架只加钩子。通用检查清单如下检查项要求操作系统Windows / macOS / Linux 均可Python 版本3.10数据库SQLite 开发PostgreSQL 生产队列可选Redis 5.0磁盘空间根据持久化数据量预估预留日志和事件表增长网络能访问 LLM API或本地模型服务GPU不需要除非 subagent 本身要跑本地模型注意这里没有写死任何第三方库的精确版本号。实际项目里依赖版本以项目 requirements.txt 或 pyproject.toml 为准避免直接用最新版导致的不兼容。5. 部署启动从数据库模型到任务执行器因为没有具体仓库地址这里给出一套通用的实现模板。你可以把它当作参考骨架根据自己项目的目录结构替换路径和配置。5.1 初始化数据库模型使用 SQLAlchemy 定义三张核心表以下是一个可运行的参考模型。# models.py from datetime import datetime from sqlalchemy import String, Integer, JSON, DateTime, ForeignKey from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class Task(Base): __tablename__ tasks id: Mapped[str] mapped_column(String(64), primary_keyTrue) parent_task_id: Mapped[str | None] mapped_column(String(64), indexTrue) agent_name: Mapped[str] mapped_column(String(128), indexTrue) status: Mapped[str] mapped_column(String(32), indexTrue) payload: Mapped[dict] mapped_column(JSON, defaultdict) result: Mapped[dict | None] mapped_column(JSON, nullableTrue) error: Mapped[str | None] mapped_column(String, nullableTrue) heartbeat_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) updated_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) class TaskEvent(Base): __tablename__ task_events id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) task_id: Mapped[str] mapped_column(String(64), indexTrue) event_type: Mapped[str] mapped_column(String(64)) payload: Mapped[dict] mapped_column(JSON, defaultdict) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class LLMSpan(Base): __tablename__ llm_spans id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) trace_id: Mapped[str] mapped_column(String(64), indexTrue) span_id: Mapped[str] mapped_column(String(64), indexTrue) parent_span_id: Mapped[str | None] mapped_column(String(64), nullableTrue) task_id: Mapped[str] mapped_column(String(64), indexTrue) model_name: Mapped[str] mapped_column(String(128)) prompt: Mapped[dict] mapped_column(JSON, defaultdict) response: Mapped[dict | None] mapped_column(JSON, nullableTrue) prompt_tokens: Mapped[int] mapped_column(Integer, default0) completion_tokens: Mapped[int] mapped_column(Integer, default0) duration_ms: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow)这张表的重点有两个tasks 表的 heartbeat_at 字段用于恢复时判断任务是否已中断llm_spans 表的 trace_id 和 parent_span_id 字段用于还原完整的调用树。没有这两个字段后面的恢复和追踪都无从谈起。5.2 任务状态变更封装所有状态变更统一走一个接口不要到处直接 update status。这样可以在每次变更时同时写事件表保证状态和事件一致。# task_store.py from datetime import datetime from sqlalchemy.orm import Session from models import Task, TaskEvent def update_task_status( db: Session, task_id: str, status: str, result: dict | None None, error: str | None None, ): task db.get(Task, task_id) if task is None: raise KeyError(ftask {task_id} not found) task.status status task.updated_at datetime.utcnow() if result is not None: task.result result if error is not None: task.error error db.add(TaskEvent( task_idtask_id, event_typestatus_changed, payload{ from: task.status, to: status, at: datetime.utcnow().isoformat(), }, )) db.commit()很多 Agent 框架里 subagent 的返回值就是普通函数返回值你可以把这个 update_task_status 放在每个 subagent 函数的 finally 块里确保无论成功还是异常都走状态更新。5.3 保存与恢复机制持久化做完之后恢复机制是这个项目真正值钱的部分。恢复逻辑要做两件事第一启动时扫描所有 RUNNING 但心跳过期的任务第二根据任务类型决定重新入队还是标记失败。# recovery.py from datetime import datetime, timedelta from sqlalchemy import select from sqlalchemy.orm import Session from models import Task def recover_interrupted_tasks(db: Session, timeout_seconds: int 300): cutoff datetime.utcnow() - timedelta(secondstimeout_seconds) running_tasks db.scalars( select(Task).where( Task.status RUNNING, Task.heartbeat_at cutoff, ) ).all() recovered [] for task in running_tasks: # 如果当前 Agent 使用幂等重试可以直接回到 PENDING task.status PENDING task.updated_at datetime.utcnow() db.add(TaskEvent( task_idtask.id, event_typerecovered, payload{reason: heartbeat_timeout, timeout_seconds: timeout_seconds}, )) recovered.append(task.id) db.commit() return recovered这段代码对应的是标题里的 persistent。所谓持久化不只是一个概念而是明确回答了“如果系统刚才挂了这个 subagent 任务怎么办”的问题。恢复时是否重新执行取决于你的 subagent 函数是否具备幂等性。如果 subagent 执行的是不可幂等的操作比如发送邮件、扣减库存那恢复时不能直接重试必须先把任务标记为 INTERRUPTED再做人工或补偿处理。6. 功能测试与效果验证环境跑起来以后建议按下面的顺序验证功能。不要一上来就堆几十个任务先把单任务、恢复、追踪链路分别测过再上批量。6.1 验证任务持久化测试目的确认 subagent 任务从 PENDING 到 COMPLETED 的全过程都写入了数据库。操作步骤启动数据库和 Worker。创建一个最简单 subagent 任务输入一条 prompt。观察 tasks 表和 task_events 表的数据变化。预期结果tasks 表出现一条记录状态从 PENDING 变为 RUNNING最后变为 COMPLETED。task_events 表至少有三条 status_changed 事件。判断成功标准进程重启后tasks 表里的 COMPLETED 记录仍然存在查询接口可以正常返回结果。6.2 验证崩溃恢复测试目的确认任务执行到一半进程被 kill重启后能找回并重新入队。操作步骤创建一个 subagent 任务在 subagent 函数里加一个 sleep人为拉长执行时间。在 RUNNING 状态下直接 kill 掉 Worker 进程。修改心跳超时阈值为 10 秒重启 Worker。查看恢复模块扫描结果。预期结果重启后这个任务的状态由 RUNNING 变成 PENDING。日志里出现 recovered 事件。Worker 重新消费任务最终进入 COMPLETED 或 FAILED。常见失败原因如果恢复后重新执行需要完整的上下文但 context_messages 没有持久化任务会在第二次调用时抛出“上下文不存在”的错误。这就是前面强调上下文必须持久化的原因。6.3 验证调用链追踪测试目的确认每次 LLM 调用都记录了 trace_id、token 和耗时。操作步骤创建一个主任务内部派发两个 subagent。每个 subagent 至少调用一次 LLM。查询 llm_spans 表按 trace_id 分组。预期结果主任务生成一个 trace_id。两个 subagent 的调用都有自己的 span_id并通过 parent_span_id 指向主任务。prompt_tokens、completion_tokens、duration_ms 都有实际数值。判断成功标准能从 API 接口返回一棵完整调用树结构类似主 Agent - subagent A - LLM 调用主 Agent - subagent B - LLM 调用。6.4 验证批量任务测试目的确认多个 subagent 任务可以并发消费失败任务不阻塞其他任务。操作步骤构造一个包含 10 个任务的批量提交。其中一条故意使用一个会报错的 prompt模拟模型调用失败。观察队列消费结果。预期结果9 个任务进入 COMPLETED1 个进入 FAILED。失败任务的 error 字段有错误信息。成功任务不受失败任务影响。7. 接口 API 与批量任务接入持久化和追踪做出来之后一定要暴露成 API否则别人没法用。下面是一套通用接口设计你拿到具体项目后按它的路由调整即可。7.1 任务 API任务 API 最少需要三个接口创建任务、查询任务状态、取消任务。# api.py 片段使用 FastAPI from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class TaskCreate(BaseModel): agent_name: str payload: dict parent_task_id: str | None None class TaskOut(BaseModel): id: str status: str result: dict | None error: str | None app.post(/api/tasks, response_modelTaskOut) def create_task(body: TaskCreate): # 生成 task_id写入 tasks 表状态为 PENDING投递到队列 ... app.get(/api/tasks/{task_id}, response_modelTaskOut) def get_task(task_id: str): # 查询 tasks 表返回当前状态、result、error ... app.post(/api/tasks/{task_id}/cancel) def cancel_task(task_id: str): # 只有 PENDING / RUNNING 状态可以取消终态不接受 ...调用示例curl -X POST http://127.0.0.1:8000/api/tasks \ -H Content-Type: application/json \ -d { agent_name: research_agent, payload: {query: how to design a persistent agent workflow} }curl http://127.0.0.1:8000/api/tasks/8f2c1a3e-4b5d-4c7e-9f0a-1b2c3d4e5f6a这个接口是 Worker 的入口也是你把自己现有 Agent 系统接进来的桥梁。不需要改 Agent 内部逻辑只需要在创建 subagent 的地方调用 create_task在完成时把结果传回来。7.2 追踪查询 API追踪查询 API 是排查问题的关键。输入一个 task_id 或 trace_id返回完整的调用链。app.get(/api/traces/{trace_id}) def get_trace(trace_id: str): # 查询 llm_spans 表按 parent_span_id 递归组装成树 ...返回结构示例{ trace_id: trace_001, root_task_id: task_main_001, spans: [ { span_id: span_root, parent_span_id: null, model_name: gpt-4o, duration_ms: 2500, prompt_tokens: 800, completion_tokens: 200, status: ok }, { span_id: span_sub_1, parent_span_id: span_root, model_name: gpt-4o-mini, duration_ms: 800, prompt_tokens: 300, completion_tokens: 80, status: ok } ] }有了这个接口主 Agent 调了哪些 subagent、每个 subagent 调了什么模型、花了多少 token、每步耗时多少就变成了可查询的数据而不只是日志里的一行模糊输出。7.3 批量提交批量提交不是简单地在 for 循环里调 create_task。更稳妥的做法是把批量任务作为一条父任务记录子任务各自入队。import requests base_url http://127.0.0.1:8000 # 1. 创建批量父任务 parent requests.post( f{base_url}/api/tasks, json{ agent_name: batch_processor, payload: {items: [ffile_{i}.md for i in range(10)]}, }, ).json() # 2. 针对每个 item 创建子任务 subtask_ids [] for item in parent[payload][items]: subtask requests.post( f{base_url}/api/tasks, json{ parent_task_id: parent[id], agent_name: file_summarizer, payload: {file: item}, }, ).json() subtask_ids.append(subtask[id])批量任务需要注意三件事子任务之间是否相互独立失败重试策略是整体重试还是单条重试是否需要等待所有子任务完成再聚合结果。推荐的做法是单条重试子任务失败不阻塞其他子任务父任务在所有子任务完成后自动汇总。如果你想自己实现不依赖外部队列也可以用 asyncio.Queue 做一个进程内队列加一个消费循环。但这种方案适合单机小规模任务遇到多 Worker 或需要任务持久化排队的情况还是要引入 Redis 或 RabbitMQ确保任务在 Worker 崩溃后不丢。8. 资源占用与性能观察这个方案的主要资源消耗不在模型推理而在数据库写入和队列消费。你需要重点观察四个指标每次任务状态变更产生 1 次 tasks 表更新和 1 次 task_events 插入。任务量很大的时候这个写入频率会直接决定数据库压力。每次 LLM 调用产生 1 条 llm_spans 记录。如果 subagent 内部有多次工具调用还要在事件表里记录工具调用事件量级会成倍增长。进程重启后的恢复扫描如果任务表数据量很大建议给 status 和 heartbeat_at 加联合索引避免全表扫描。队列积压情况Worker 消费速度跟不上生产速度时任务会长期停留在 PENDING。在开发阶段SQLite 足够你用数据量通常也就几千条。到了生产环境必须切 PostgreSQL因为并发写入和索引查询的差距会非常明显。另外llm_spans 表的数据增长往往比 tasks 表快得多因为它记录了每次调用的完整 prompt 和 response建议做定期归档或者只保留 prompt 摘要和 token 统计完整请求体写到对象存储里。如果发现响应变慢优先检查是不是每次状态变更都同步 commit。批量场景下可以把多个状态变更合并成一次事务提交或者使用异步方式写事件表让主流程不被磁盘写入拖慢。9. 常见问题与排查方法这里整理几个最容易踩的坑和排查路径。问题现象可能原因排查方式解决方案任务一直处于 RUNNING 状态Worker 崩溃后没有心跳更新且恢复机制未启动查询任务的 heartbeat_at 是否早于当前时间增加心跳更新循环启动时执行 recover_interrupted_tasks恢复后任务重新执行但上下文不完整只保存了任务状态没有保存 context_messages检查 tasks 表 payload 或关联表是否包含历史消息把多轮消息数组完整持久化调用链里没有 LLM 调用记录LLM 调用没有走统一的记录包装器查看代码是否直接在业务函数里调用模型封装一个 llm_call 函数统一写 llm_spans事件表只有状态变更没有错误原因异常被吞掉没有写入 error 字段检查 subagent 函数是否有 try/except异常时调用 update_task_status(statusFAILED, error堆栈)批量任务中一个失败影响其他所有任务使用了全局异常终止逻辑看队列消费代码是否在异常时 break单个任务异常只标记该任务失败继续消费队列存储增长过快每次调用都保存了完整 prompt 和 response查看 llm_spans 表数据量开启归档或者只保留摘要重复执行产生重复结果任务不是幂等的恢复逻辑直接重试检查任务是否写了外部系统非幂等任务恢复时标记 INTERRUPTED人工介入进程重启后任务丢失任务没有入持久化队列或者任务表未写入就投递检查创建任务的顺序先写 tasks 表再投递队列保证二者一致这里面最值得关注的是最后两个。持久化虽然能保存状态但如果你不处理幂等性恢复机制反而可能造成重复操作。非幂等任务要么不自动重试要么在重试前做幂等检查这是生产环境必须考虑的问题。10. 最佳实践与使用建议基于这类系统常见的落地经验给你几条具体建议。第一第一次跑通先用最小配置。不要一上来就接 Redis、PostgreSQL、Agent 框架全家桶。先用 SQLite 一个最简单的 subagent 函数验证任务能从 PENDING 走到 COMPLETED再逐步加恢复、追踪、队列。环境越简单定位问题越快。第二保留一套最小可运行配置。把模型定义、任务状态机、恢复逻辑、一个示例 subagent 放在一起作为项目骨架。后续加新 Agent 类型只需要在统一入口注册新的 agent_name不需要重写持久化逻辑。第三任务数据分目录管理。tasks 表、事件表、模型调用表分开存放输入素材、中间产物、输出结果在文件系统里分目录。不要在 payload 里塞大量文件内容存文件路径即可。第四所有 LLM 调用必须统一封装。不要每个 subagent 各自直接调用 OpenAI SDK。封装一个 llm_call 函数内部完成调用、计时、token 统计、llm_spans 写入。只有统一封装追踪数据才会完整。第五批量任务要加日志、心跳和失败重试。每个任务都要有 trace_id日志里打印 task_id、subagent 名称、状态变更。不要让 Worker 默默消费队列要能看到每个任务的执行轨迹。第六接口服务要限制访问范围。任务 API 会暴露 subagent 的输入输出默认不要监听 0.0.0.0最好绑定 127.0.0.1如果有跨机器调用需求加认证和访问控制。第七涉及隐私和数据合规时先确认授权。subagent 处理的数据可能包含个人信息、受版权保护的素材或内部业务数据。持久化之前评估哪些字段需要加密追踪日志里是否可能包含敏感 prompt 内容发布或商用前做好效果复核。11. 总结与下一步这个项目方向最值得尝试的点在于它不需要改造你现有的 Agent 框架只要在 subagent 创建和结束的地方加持久化和追踪钩子就能解决任务中断恢复、调用链审计、token 成本统计这些生产问题。你最先应该验证的是单任务从创建到完成的持久化流程然后人为 kill 进程验证恢复最后再做批量任务。最容易踩的坑有两个一是只保存任务状态却没有保存上下文消息导致恢复后的任务技术上“恢复”了、实际上一执行就错二是所有 LLM 调用没有统一封装追踪数据缺失最后又回到对着日志猜问题的状态。接下来可以扩展的方向包括把追踪数据接到 OpenTelemetry做可视化调用链增加 tag 和 label 支持按业务线、模型、时间范围做成本分析在任务事件上做更细粒度的审计报表把恢复策略从简单的重试升级为检查和补偿机制让非幂等任务也能安全恢复。如果你正在做 Agent 系统目前卡在“任务跑起来之后没法复盘、进程一停就前功尽弃”这个阶段建议直接按上面的模型落地一版数据库选 SQLite 先跑通验证完恢复和追踪的价值再上生产配置。