在实际使用 LLM Agent 的项目里subagent 工作流正在成为最常见的任务拆分方式。主代理把“查资料、写代码、算指标、生成报告”这类小任务分别交给子代理执行自己只负责编排和汇总。子代理执行结束后结果返回主代理整个交互随之结束。这种模式很自然但它有一个容易被忽略的短板默认情况下subagent 工作流是瞬时的既不会跨进程保留也不会留下可用于审计和排查的轨迹。标题里的 persistent 和 trackable指的就是把“调用完就消失”的子代理执行过程改造成一个可恢复、可追踪、可复现的工程单元。文章会围绕一个最小可运行的执行器展开。这个执行器接收一个主代理定义的工作流将节点逐一派发给 subagent并把每次执行的状态、事件、输入输出和产物写入 SQLite。之后我会说明为什么用状态机加 append-only 事件日志来建模而不是直接用一张表保存最终结果然后给出完整代码、建表语句、恢复逻辑和查询方式最后补充常见故障、排查链路和生产环境需要补强的地方。读完以后你可以把同一套设计迁移到 LangGraph、AutoGen、CrewAI 或者其他自定义 Agent 框架中。1. 先理解 subagent 工作流为何需要持久化1.1 subagent 工作流到底是什么在 Agent 系统里subagent 是一个独立的执行单元。它接收任务描述、必要上下文和可用工具然后调用大模型推理、执行工具并返回结果。与直接在同一个 prompt 里让模型完成多个步骤不同subagent 方式强调隔离和复用主代理不需要关心子代理内部提示词怎么组织只需要关心子任务是否成功、返回格式是否符合约定。一个典型示例如下主代理负责“规划一次三天旅行”。主代理拆分出“查机票”“查酒店”“查天气”三个子任务。三个子代理分别执行并把结果返回主代理。主代理汇总结果生成最终旅行计划。这个过程如果只是内存里的函数调用运行结束后所有中间状态都会被回收。一旦某个子代理执行到一半出现网络超时、模型 API 报错或者服务进程被重新部署整个工作流就只能从头再跑。1.2 瞬时执行带来的四个问题只依赖内存的 subagent 工作流会带来四个实际工程问题第一进程崩溃后上下文丢失。子代理已经完成的结果没有落盘主代理已经推进到的步骤也没有记录重启后只能重新开始。第二无法做失败重试。重试的前提是知道“失败发生之前完成了什么”。如果没有任何状态记录重试只能退化为整体重跑成本非常高。第三无法审计和解释结果。当领导或客户问“这个结论是哪几个子代理、基于什么输入算出来的”如果你只能打开终端看到一段最新日志很难回答清楚。第四难以并发和调度。多子代理并行执行时调度器需要知道每个节点的状态才能决定哪些节点可以继续、哪些节点已经完成、哪些节点需要重新执行。瞬时执行没有这种信息。1.3 持久化与可追踪的分工持久化解决的是“状态能不能活下来”的问题。它要求把工作流运行进度、节点状态、输入输出写入数据库这样进程重启后可以从最近一个稳定点继续执行。可追踪解决的是“发生了什么”的问题。它要求记录事件序列什么时候创建了任务、什么时候交给子代理、子代理返回了什么、失败原因是什么、重试了几次。事件日志本身也是持久化的但它的设计目的不是恢复执行而是复现执行过程。在这套设计中持久化面向恢复可追踪面向审计与调试。两者结合起来才是完整的工程闭环。1.4 长时间任务更依赖稳定落盘有一类 subagent 任务格外需要持久化计算型任务。例如某个子代理负责计算 cubical persistent homology它需要处理拓扑特征、构造 cubical complex、计算不同参数下的同调群。这类任务可能运行几十分钟产生多个中间结果文件如果中途断掉重复计算的时间成本很难接受。这类场景与常见的“调用一次模型拿一个 JSON”完全不同。LLM 调用通常以秒为单位而计算型子任务可能以分钟甚至小时为单位。对工作流引擎来说任务越重持久化和可追踪的价值越大。状态恢复不是“加分项”而是“继续运行”的基本条件。2. 设计 subagent 工作流的数据模型2.1 核心实体运行、节点、事件、产物要让 subagent 工作流可持久化、可追踪数据库不能只记录“完成了哪几步”。建议拆成四个核心实体workflow_run一次完整的工作流执行包含名称、状态、入口 payload。agent_node工作流里的一个 subagent 调用节点记录输入输出和错误。agent_event每一条执行事件记录“谁在什么时间做了什么”。artifact节点运行过程中产生的结果文件或结构化数据。以“旅行规划”为例一次 workflow_run 对应整个规划过程。查天气、查酒店分别是两个 agent_node。子代理开始执行、成功返回、结果落盘这些行为都对应 agent_event。最终返回的 JSON 包可以存入 artifact。这样建模的好处是职责分离。节点表回答“当前执行到哪一步”事件表回答“每一步都发生了什么”产物表回答“最终产出和中间文件在哪里”。2.2 用状态机管理节点生命周期节点不能只有“成功”和“失败”两种状态否则无法表达“正在执行”“等待重试”“被暂停”等中间状态。建议的状态集合如下状态含义可能流向pending节点已创建等待执行runningrunning子代理正在执行succeeded、failedsucceeded节点执行成功无failed节点执行失败pending、runningpaused节点因外部原因暂停pending、runningcanceled节点被取消无状态机本身不需要复杂框架。只需要保证两条规则更新节点状态时同时写入事件状态转换只能从合法来源跳到目标状态。例如一个已经 succeeded 的节点不能再次进入 running除非创建了新的重跑节点。2.3 事件日志采用 append-only 追加模式事件日志是追踪的基础。它的核心原则是“只追加不修改”。一旦写入node_succeeded这条记录就永久保留即使后续因为数据修正要重新执行也不能删除原事件而是追加一条node_retried事件。事件表建议包含这些字段字段说明id自增主键保证事件顺序workflow_id所属工作流node_id产生事件的节点trace_id贯穿整个工作流的追踪 IDevent_type事件类型如 node_started、node_succeededactor事件发起者如 runner、subagent、schedulerpayload事件详细内容JSON 格式created_at时间戳为什么把 trace_id 单独拆出来因为一个工作流可能分裂成多个子流程又可能通过消息队列由不同进程执行。trace_id 可以在不依赖数据库自增序列的情况下把散落在不同服务里的日志串成一条链路。注意事件表不要保存大对象。模型返回的完整 JSON 可以存入 artifact 表事件 payload 只保存关联 ID 和摘要否则事件表会膨胀到难以查询。2.4 通过父子关系形成追踪树subagent 工作流天然是树形结构主代理是根节点拆出的子任务是子节点子任务还可以继续拆分。因此在agent_node表中必须保存parent_id。有了父子关系就可以根据任意一个节点向上找到根节点向下找到所有子节点。如果配合 trace_id还能把“同一节点在不同尝试中的事件”按时间拼接起来。追踪树是展示工作流执行过程的基础。查询时先按workflow_id取回所有节点再用parent_id在内存中组装树结构比在 SQL 里做递归查询更直观。3. 实现一个最小持久化 subagent 执行器3.1 环境准备与项目结构下面示例使用 Python 3.10 和标准库 sqlite3不依赖额外框架。这样做是为了让读者看清持久化和追踪的最小逻辑不被 LangGraph 等框架的封装干扰。实际项目可以替换成 PostgreSQL 和任意 Agent 框架。项目结构建议如下subagent-persist/ ├── db.py ├── runner.py ├── main.py ├── trace.py └── workflows/ └── travel.jsondb.py负责建表和数据库连接runner.py是核心执行器main.py是演示入口trace.py负责把数据库记录组装成可读的追踪树。3.2 建表语句与数据库访问层先实现db.py。这里把四个核心实体建成四张表并开启 WAL 模式避免读写互相阻塞。import sqlite3 from contextlib import contextmanager SCHEMA CREATE TABLE IF NOT EXISTS workflow_run ( id TEXT PRIMARY KEY, name TEXT NOT NULL, status TEXT NOT NULL, payload TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS agent_node ( id TEXT PRIMARY KEY, workflow_id TEXT NOT NULL, parent_id TEXT, agent_name TEXT NOT NULL, status TEXT NOT NULL, input TEXT, output TEXT, error TEXT, attempts INTEGER DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (workflow_id) REFERENCES workflow_run(id) ); CREATE TABLE IF NOT EXISTS agent_event ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, node_id TEXT, trace_id TEXT NOT NULL, event_type TEXT NOT NULL, actor TEXT NOT NULL, payload TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS artifact ( id TEXT PRIMARY KEY, node_id TEXT NOT NULL, name TEXT NOT NULL, content_type TEXT NOT NULL, uri TEXT NOT NULL, created_at TEXT NOT NULL ); contextmanager def get_conn(db_path): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA foreign_keysON) try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(db_path): with get_conn(db_path) as conn: conn.executescript(SCHEMA)建表时把agent_node.workflow_id和agent_event.workflow_id都建为普通字段。若想保证数据一致性可以加外键约束但要注意 SQLite 默认未开启外键需要在连接时执行PRAGMA foreign_keysON。3.3 执行器核心创建节点、更新状态、写入事件接下来实现runner.py。这里的关键是不把业务逻辑写死在子代理调用里而是通过dispatch_fn注入真正的执行函数。这样在演示时可以用 mock 函数模拟成功和失败在生产环境替换成真实 LLM 调用即可。import json import uuid from datetime import datetime, timezone from db import get_conn, init_db def now_iso(): return datetime.now(timezone.utc).isoformat() class WorkflowRunner: def __init__(self, db_path, dispatch_fnNone): self.db_path db_path self.dispatch_fn dispatch_fn or self._default_dispatch init_db(db_path) def _default_dispatch(self, agent_name, input_data): # 演示用默认实现生产环境替换为实际 subagent 调用 if agent_name flaky: raise RuntimeError(simulated failure) return {message: f{agent_name} processed ok} def _emit(self, conn, workflow_id, node_id, trace_id, event_type, actor, payloadNone): conn.execute( INSERT INTO agent_event (workflow_id, node_id, trace_id, event_type, actor, payload, created_at) VALUES (?, ?, ?, ?, ?, ?, ?), ( workflow_id, node_id, trace_id, event_type, actor, json.dumps(payload, ensure_asciiFalse), now_iso(), ), ) def create_run(self, name, payload): run_id uuid.uuid4().hex with get_conn(self.db_path) as conn: conn.execute( INSERT INTO workflow_run (id, name, status, payload, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?), (run_id, name, pending, json.dumps(payload), now_iso(), now_iso()), ) return run_id def add_node(self, workflow_id, parent_id, agent_name, input_data): node_id uuid.uuid4().hex trace_id uuid.uuid4().hex with get_conn(self.db_path) as conn: conn.execute( INSERT INTO agent_node (id, workflow_id, parent_id, agent_name, status, input, attempts, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?), ( node_id, workflow_id, parent_id, agent_name, pending, json.dumps(input_data), 0, now_iso(), now_iso(), ), ) self._emit( conn, workflow_id, node_id, trace_id, node_created, runner, {agent_name: agent_name}, ) return node_id def run_node(self, node_id, max_retries3): with get_conn(self.db_path) as conn: row conn.execute( SELECT * FROM agent_node WHERE id ?, (node_id,) ).fetchone() if row is None: raise ValueError(fnode not found: {node_id}) node dict(row) workflow_id node[workflow_id] trace_id uuid.uuid4().hex next_attempt node[attempts] 1 if next_attempt max_retries: self._emit( conn, workflow_id, node_id, trace_id, node_gave_up, runner, {attempts: next_attempt - 1, max_retries: max_retries}, ) return failed self._transition_node( conn, node_id, running, attemptsnext_attempt, errorNone, ) self._emit( conn, workflow_id, node_id, trace_id, node_started, runner, {attempt: next_attempt}, ) input_data json.loads(node[input]) try: output self.dispatch_fn( agent_namenode[agent_name], input_datainput_data, ) output_text json.dumps(output, ensure_asciiFalse) self._save_artifact(conn, node_id, result.json, application/json, output_text) self._transition_node(conn, node_id, succeeded, outputoutput_text) self._emit( conn, workflow_id, node_id, trace_id, node_succeeded, runner, {artifact: result.json}, ) return succeeded except Exception as exc: self._transition_node(conn, node_id, failed, errorstr(exc)) self._emit( conn, workflow_id, node_id, trace_id, node_failed, runner, {error: str(exc)}, ) return failed def _transition_node(self, conn, node_id, status, attemptsNone, outputNone, errorNone): set_parts [status ?, updated_at ?] params [status, now_iso()] if attempts is not None: set_parts.append(attempts ?) params.append(attempts) if output is not None: set_parts.append(output ?) params.append(output) if error is not None: set_parts.append(error ?) params.append(error) params.append(node_id) conn.execute( fUPDATE agent_node SET {, .join(set_parts)} WHERE id ?, params, ) def _save_artifact(self, conn, node_id, name, content_type, uri): artifact_id uuid.uuid4().hex conn.execute( INSERT INTO artifact (id, node_id, name, content_type, uri, created_at) VALUES (?, ?, ?, ?, ?, ?), (artifact_id, node_id, name, content_type, uri, now_iso()), ) return artifact_id这段代码体现了持久化和可追踪的最小实现。每次节点状态变化都先更新agent_node再写入agent_event。这个顺序很重要先有当前状态再有状态变化痕迹排查时才能看到完整链路。attempts记录重试次数max_retries作为参数传入run_node。3.4 设计一个可恢复的执行流程上面的run_node已经可以把失败的节点标记为 failed。但要实现“断点续跑”还需要一个方法在工作流重启后找到所有未成功节点并继续执行。恢复策略可以根据业务不同而不同。这里采用最简单的一种从根节点开始遍历所有节点对于已成功的节点直接跳过对于未成功的节点调用run_node如果某个节点的子节点还未执行就递归执行子节点。def resume_run(self, workflow_id, max_retries3): with get_conn(self.db_path) as conn: root conn.execute( SELECT * FROM agent_node WHERE workflow_id ? AND parent_id IS NULL, (workflow_id,), ).fetchone() if root is None: raise ValueError(root node not found) root_id root[id] self._run_subtree(root_id, max_retries) def _run_subtree(self, node_id, max_retries): with get_conn(self.db_path) as conn: row conn.execute( SELECT * FROM agent_node WHERE id ?, (node_id,) ).fetchone() node dict(row) children conn.execute( SELECT * FROM agent_node WHERE parent_id ? ORDER BY created_at, (node_id,), ).fetchall() if node[status] succeeded: # 叶子节点已成功只需要处理子节点 for child in children: self._run_subtree(child[id], max_retries) return # 当前节点未完成先执行当前节点 self.run_node(node_id, max_retries) # 执行成功后再执行子节点 with get_conn(self.db_path) as conn: row conn.execute( SELECT * FROM agent_node WHERE id ?, (node_id,) ).fetchone() node dict(row) children conn.execute( SELECT * FROM agent_node WHERE parent_id ? ORDER BY created_at, (node_id,), ).fetchall() if node[status] succeeded: for child in children: self._run_subtree(child[id], max_retries)恢复逻辑的核心是“状态检查优先于执行”。节点已经成功就不要再执行即使它的输出在数据库中仍在。节点失败就重新执行但要控制重试次数避免无限循环。3.5 编写查询与追踪树 API持久化做完后还有一个关键问题是“怎么查”。需要提供两个查询按运行 ID 查出所有事件按自增 ID 排序。按运行 ID 查出所有节点组装成树结构。# trace.py import json import sqlite3 def query_events(db_path, workflow_id): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT * FROM agent_event WHERE workflow_id ? ORDER BY id ASC, (workflow_id,), ).fetchall() conn.close() return [dict(row) for row in rows] def build_trace_tree(db_path, workflow_id): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT * FROM agent_node WHERE workflow_id ? ORDER BY created_at ASC, (workflow_id,), ).fetchall() conn.close() nodes {row[id]: dict(row) for row in rows} trees [] for node in nodes.values(): parent_id node[parent_id] if parent_id is None: trees.append(node) else: nodes[parent_id].setdefault(children, []).append(node) return trees在main.py里可以这样用# main.py from runner import WorkflowRunner from trace import query_events, build_trace_tree def real_dispatch(agent_name, input_data): # 在测试环境替换成调用真实 subagent并返回标准 JSON print(fdispatch subagent: {agent_name}, input{input_data}) return {status: ok, data: input_data} runner WorkflowRunner(demo.db, dispatch_fnreal_dispatch) run_id runner.create_run(travel, {city: Hangzhou}) book runner.add_node(run_id, None, booking, {type: hotel}) weather runner.add_node(run_id, book, weather, {date: 2025-01-01}) # 模拟一次成功执行 print(runner.run_node(book)) # 模拟天气子代理失败一次后恢复 try: runner.run_node(weather) except Exception as e: print(failed:, e) runner.run_node(weather) # 打印事件和节点树 print(query_events(demo.db, run_id)) print(build_trace_tree(demo.db, run_id))运行示例会输出事件列表和节点树。可以看到每个节点都有独立的node_id事件按时间顺序排列内容中包含 agent_name、输入输出摘要和错误信息。4. 参数、错误处理与验证方法4.1 关键参数说明执行器中的几个参数对运行行为影响很大实际项目里必须明确给出配置。参数默认值 / 建议值调小的影响调大的影响推荐场景max_retries3失败后恢复能力弱可能瞬时错误直接放弃重试时间过长可能拖慢整个工作流网络不稳定场景适当调大计算类任务根据估算总时长设置timeout按子代理类型设置慢节点容易超时失败节点长时间占用资源难以发现异常LLM 调用可设 30-60 秒复杂计算任务单独设置max_parallelism1串行执行速度慢并发高时可能打满 API 配额依赖外部 API 时控制在 3-5内部计算节点可调高resume_from_event0每次从根节点重跑成本高跳过事件容易丢失必要上下文推荐使用“节点级状态恢复”不要依赖事件 ID 恢复max_retries在示例中用于控制每个节点的总尝试次数。要注意这里统计的是“当前节点总尝试次数”。如果业务要求“每个输入最多重试 3 次”不是每个节点初始化后清零而是把 attempts 累加上去。4.2 构造一个验证场景为了验证持久化和可追踪是否真正生效需要构造一个包含“失败后恢复”的测试场景。推荐的验证步骤如下创建一个工作流包含一个根节点和一个依赖根节点的子节点。把dispatch_fn写成第一次调用指定 agent 时抛异常第二次调用成功。执行根节点成功后执行子节点第一次失败。再次调用run_node第二次成功。查询agent_event确认出现了node_failed和node_succeeded两条事件。查询agent_node确认 attempts 变为 2status 为 succeeded。这个验证覆盖了三个核心能力状态更新、事件记录、重试恢复。只验证“程序能启动”没有意义还要验证状态从 failed 恢复到 succeeded 后数据库里的记录仍然一致。4.3 预期输出与验证结果当执行成功后查询agent_node表会看到类似下面的记录id: abc123 workflow_id: run_111 parent_id: null agent_name: booking status: succeeded input: {type: hotel} output: {status: ok, data: {type: hotel}} error: null attempts: 1查询agent_event时会看到两条事件event_type: node_started actor: runner payload: {attempt: 1} event_type: node_succeeded actor: runner payload: {artifact: result.json}如果子代理失败过一次再成功事件表会多一条node_failed并且节点表的 attempts 值会是 2。通过事件表的顺序可以完整还原执行过程。4.4 学习环境与生产环境的差异SQLite 版本适合学习和小型工具但生产环境必须做替换或增强。主要差异集中在以下几点维度学习环境生产环境数据库SQLite 单文件PostgreSQL 或 MySQL支持行级锁和事务隔离并发控制单进程WAL 缓解读写冲突多 worker 同时更新节点状态必须用SELECT ... FOR UPDATE持久化范围状态、事件、产物 URI增加调度信息、队列信息、执行环境元数据追踪链路单库查询通过 trace_id 对接 OpenTelemetry 等分布式追踪系统恢复策略当前节点失败则重跑当前节点需要处理多节点并发失败、等待子任务完成、分布式锁安全问题本地示例事件内容可能包含敏感数据需要脱敏、权限控制、审计日志保留策略不要直接把 SQLite 建表语句搬到 PostgreSQL 后就不管了。sqlite3 的自动类型约束很弱字段存 JSON 时也不会校验PostgreSQL 建议使用 JSONB 字段类型并对 workflow_id、status、parent_id 建联合索引。注意生产环境应该为agent_event表和artifact表设置数据保留策略。追踪数据是很有价值但无限保留会带来成本和合规风险。建议按天数归档或压缩。5. 常见问题与排查链路5.1 状态不一致节点显示 running但进程已经重启现象数据库里某个节点状态是 running但对应进程已经退出事件表里只有 node_started没有 node_succeeded 或 node_failed。可能原因进程被强杀异常处理没有机会写入失败事件。网络超时后子代理实际已成功但进程没有收到返回就崩溃了。检查方式查询事件表确认该节点 latest event ID。查看 artifact 表确认是否有 result.json。对照进程日志看是否在 node_started 之后发生了崩溃。处理建议启动时执行一轮“孤儿节点修复”把超过 timeout 且状态为 running 的节点标记为 failed。如果 artifact 已存在可以将节点标记为 succeeded避免重复执行。5.2 恢复后重复执行子代理现象使用resume_run后某些已经成功执行过的节点又被重新调用了一次。可能原因恢复逻辑判断的不是节点状态而是事件是否存在。代码在遍历子节点时误把“有 node_started 事件”当作“待恢复”导致已经成功但事件不全的节点被重跑。检查方式检查该节点的 status 是否为 succeeded。检查事件表确认 node_succeeded 是否存在。如果 status 是 succeeded但 node_succeeded 事件缺失说明事件写入逻辑有遗漏。处理建议恢复判断以agent_node.status为准而不是以事件为准。写一个幂等检查如果一个节点已经有成功产物就不要再次执行。5.3 事件缺失或乱序现象节点已经 succeeded但事件列表里没有对应的 node_succeeded。可能原因在写入事件前抛了异常导致状态更新成功但事件未写入。多线程并发写事件时自增 ID 顺序和业务顺序不一致。检查方式查看状态更新和事件写入是否在同一个事务里。检查是否使用了多连接并发写入事件表的时间戳是否可信。处理建议状态更新和事件写入必须放在同一个连接、同一个事务里。事件表增加sequence_no或使用created_at排序时要注意多进程时钟不一致问题推荐依赖自增 ID 加业务时间戳双重排序。5.4 并发更新导致 attempts 被覆盖现象两个 worker 同时处理同一个节点最终 attempts 值小于真实执行次数。可能原因代码先读 attempts再加一再写回但两个请求同时读到同一个旧值。SQLite 默认串行写但 PostgreSQL 在SELECT后UPDATE时如果没有行锁也会出现覆盖。检查方式查询节点更新历史看是否有两次 node_started 事件但 attempts 只增加了 1。处理建议更新语句改为原子操作UPDATE agent_node SET attempts attempts 1 WHERE id ?。生产环境使用SELECT ... FOR UPDATE锁定节点行再执行后续操作。5.5 排查清单遇到 subagent 工作流异常时按以下顺序排查查询workflow_run确认工作流整体状态。查询agent_node找出所有 status 不为 succeeded 的节点。查询agent_event按节点 id 过滤按自增 id 排序确认最后一个事件。查看artifact确认是否有产物落盘。检查节点输入与输出是否匹配排除模型返回异常格式的情况。查看节点错误字段确认是否真的执行失败还是仅追踪失败。再检查外围信息API 配额、网络超时、依赖服务状态。6. 最佳实践与扩展方向6.1 持久化工作流时的可复用清单下面的清单可以用于设计自己的 subagent 工作流引擎每次运行必须有唯一 run_id每个节点必须有唯一 node_id。每个节点必须记录 parent_id形成可追踪的树结构。每个节点必须有状态字段推荐使用 pending、running、succeeded、failed、paused、canceled。每次状态变化必须追加一条事件事件 actor 要明确。状态更新和事件写入要放在同一事务中。输入输出建议存 JSON 快照至少保存必要上下文。大文件不要直接进事件表存入 artifact事件里只留 URI。恢复逻辑以节点状态为准不能依赖事件存在性。每个节点执行前生成幂等键特别是有外部副作用的子代理。对失败的节点保留原始错误方便后续重试时做不同处理。6.2 用幂等键避免重复执行副作用如果 subagent 不只是“调用模型返回文本”还会调用外部 API、写数据库、发消息那么幂等性比状态恢复更重要。一个简单的做法是在输入数据中加入request_id并在目标系统里按 request_id 去重。input_data { request_id: uuid.uuid5(uuid.NAMESPACE_URL, f{node_id}-{attempt}).hex, payload: original_input, }重试时重新使用同一 request_id外部系统就可以识别“这次调用的任务和上次一样”从而避免重复下单、重复发消息等副作用。6.3 从单机示例扩展到分布式追踪本地示例把事件写入 SQLite生产环境通常需要把追踪数据接入 OpenTelemetry。思路是保留 trace_id在事件写入数据库的同时额外发送一个 span 到追踪后端。这样既能通过数据库查询恢复状态又能通过追踪系统查看调用链、耗时和资源消耗。具体可以分为两层业务追踪层记录 subagent 的输入、输出、状态流转保存在业务库。性能追踪层记录每次 subagent 调用的耗时、token 用量、模型名称、重试次数发送到时序数据库或链路追踪平台。即使正在使用 LangGraph、AutoGen 等框架也可以在工具调用装饰器里接入这层追踪不必重写框架本身。6.4 可视化与血缘是下一步重点状态表和事件表是可追踪的基础但直接看 SQL 记录不够直观。建议在追踪树之上提供两类视图执行时间线横向展示每个节点从创建到成功或失败的完整事件流。血缘依赖图展示父节点和子节点的依赖关系以及产物文件之间的关系。这两类视图可以从agent_node的 parent_id 和artifact表推导出来。展示层可以把节点状态映射成颜色绿色 succeeded、红色 failed、黄色 running、灰色 pending。把这两类视图做出来subagent 工作流就不再是一堆无法复现的临时调用而是一个具备可观测性和可维护性的业务系统。最终要记住一个核心判断subagent 工作流的价值不只在于“让模型把任务拆给更小的模型”更在于“把任务拆解、执行、失败、恢复的整个过程变成可以追溯的数据资产”。持久化和可追踪不是附加功能而是让 subagent 模式进入生产环境的必要条件。下一步建议从本文的 SQLite 示例出发选择一个你正在使用的 Agent 框架把状态表、事件表和恢复逻辑先加进去再逐步补充可视化、幂等和分布式追踪。 subagent工作流持久化与追踪：状态机与事件日志的工程实践
在实际使用 LLM Agent 的项目里subagent 工作流正在成为最常见的任务拆分方式。主代理把“查资料、写代码、算指标、生成报告”这类小任务分别交给子代理执行自己只负责编排和汇总。子代理执行结束后结果返回主代理整个交互随之结束。这种模式很自然但它有一个容易被忽略的短板默认情况下subagent 工作流是瞬时的既不会跨进程保留也不会留下可用于审计和排查的轨迹。标题里的 persistent 和 trackable指的就是把“调用完就消失”的子代理执行过程改造成一个可恢复、可追踪、可复现的工程单元。文章会围绕一个最小可运行的执行器展开。这个执行器接收一个主代理定义的工作流将节点逐一派发给 subagent并把每次执行的状态、事件、输入输出和产物写入 SQLite。之后我会说明为什么用状态机加 append-only 事件日志来建模而不是直接用一张表保存最终结果然后给出完整代码、建表语句、恢复逻辑和查询方式最后补充常见故障、排查链路和生产环境需要补强的地方。读完以后你可以把同一套设计迁移到 LangGraph、AutoGen、CrewAI 或者其他自定义 Agent 框架中。1. 先理解 subagent 工作流为何需要持久化1.1 subagent 工作流到底是什么在 Agent 系统里subagent 是一个独立的执行单元。它接收任务描述、必要上下文和可用工具然后调用大模型推理、执行工具并返回结果。与直接在同一个 prompt 里让模型完成多个步骤不同subagent 方式强调隔离和复用主代理不需要关心子代理内部提示词怎么组织只需要关心子任务是否成功、返回格式是否符合约定。一个典型示例如下主代理负责“规划一次三天旅行”。主代理拆分出“查机票”“查酒店”“查天气”三个子任务。三个子代理分别执行并把结果返回主代理。主代理汇总结果生成最终旅行计划。这个过程如果只是内存里的函数调用运行结束后所有中间状态都会被回收。一旦某个子代理执行到一半出现网络超时、模型 API 报错或者服务进程被重新部署整个工作流就只能从头再跑。1.2 瞬时执行带来的四个问题只依赖内存的 subagent 工作流会带来四个实际工程问题第一进程崩溃后上下文丢失。子代理已经完成的结果没有落盘主代理已经推进到的步骤也没有记录重启后只能重新开始。第二无法做失败重试。重试的前提是知道“失败发生之前完成了什么”。如果没有任何状态记录重试只能退化为整体重跑成本非常高。第三无法审计和解释结果。当领导或客户问“这个结论是哪几个子代理、基于什么输入算出来的”如果你只能打开终端看到一段最新日志很难回答清楚。第四难以并发和调度。多子代理并行执行时调度器需要知道每个节点的状态才能决定哪些节点可以继续、哪些节点已经完成、哪些节点需要重新执行。瞬时执行没有这种信息。1.3 持久化与可追踪的分工持久化解决的是“状态能不能活下来”的问题。它要求把工作流运行进度、节点状态、输入输出写入数据库这样进程重启后可以从最近一个稳定点继续执行。可追踪解决的是“发生了什么”的问题。它要求记录事件序列什么时候创建了任务、什么时候交给子代理、子代理返回了什么、失败原因是什么、重试了几次。事件日志本身也是持久化的但它的设计目的不是恢复执行而是复现执行过程。在这套设计中持久化面向恢复可追踪面向审计与调试。两者结合起来才是完整的工程闭环。1.4 长时间任务更依赖稳定落盘有一类 subagent 任务格外需要持久化计算型任务。例如某个子代理负责计算 cubical persistent homology它需要处理拓扑特征、构造 cubical complex、计算不同参数下的同调群。这类任务可能运行几十分钟产生多个中间结果文件如果中途断掉重复计算的时间成本很难接受。这类场景与常见的“调用一次模型拿一个 JSON”完全不同。LLM 调用通常以秒为单位而计算型子任务可能以分钟甚至小时为单位。对工作流引擎来说任务越重持久化和可追踪的价值越大。状态恢复不是“加分项”而是“继续运行”的基本条件。2. 设计 subagent 工作流的数据模型2.1 核心实体运行、节点、事件、产物要让 subagent 工作流可持久化、可追踪数据库不能只记录“完成了哪几步”。建议拆成四个核心实体workflow_run一次完整的工作流执行包含名称、状态、入口 payload。agent_node工作流里的一个 subagent 调用节点记录输入输出和错误。agent_event每一条执行事件记录“谁在什么时间做了什么”。artifact节点运行过程中产生的结果文件或结构化数据。以“旅行规划”为例一次 workflow_run 对应整个规划过程。查天气、查酒店分别是两个 agent_node。子代理开始执行、成功返回、结果落盘这些行为都对应 agent_event。最终返回的 JSON 包可以存入 artifact。这样建模的好处是职责分离。节点表回答“当前执行到哪一步”事件表回答“每一步都发生了什么”产物表回答“最终产出和中间文件在哪里”。2.2 用状态机管理节点生命周期节点不能只有“成功”和“失败”两种状态否则无法表达“正在执行”“等待重试”“被暂停”等中间状态。建议的状态集合如下状态含义可能流向pending节点已创建等待执行runningrunning子代理正在执行succeeded、failedsucceeded节点执行成功无failed节点执行失败pending、runningpaused节点因外部原因暂停pending、runningcanceled节点被取消无状态机本身不需要复杂框架。只需要保证两条规则更新节点状态时同时写入事件状态转换只能从合法来源跳到目标状态。例如一个已经 succeeded 的节点不能再次进入 running除非创建了新的重跑节点。2.3 事件日志采用 append-only 追加模式事件日志是追踪的基础。它的核心原则是“只追加不修改”。一旦写入node_succeeded这条记录就永久保留即使后续因为数据修正要重新执行也不能删除原事件而是追加一条node_retried事件。事件表建议包含这些字段字段说明id自增主键保证事件顺序workflow_id所属工作流node_id产生事件的节点trace_id贯穿整个工作流的追踪 IDevent_type事件类型如 node_started、node_succeededactor事件发起者如 runner、subagent、schedulerpayload事件详细内容JSON 格式created_at时间戳为什么把 trace_id 单独拆出来因为一个工作流可能分裂成多个子流程又可能通过消息队列由不同进程执行。trace_id 可以在不依赖数据库自增序列的情况下把散落在不同服务里的日志串成一条链路。注意事件表不要保存大对象。模型返回的完整 JSON 可以存入 artifact 表事件 payload 只保存关联 ID 和摘要否则事件表会膨胀到难以查询。2.4 通过父子关系形成追踪树subagent 工作流天然是树形结构主代理是根节点拆出的子任务是子节点子任务还可以继续拆分。因此在agent_node表中必须保存parent_id。有了父子关系就可以根据任意一个节点向上找到根节点向下找到所有子节点。如果配合 trace_id还能把“同一节点在不同尝试中的事件”按时间拼接起来。追踪树是展示工作流执行过程的基础。查询时先按workflow_id取回所有节点再用parent_id在内存中组装树结构比在 SQL 里做递归查询更直观。3. 实现一个最小持久化 subagent 执行器3.1 环境准备与项目结构下面示例使用 Python 3.10 和标准库 sqlite3不依赖额外框架。这样做是为了让读者看清持久化和追踪的最小逻辑不被 LangGraph 等框架的封装干扰。实际项目可以替换成 PostgreSQL 和任意 Agent 框架。项目结构建议如下subagent-persist/ ├── db.py ├── runner.py ├── main.py ├── trace.py └── workflows/ └── travel.jsondb.py负责建表和数据库连接runner.py是核心执行器main.py是演示入口trace.py负责把数据库记录组装成可读的追踪树。3.2 建表语句与数据库访问层先实现db.py。这里把四个核心实体建成四张表并开启 WAL 模式避免读写互相阻塞。import sqlite3 from contextlib import contextmanager SCHEMA CREATE TABLE IF NOT EXISTS workflow_run ( id TEXT PRIMARY KEY, name TEXT NOT NULL, status TEXT NOT NULL, payload TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS agent_node ( id TEXT PRIMARY KEY, workflow_id TEXT NOT NULL, parent_id TEXT, agent_name TEXT NOT NULL, status TEXT NOT NULL, input TEXT, output TEXT, error TEXT, attempts INTEGER DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (workflow_id) REFERENCES workflow_run(id) ); CREATE TABLE IF NOT EXISTS agent_event ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, node_id TEXT, trace_id TEXT NOT NULL, event_type TEXT NOT NULL, actor TEXT NOT NULL, payload TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS artifact ( id TEXT PRIMARY KEY, node_id TEXT NOT NULL, name TEXT NOT NULL, content_type TEXT NOT NULL, uri TEXT NOT NULL, created_at TEXT NOT NULL ); contextmanager def get_conn(db_path): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA foreign_keysON) try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(db_path): with get_conn(db_path) as conn: conn.executescript(SCHEMA)建表时把agent_node.workflow_id和agent_event.workflow_id都建为普通字段。若想保证数据一致性可以加外键约束但要注意 SQLite 默认未开启外键需要在连接时执行PRAGMA foreign_keysON。3.3 执行器核心创建节点、更新状态、写入事件接下来实现runner.py。这里的关键是不把业务逻辑写死在子代理调用里而是通过dispatch_fn注入真正的执行函数。这样在演示时可以用 mock 函数模拟成功和失败在生产环境替换成真实 LLM 调用即可。import json import uuid from datetime import datetime, timezone from db import get_conn, init_db def now_iso(): return datetime.now(timezone.utc).isoformat() class WorkflowRunner: def __init__(self, db_path, dispatch_fnNone): self.db_path db_path self.dispatch_fn dispatch_fn or self._default_dispatch init_db(db_path) def _default_dispatch(self, agent_name, input_data): # 演示用默认实现生产环境替换为实际 subagent 调用 if agent_name flaky: raise RuntimeError(simulated failure) return {message: f{agent_name} processed ok} def _emit(self, conn, workflow_id, node_id, trace_id, event_type, actor, payloadNone): conn.execute( INSERT INTO agent_event (workflow_id, node_id, trace_id, event_type, actor, payload, created_at) VALUES (?, ?, ?, ?, ?, ?, ?), ( workflow_id, node_id, trace_id, event_type, actor, json.dumps(payload, ensure_asciiFalse), now_iso(), ), ) def create_run(self, name, payload): run_id uuid.uuid4().hex with get_conn(self.db_path) as conn: conn.execute( INSERT INTO workflow_run (id, name, status, payload, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?), (run_id, name, pending, json.dumps(payload), now_iso(), now_iso()), ) return run_id def add_node(self, workflow_id, parent_id, agent_name, input_data): node_id uuid.uuid4().hex trace_id uuid.uuid4().hex with get_conn(self.db_path) as conn: conn.execute( INSERT INTO agent_node (id, workflow_id, parent_id, agent_name, status, input, attempts, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?), ( node_id, workflow_id, parent_id, agent_name, pending, json.dumps(input_data), 0, now_iso(), now_iso(), ), ) self._emit( conn, workflow_id, node_id, trace_id, node_created, runner, {agent_name: agent_name}, ) return node_id def run_node(self, node_id, max_retries3): with get_conn(self.db_path) as conn: row conn.execute( SELECT * FROM agent_node WHERE id ?, (node_id,) ).fetchone() if row is None: raise ValueError(fnode not found: {node_id}) node dict(row) workflow_id node[workflow_id] trace_id uuid.uuid4().hex next_attempt node[attempts] 1 if next_attempt max_retries: self._emit( conn, workflow_id, node_id, trace_id, node_gave_up, runner, {attempts: next_attempt - 1, max_retries: max_retries}, ) return failed self._transition_node( conn, node_id, running, attemptsnext_attempt, errorNone, ) self._emit( conn, workflow_id, node_id, trace_id, node_started, runner, {attempt: next_attempt}, ) input_data json.loads(node[input]) try: output self.dispatch_fn( agent_namenode[agent_name], input_datainput_data, ) output_text json.dumps(output, ensure_asciiFalse) self._save_artifact(conn, node_id, result.json, application/json, output_text) self._transition_node(conn, node_id, succeeded, outputoutput_text) self._emit( conn, workflow_id, node_id, trace_id, node_succeeded, runner, {artifact: result.json}, ) return succeeded except Exception as exc: self._transition_node(conn, node_id, failed, errorstr(exc)) self._emit( conn, workflow_id, node_id, trace_id, node_failed, runner, {error: str(exc)}, ) return failed def _transition_node(self, conn, node_id, status, attemptsNone, outputNone, errorNone): set_parts [status ?, updated_at ?] params [status, now_iso()] if attempts is not None: set_parts.append(attempts ?) params.append(attempts) if output is not None: set_parts.append(output ?) params.append(output) if error is not None: set_parts.append(error ?) params.append(error) params.append(node_id) conn.execute( fUPDATE agent_node SET {, .join(set_parts)} WHERE id ?, params, ) def _save_artifact(self, conn, node_id, name, content_type, uri): artifact_id uuid.uuid4().hex conn.execute( INSERT INTO artifact (id, node_id, name, content_type, uri, created_at) VALUES (?, ?, ?, ?, ?, ?), (artifact_id, node_id, name, content_type, uri, now_iso()), ) return artifact_id这段代码体现了持久化和可追踪的最小实现。每次节点状态变化都先更新agent_node再写入agent_event。这个顺序很重要先有当前状态再有状态变化痕迹排查时才能看到完整链路。attempts记录重试次数max_retries作为参数传入run_node。3.4 设计一个可恢复的执行流程上面的run_node已经可以把失败的节点标记为 failed。但要实现“断点续跑”还需要一个方法在工作流重启后找到所有未成功节点并继续执行。恢复策略可以根据业务不同而不同。这里采用最简单的一种从根节点开始遍历所有节点对于已成功的节点直接跳过对于未成功的节点调用run_node如果某个节点的子节点还未执行就递归执行子节点。def resume_run(self, workflow_id, max_retries3): with get_conn(self.db_path) as conn: root conn.execute( SELECT * FROM agent_node WHERE workflow_id ? AND parent_id IS NULL, (workflow_id,), ).fetchone() if root is None: raise ValueError(root node not found) root_id root[id] self._run_subtree(root_id, max_retries) def _run_subtree(self, node_id, max_retries): with get_conn(self.db_path) as conn: row conn.execute( SELECT * FROM agent_node WHERE id ?, (node_id,) ).fetchone() node dict(row) children conn.execute( SELECT * FROM agent_node WHERE parent_id ? ORDER BY created_at, (node_id,), ).fetchall() if node[status] succeeded: # 叶子节点已成功只需要处理子节点 for child in children: self._run_subtree(child[id], max_retries) return # 当前节点未完成先执行当前节点 self.run_node(node_id, max_retries) # 执行成功后再执行子节点 with get_conn(self.db_path) as conn: row conn.execute( SELECT * FROM agent_node WHERE id ?, (node_id,) ).fetchone() node dict(row) children conn.execute( SELECT * FROM agent_node WHERE parent_id ? ORDER BY created_at, (node_id,), ).fetchall() if node[status] succeeded: for child in children: self._run_subtree(child[id], max_retries)恢复逻辑的核心是“状态检查优先于执行”。节点已经成功就不要再执行即使它的输出在数据库中仍在。节点失败就重新执行但要控制重试次数避免无限循环。3.5 编写查询与追踪树 API持久化做完后还有一个关键问题是“怎么查”。需要提供两个查询按运行 ID 查出所有事件按自增 ID 排序。按运行 ID 查出所有节点组装成树结构。# trace.py import json import sqlite3 def query_events(db_path, workflow_id): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT * FROM agent_event WHERE workflow_id ? ORDER BY id ASC, (workflow_id,), ).fetchall() conn.close() return [dict(row) for row in rows] def build_trace_tree(db_path, workflow_id): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT * FROM agent_node WHERE workflow_id ? ORDER BY created_at ASC, (workflow_id,), ).fetchall() conn.close() nodes {row[id]: dict(row) for row in rows} trees [] for node in nodes.values(): parent_id node[parent_id] if parent_id is None: trees.append(node) else: nodes[parent_id].setdefault(children, []).append(node) return trees在main.py里可以这样用# main.py from runner import WorkflowRunner from trace import query_events, build_trace_tree def real_dispatch(agent_name, input_data): # 在测试环境替换成调用真实 subagent并返回标准 JSON print(fdispatch subagent: {agent_name}, input{input_data}) return {status: ok, data: input_data} runner WorkflowRunner(demo.db, dispatch_fnreal_dispatch) run_id runner.create_run(travel, {city: Hangzhou}) book runner.add_node(run_id, None, booking, {type: hotel}) weather runner.add_node(run_id, book, weather, {date: 2025-01-01}) # 模拟一次成功执行 print(runner.run_node(book)) # 模拟天气子代理失败一次后恢复 try: runner.run_node(weather) except Exception as e: print(failed:, e) runner.run_node(weather) # 打印事件和节点树 print(query_events(demo.db, run_id)) print(build_trace_tree(demo.db, run_id))运行示例会输出事件列表和节点树。可以看到每个节点都有独立的node_id事件按时间顺序排列内容中包含 agent_name、输入输出摘要和错误信息。4. 参数、错误处理与验证方法4.1 关键参数说明执行器中的几个参数对运行行为影响很大实际项目里必须明确给出配置。参数默认值 / 建议值调小的影响调大的影响推荐场景max_retries3失败后恢复能力弱可能瞬时错误直接放弃重试时间过长可能拖慢整个工作流网络不稳定场景适当调大计算类任务根据估算总时长设置timeout按子代理类型设置慢节点容易超时失败节点长时间占用资源难以发现异常LLM 调用可设 30-60 秒复杂计算任务单独设置max_parallelism1串行执行速度慢并发高时可能打满 API 配额依赖外部 API 时控制在 3-5内部计算节点可调高resume_from_event0每次从根节点重跑成本高跳过事件容易丢失必要上下文推荐使用“节点级状态恢复”不要依赖事件 ID 恢复max_retries在示例中用于控制每个节点的总尝试次数。要注意这里统计的是“当前节点总尝试次数”。如果业务要求“每个输入最多重试 3 次”不是每个节点初始化后清零而是把 attempts 累加上去。4.2 构造一个验证场景为了验证持久化和可追踪是否真正生效需要构造一个包含“失败后恢复”的测试场景。推荐的验证步骤如下创建一个工作流包含一个根节点和一个依赖根节点的子节点。把dispatch_fn写成第一次调用指定 agent 时抛异常第二次调用成功。执行根节点成功后执行子节点第一次失败。再次调用run_node第二次成功。查询agent_event确认出现了node_failed和node_succeeded两条事件。查询agent_node确认 attempts 变为 2status 为 succeeded。这个验证覆盖了三个核心能力状态更新、事件记录、重试恢复。只验证“程序能启动”没有意义还要验证状态从 failed 恢复到 succeeded 后数据库里的记录仍然一致。4.3 预期输出与验证结果当执行成功后查询agent_node表会看到类似下面的记录id: abc123 workflow_id: run_111 parent_id: null agent_name: booking status: succeeded input: {type: hotel} output: {status: ok, data: {type: hotel}} error: null attempts: 1查询agent_event时会看到两条事件event_type: node_started actor: runner payload: {attempt: 1} event_type: node_succeeded actor: runner payload: {artifact: result.json}如果子代理失败过一次再成功事件表会多一条node_failed并且节点表的 attempts 值会是 2。通过事件表的顺序可以完整还原执行过程。4.4 学习环境与生产环境的差异SQLite 版本适合学习和小型工具但生产环境必须做替换或增强。主要差异集中在以下几点维度学习环境生产环境数据库SQLite 单文件PostgreSQL 或 MySQL支持行级锁和事务隔离并发控制单进程WAL 缓解读写冲突多 worker 同时更新节点状态必须用SELECT ... FOR UPDATE持久化范围状态、事件、产物 URI增加调度信息、队列信息、执行环境元数据追踪链路单库查询通过 trace_id 对接 OpenTelemetry 等分布式追踪系统恢复策略当前节点失败则重跑当前节点需要处理多节点并发失败、等待子任务完成、分布式锁安全问题本地示例事件内容可能包含敏感数据需要脱敏、权限控制、审计日志保留策略不要直接把 SQLite 建表语句搬到 PostgreSQL 后就不管了。sqlite3 的自动类型约束很弱字段存 JSON 时也不会校验PostgreSQL 建议使用 JSONB 字段类型并对 workflow_id、status、parent_id 建联合索引。注意生产环境应该为agent_event表和artifact表设置数据保留策略。追踪数据是很有价值但无限保留会带来成本和合规风险。建议按天数归档或压缩。5. 常见问题与排查链路5.1 状态不一致节点显示 running但进程已经重启现象数据库里某个节点状态是 running但对应进程已经退出事件表里只有 node_started没有 node_succeeded 或 node_failed。可能原因进程被强杀异常处理没有机会写入失败事件。网络超时后子代理实际已成功但进程没有收到返回就崩溃了。检查方式查询事件表确认该节点 latest event ID。查看 artifact 表确认是否有 result.json。对照进程日志看是否在 node_started 之后发生了崩溃。处理建议启动时执行一轮“孤儿节点修复”把超过 timeout 且状态为 running 的节点标记为 failed。如果 artifact 已存在可以将节点标记为 succeeded避免重复执行。5.2 恢复后重复执行子代理现象使用resume_run后某些已经成功执行过的节点又被重新调用了一次。可能原因恢复逻辑判断的不是节点状态而是事件是否存在。代码在遍历子节点时误把“有 node_started 事件”当作“待恢复”导致已经成功但事件不全的节点被重跑。检查方式检查该节点的 status 是否为 succeeded。检查事件表确认 node_succeeded 是否存在。如果 status 是 succeeded但 node_succeeded 事件缺失说明事件写入逻辑有遗漏。处理建议恢复判断以agent_node.status为准而不是以事件为准。写一个幂等检查如果一个节点已经有成功产物就不要再次执行。5.3 事件缺失或乱序现象节点已经 succeeded但事件列表里没有对应的 node_succeeded。可能原因在写入事件前抛了异常导致状态更新成功但事件未写入。多线程并发写事件时自增 ID 顺序和业务顺序不一致。检查方式查看状态更新和事件写入是否在同一个事务里。检查是否使用了多连接并发写入事件表的时间戳是否可信。处理建议状态更新和事件写入必须放在同一个连接、同一个事务里。事件表增加sequence_no或使用created_at排序时要注意多进程时钟不一致问题推荐依赖自增 ID 加业务时间戳双重排序。5.4 并发更新导致 attempts 被覆盖现象两个 worker 同时处理同一个节点最终 attempts 值小于真实执行次数。可能原因代码先读 attempts再加一再写回但两个请求同时读到同一个旧值。SQLite 默认串行写但 PostgreSQL 在SELECT后UPDATE时如果没有行锁也会出现覆盖。检查方式查询节点更新历史看是否有两次 node_started 事件但 attempts 只增加了 1。处理建议更新语句改为原子操作UPDATE agent_node SET attempts attempts 1 WHERE id ?。生产环境使用SELECT ... FOR UPDATE锁定节点行再执行后续操作。5.5 排查清单遇到 subagent 工作流异常时按以下顺序排查查询workflow_run确认工作流整体状态。查询agent_node找出所有 status 不为 succeeded 的节点。查询agent_event按节点 id 过滤按自增 id 排序确认最后一个事件。查看artifact确认是否有产物落盘。检查节点输入与输出是否匹配排除模型返回异常格式的情况。查看节点错误字段确认是否真的执行失败还是仅追踪失败。再检查外围信息API 配额、网络超时、依赖服务状态。6. 最佳实践与扩展方向6.1 持久化工作流时的可复用清单下面的清单可以用于设计自己的 subagent 工作流引擎每次运行必须有唯一 run_id每个节点必须有唯一 node_id。每个节点必须记录 parent_id形成可追踪的树结构。每个节点必须有状态字段推荐使用 pending、running、succeeded、failed、paused、canceled。每次状态变化必须追加一条事件事件 actor 要明确。状态更新和事件写入要放在同一事务中。输入输出建议存 JSON 快照至少保存必要上下文。大文件不要直接进事件表存入 artifact事件里只留 URI。恢复逻辑以节点状态为准不能依赖事件存在性。每个节点执行前生成幂等键特别是有外部副作用的子代理。对失败的节点保留原始错误方便后续重试时做不同处理。6.2 用幂等键避免重复执行副作用如果 subagent 不只是“调用模型返回文本”还会调用外部 API、写数据库、发消息那么幂等性比状态恢复更重要。一个简单的做法是在输入数据中加入request_id并在目标系统里按 request_id 去重。input_data { request_id: uuid.uuid5(uuid.NAMESPACE_URL, f{node_id}-{attempt}).hex, payload: original_input, }重试时重新使用同一 request_id外部系统就可以识别“这次调用的任务和上次一样”从而避免重复下单、重复发消息等副作用。6.3 从单机示例扩展到分布式追踪本地示例把事件写入 SQLite生产环境通常需要把追踪数据接入 OpenTelemetry。思路是保留 trace_id在事件写入数据库的同时额外发送一个 span 到追踪后端。这样既能通过数据库查询恢复状态又能通过追踪系统查看调用链、耗时和资源消耗。具体可以分为两层业务追踪层记录 subagent 的输入、输出、状态流转保存在业务库。性能追踪层记录每次 subagent 调用的耗时、token 用量、模型名称、重试次数发送到时序数据库或链路追踪平台。即使正在使用 LangGraph、AutoGen 等框架也可以在工具调用装饰器里接入这层追踪不必重写框架本身。6.4 可视化与血缘是下一步重点状态表和事件表是可追踪的基础但直接看 SQL 记录不够直观。建议在追踪树之上提供两类视图执行时间线横向展示每个节点从创建到成功或失败的完整事件流。血缘依赖图展示父节点和子节点的依赖关系以及产物文件之间的关系。这两类视图可以从agent_node的 parent_id 和artifact表推导出来。展示层可以把节点状态映射成颜色绿色 succeeded、红色 failed、黄色 running、灰色 pending。把这两类视图做出来subagent 工作流就不再是一堆无法复现的临时调用而是一个具备可观测性和可维护性的业务系统。最终要记住一个核心判断subagent 工作流的价值不只在于“让模型把任务拆给更小的模型”更在于“把任务拆解、执行、失败、恢复的整个过程变成可以追溯的数据资产”。持久化和可追踪不是附加功能而是让 subagent 模式进入生产环境的必要条件。下一步建议从本文的 SQLite 示例出发选择一个你正在使用的 Agent 框架把状态表、事件表和恢复逻辑先加进去再逐步补充可视化、幂等和分布式追踪。