1. subagent 工作流是什么为什么需要 Persistent Trackable1.1 从单 Agent 到多 Agentsubagent 的出现过去一年多LLM 应用开发逐渐从“单次问答”走向“多步任务编排”。单 Agent 的模式很简单用户给一个 prompt模型调用工具返回结果。这种模式适合简单问答但遇到“需要分阶段调研、多轮验证、多角色协作”的复杂任务时单 Agent 经常顾此失彼。于是 subagent子代理的概念开始普及。subagent 本质上是一个承载独立任务的 Agent它有自己的 system prompt、自己的工具集、自己的上下文窗口由主 Agent也可以叫 orchestrator / supervisor按需派发任务。举个例子一个“行业调研报告生成”工作流可以拆成三个 subagent资料收集 Agent负责搜索和整理行业公开数据。数据分析 Agent负责对收集到的数据进行统计和图表生成。报告撰写 Agent负责把分析结果组织成结构化报告。主 Agent 按顺序调度这三个 subagent每个 subagent 专注做一件事最后汇总输出。这种分层协作的好处很明显上下文不会被撑爆每个子任务可以并行或串联执行整体效果比单 Agent 硬扛要好得多。但当你真的把 subagent 工作流跑起来尤其是高频、长时间、多轮次地运行时管理问题就浮现了。1.2 普通 subagent 工作流的三大痛点第一状态丢失。很多 Agent 编排框架默认把 subagent 的运行状态放在内存里。进程一重启、任务一超时、服务器一更新所有中间状态全部归零。用户可能已经跑到第 5 个子任务系统一崩就要从头再来。第二不可追踪。普通工作流的日志往往是分散的主 Agent 一条日志、subagent 一条日志、工具调用又一条日志彼此之间没有统一的 trace_id。排查问题时连“这个结果到底是哪个 subagent 在什么上下文下生成的”都很难回答。第三结果不可复现。AI 模型本身有随机性如果工作流不记录输入、输出、模型参数、工具调用顺序那么同一个任务跑两次结果可能不同而你不知道差异来自哪里。这三个痛点本质上指向同一个问题subagent 工作流缺少持久化persistent和可追踪性trackable设计。1.3 Persistent 和 Trackable 到底解决什么问题所谓 persistent就是把工作流的运行状态、中间结果、任务进度、上下文信息持久化到外部存储中而不是只放在内存里。这样即使进程重启也能从断点恢复即使任务执行了很长时间也不会因为临时故障而丢失。所谓 trackable则是对工作流的每一次执行、每一个 subagent 运行单元、每一个工具调用做全链路记录。记录内容包括本次执行的整体 IDtrace_id。每个 subagent 任务的 ID 和父级 IDspan_id / parent_span_id。任务的输入、输出、耗时、状态。模型调用参数temperature、model 版本等。工具调用细节。Persistent 解决的是“任务能不能继续跑下去”的问题Trackable 解决的是“任务跑得好不好、过程中发生了什么”的问题。两者叠加起来工作流才真正进入可管、可控、可审计、可优化的状态。需要注意的是persistent 并不等于简单的日志打印。日志只是输出到文件里工作流本身并不具备从日志中恢复执行状态的能力。真正的持久化是把状态写入数据库或事件流让系统可以在任意时间点重建工作流的上下文。2. 架构设计给 subagent 工作流加上“记忆”和“日志”2.1 整体设计原则在实现 persistent 和 trackable 之前先明确几个设计原则这些原则会直接决定后面代码怎么写用事件驱动的方式来记录状态变化。不要只记录最终结果每一步的状态变更创建、开始、成功、失败、重试都视为一个事件。任务和数据分离。subagent 的任务定义做什么和任务执行记录做得怎么样要分开存储。为每次执行生成全局唯一的追踪 ID。从主 Agent 下发任务开始所有 subagent 都共享同一个 trace_id通过 parent_span_id 维护父子关系。状态存储需要支持并发。多个 subagent 可能同时运行存储层要保证并发更新的安全性。写操作要轻量。不要在关键路径上做昂贵的序列化或网络调用否则会拖慢工作流。2.2 核心概念事件溯源与运行上下文事件溯源Event Sourcing是一种架构模式不保存系统的“当前状态”而是把每一次状态变更都作为不可变事件记录下来需要当前状态时把事件按顺序重放出来。把事件溯源用在 subagent 工作流里非常合适。原因有三点天然可追踪。事件本身就是日志按时间顺序回放能看到整个工作流完整的历史。天然可恢复。进程崩溃后从事件存储中重放就能恢复所有 subagent 的状态。天然可审计。谁在什么时候给哪个 subagent 派发了什么任务模型返回了什么结果全部有据可查。不过完整的事件溯源维护成本较高。实际工程中更常见的是“事件记录 状态快照”的混合方案每个任务的关键节点记录事件同时每隔一段时间保存一份状态快照恢复时加载快照再重放增量事件。运行上下文RunContext是另一个重要概念。每个 subagent 在执行时需要一个包含以下信息的上下文对象dataclass class RunContext: trace_id: str # 全局追踪 ID parent_span_id: str # 父任务 ID主 Agent 发起的任务为空 span_id: str # 当前任务 ID task_name: str # 任务名称 attempt: int # 第几次尝试 created_at: str # 创建时间 metadata: dict # 附加元数据2.3 追踪模型Trace、Span、Event可追踪性的基础模型可以借用分布式链路追踪领域的概念Trace追踪一次完整的工作流执行包含主 Agent 和所有 subagent 的调用。Span跨度一次具体的工作单元比如一个 subagent 的执行、一次工具调用、一次模型请求。Event事件Span 内部发生的具体事件比如子任务开始、结束、报错、重试。它们的关系是一个 Trace 包含多个 Span一个 Span 内包含多个 Event。通过 trace_id 可以把所有相关记录串联起来通过 parent_span_id 可以还原出层级关系。这种模型的好处是上层可以复用开源生态的很多工具比如 OpenTelemetry 的链路追踪 UI、Jaeger 的依赖分析面板、Grafana 的日志查询都能和你的系统对接而不用从零开发一套可视化方案。3. 环境准备与项目结构3.1 技术选型说明本文示例用 Python 3.10 来实现存储层使用 SQLite通过 sqlite3 标准库操作本地数据库文件。SQLite 对零基础读者最友好不需要额外安装数据库服务也方便在本地快速验证。生产环境建议替换为 PostgreSQL 或 MySQL代码逻辑基本不变只要把存储实现换成对应数据库驱动即可。版本说明AI Agent 编排框架如 LangGraph、CrewAI、AutoGen 等迭代速度非常快不同版本的 API 差异很大。本文示例不依赖任何特定框架使用 Python 标准库模拟 subagent 工作流的编排过程重点演示持久化和追踪设计。核心思路可以适配到任何 Agent 框架中。3.2 项目目录结构先创建项目目录mkdir subagent-workflow-demo cd subagent-workflow-demo目录结构如下subagent-workflow-demo/ ├── main.py # 入口程序演示工作流运行 ├── storage.py # 持久化存储层SQLite ├── tracer.py # 追踪上下文管理 ├── workflow.py # subagent 工作流编排器 └── database.db # SQLite 数据库文件运行后生成本文需要创建storage.py、tracer.py、workflow.py、main.py四个文件下面逐一实现。4. 实战构建一个可持久化、可追踪的 subagent 工作流4.1 定义工作流与任务模型先定义工作流中需要使用的数据模型。为了简洁直接在storage.py中定义。文件路径storage.pyimport json import sqlite3 from dataclasses import dataclass, asdict from datetime import datetime from typing import Optional import uuid def gen_id(prefix: str) - str: 生成带前缀的 ID例如 subagent_20250217_xxxx return f{prefix}_{datetime.now().strftime(%Y%m%d%H%M%S)}_{uuid.uuid4().hex[:8]} dataclass class TaskRecord: 任务记录对应一次 subagent 执行 task_id: str # 任务 ID trace_id: str # 全局追踪 ID parent_span_id: str # 父任务 ID task_name: str # 任务名称 status: str # 状态: created/running/success/failed input_data: str # 输入数据JSON 字符串 output_data: Optional[str] # 输出数据JSON 字符串 error_message: Optional[str] # 错误信息 attempt: int # 重试次数 created_at: str # 创建时间 updated_at: str # 更新时间这个模型里trace_id用于全局串联parent_span_id用于维护树形结构status用于追踪状态流转attempt用于记录重试次数。4.2 实现持久化存储存储层需要支持任务记录的保存和查询。使用 SQLite 存储建表语句和操作函数如下仍然在storage.py中添加class Storage: SQLite 持久化存储保存 subagent 工作流的所有任务记录 def __init__(self, db_path: str database.db): self.db_path db_path self.conn sqlite3.connect(db_path, check_same_threadFalse) self.conn.row_factory sqlite3.Row self._init_table() def _init_table(self): 初始化数据表 self.conn.execute( CREATE TABLE IF NOT EXISTS task_records ( task_id TEXT PRIMARY KEY, trace_id TEXT NOT NULL, parent_span_id TEXT NOT NULL, task_name TEXT NOT NULL, status TEXT NOT NULL, input_data TEXT NOT NULL, output_data TEXT, error_message TEXT, attempt INTEGER DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_trace_id ON task_records(trace_id); ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_parent_span_id ON task_records(parent_span_id); ) self.conn.commit() def save_task(self, record: TaskRecord): 保存或更新任务记录 self.conn.execute( INSERT INTO task_records ( task_id, trace_id, parent_span_id, task_name, status, input_data, output_data, error_message, attempt, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(task_id) DO UPDATE SET status excluded.status, output_data excluded.output_data, error_message excluded.error_message, attempt excluded.attempt, updated_at excluded.updated_at , ( record.task_id, record.trace_id, record.parent_span_id, record.task_name, record.status, record.input_data, record.output_data, record.error_message, record.attempt, record.created_at, record.updated_at, ), ) self.conn.commit() def get_task(self, task_id: str) - Optional[dict]: 根据任务 ID 查询记录 cursor self.conn.execute( SELECT * FROM task_records WHERE task_id ?, (task_id,) ) row cursor.fetchone() return dict(row) if row else None def get_tasks_by_trace(self, trace_id: str) - list[dict]: 根据追踪 ID 查询整个工作流的所有任务记录 cursor self.conn.execute( SELECT * FROM task_records WHERE trace_id ? ORDER BY created_at, (trace_id,), ) return [dict(row) for row in cursor.fetchall()] def get_running_tasks(self) - list[dict]: 查询所有未完成任务用于断点恢复 cursor self.conn.execute( SELECT * FROM task_records WHERE status IN (created, running) ) return [dict(row) for row in cursor.fetchall()] def close(self): self.conn.close()这里有一个关键设计save_task使用INSERT ... ON CONFLICT DO UPDATE实现“插入或更新”。任务创建后状态是created运行中更新为running完成时更新为success或failed每次都复用同一行记录。get_running_tasks()是断点恢复的核心方法。进程重启后可以通过它找到所有尚未完成的任务从断点继续执行。4.3 实现追踪上下文追踪上下文管理器的职责是为每个 subagent 生成唯一的 ID并自动维护父子关系。文件路径tracer.pyfrom dataclasses import dataclass, field from datetime import datetime dataclass class Span: 追踪跨度对应一次 subagent 执行 trace_id: str parent_span_id: str span_id: str name: str start_time: str field(default_factorylambda: datetime.now().isoformat()) class Tracer: 追踪上下文管理器 def __init__(self, trace_id: str None): self.trace_id trace_id or self._gen_id(trace) self._spans: list[Span] [] staticmethod def _gen_id(prefix: str) - str: import uuid return f{prefix}_{datetime.now().strftime(%H%M%S)}_{uuid.uuid4().hex[:8]} def start_span(self, name: str, parent_span_id: str ) - Span: 开启一个新的追踪跨度 span Span( trace_idself.trace_id, parent_span_idparent_span_id, span_idself._gen_id(span), namename, ) self._spans.append(span) return span def get_spans(self) - list[Span]: 获取当前 trace 下的所有 span return self._spans这里start_span接收一个parent_span_id。主 Agent 启动工作流时传入空字符串subagent 创建时传入上层任务的span_id这样就形成了树形追踪结构。注意Tracer只负责在内存中维护追踪上下文真正的持久化仍然通过Storage完成两者配合才是完整方案。4.4 编写工作流编排器下面进入核心部分一个简化版的 subagent 工作流编排器。它模拟了主 Agent 派发任务、subagent 执行任务、记录持久化和追踪信息的过程。文件路径workflow.pyimport json import time from datetime import datetime from storage import Storage, TaskRecord, gen_id from tracer import Tracer class SubAgent: 模拟一个 subagent实际项目中可以替换为 LLM 工具的调用 def __init__(self, name: str, process_func): self.name name self.process_func process_func def run(self, input_data: str) - str: 执行任务输入和输出都是 JSON 字符串 return self.process_func(input_data) class WorkflowEngine: 工作流编排器负责派发任务、持久化记录、维护追踪 def __init__(self, storage: Storage, tracer: Tracer): self.storage storage self.tracer tracer self.subagents: dict[str, SubAgent] {} def register_subagent(self, name: str, process_func): 注册一个 subagent self.subagents[name] SubAgent(name, process_func) def create_task_record( self, task_id: str, parent_span_id: str, task_name: str, input_data: dict, ) - TaskRecord: 创建任务记录并持久化 record TaskRecord( task_idtask_id, trace_idself.tracer.trace_id, parent_span_idparent_span_id, task_nametask_name, statuscreated, input_datajson.dumps(input_data, ensure_asciiFalse), output_dataNone, error_messageNone, attempt0, created_atdatetime.now().isoformat(), updated_atdatetime.now().isoformat(), ) self.storage.save_task(record) return record def run_subagent( self, name: str, input_data: dict, parent_span_id: str , max_retries: int 2, ) - dict: 运行一个 subagent 并记录执行状态 if name not in self.subagents: raise ValueError(fsubagent {name} 未注册) subagent self.subagents[name] span self.tracer.start_span(namename, parent_span_idparent_span_id) task_id span.span_id # 创建初始记录 record self.create_task_record( task_idtask_id, parent_span_idparent_span_id, task_namename, input_datainput_data, ) # 更新状态为 running record.status running record.updated_at datetime.now().isoformat() self.storage.save_task(record) # 执行任务支持重试 attempt 0 while attempt max_retries: try: # 调用 subagent 的处理函数 output subagent.run(json.dumps(input_data, ensure_asciiFalse)) # 更新为成功状态 record.status success record.output_data output record.attempt attempt record.updated_at datetime.now().isoformat() self.storage.save_task(record) return { task_id: task_id, status: success, output: json.loads(output), } except Exception as e: attempt 1 if attempt max_retries: # 最终失败 record.status failed record.error_message str(e) record.attempt attempt - 1 record.updated_at datetime.now().isoformat() self.storage.save_task(record) raise # 重试前更新状态 record.status running record.attempt attempt record.updated_at datetime.now().isoformat() self.storage.save_task(record) time.sleep(0.2) # 理论不会走到这里防御性代码 raise RuntimeError(subagent 执行异常)在run_subagent中体现了完整的持久化和追踪流程创建一个 Span以 span_id 作为 task_id。创建任务记录状态为created。更新状态为running持久化。调用 subagent 处理函数。成功则更新为success并保存输出数据。失败则重试每次重试都记录 attempt超过最大重试次数则更新为failed。这样无论工作流执行到哪一步数据库中都留有一条完整的任务记录。即使进程崩溃也能通过get_running_tasks()找到中断的任务。4.5 运行与验证现在写主程序演示一个“数据收集 → 清洗 → 汇总”的三层 subagent 工作流。文件路径main.pyimport json from storage import Storage from tracer import Tracer from workflow import WorkflowEngine def collect_data(input_json: str): 模拟数据收集 subagent data json.loads(input_json) keywords data[keywords] # 模拟耗时操作 time.sleep(0.5) return json.dumps({sources: [fsource_{kw} for kw in keywords]}, ensure_asciiFalse) def clean_data(input_json: str): 模拟数据清洗 subagent data json.loads(input_json) sources data[sources] cleaned [s.upper() for s in sources] return json.dumps({cleaned_sources: cleaned}, ensure_asciiFalse) def summarize_data(input_json: str): 模拟数据汇总 subagent data json.loads(input_json) cleaned data[cleaned_sources] return json.dumps({summary: f共收集 {len(cleaned)} 条来源{, .join(cleaned)}}, ensure_asciiFalse) def main(): storage Storage(database.db) tracer Tracer() engine WorkflowEngine(storage, tracer) # 注册 subagent engine.register_subagent(collect_data, collect_data) engine.register_subagent(clean_data, clean_data) engine.register_subagent(summarize_data, summarize_data) # 主任务作为根 span root_span tracer.start_span(main_workflow) root_task_id root_span.span_id # 第一步收集数据 collect_result engine.run_subagent( collect_data, {keywords: [AI, Agent, LLM]}, parent_span_idroot_task_id, ) # 第二步清洗数据 clean_result engine.run_subagent( clean_data, collect_result[output], parent_span_idroot_task_id, ) # 第三步汇总数据 summary_result engine.run_subagent( summarize_data, clean_result[output], parent_span_idroot_task_id, ) print( * 50) print(工作流执行完成) print(fTrace ID: {tracer.trace_id}) print(f最终结果: {summary_result[output][summary]}) print( * 50) # 查询整个工作流的任务记录 tasks storage.get_tasks_by_trace(tracer.trace_id) print(\n任务追踪记录) print(f{task_id:20} {任务名称:20} {状态:10} {Attempt:8}) print(- * 60) for task in tasks: print( f{task[task_id]:20} {task[task_name]:20} f{task[status]:10} {task[attempt]:8} ) storage.close() if __name__ __main__: import time main()运行程序python main.py预期输出类似下面这样ID 部分因时间戳和随机数而不同 工作流执行完成 Trace ID: trace_141530_ab12cd34 最终结果: 共收集 3 条来源SOURCE_AI, SOURCE_AGENT, SOURCE_LLM 任务追踪记录 task_id 任务名称 状态 Attempt ------------------------------------------------------------ span_141530_ef56 main_workflow created 0 span_141530_ab78 collect_data success 0 span_141530_cd90 clean_data success 0 span_141530_ef12 summarize_data success 0注意main_workflow的状态是created因为主任务只是发起者没有执行体。如果你希望主任务也显示为success需要在main()中手动更新其状态。现在打开database.db你能看到所有任务记录都保存在 SQLite 表中。这验证了持久化的效果工作流执行完以后所有中间数据和状态都落盘了而不是跟着进程一起消失。5. 常见问题与排查思路5.1 数据库被锁database is locked问题现象常见原因解决思路sqlite3.OperationalError: database is lockedSQLite 同一时间只允许一个写事务多个 subagent 并发写时会出现锁冲突使用check_same_threadFalse生产环境换 PostgreSQL/MySQL或对写操作加队列串行化SQLite 适合单机、低并发的场景。如果你的 subagent 工作流要支持多个 subagent 并行运行每个 subagent 都在更新数据库就可能出现锁冲突。建议把写操作统一收口到一个队列或者直接升级为 PostgreSQL。5.2 进程崩溃后如何恢复问题现象常见原因解决思路工作流运行一半进程被 kill重启后所有任务状态丢失没有把created/running状态的任务持久化启动时调用get_running_tasks()找出未完成任务根据任务名称和输入数据重新派发恢复逻辑的伪代码如下def recover_workflow(storage, engine): running_tasks storage.get_running_tasks() for task in running_tasks: print(f恢复任务: {task[task_name]}, task_id{task[task_id]}) # 根据 task_name 重新注册 subagent继续执行 # 注意需要处理输入数据从 JSON 字符串还原这里有一个工程细节需要注意恢复时子任务可能已经执行过但没写成功状态也可能没执行就崩溃了。最稳妥的方案是让任务实现“幂等”也就是同一个任务重复执行多次结果一致。对于纯数据转换类任务很容易做到如果涉及调用外部 API则可能需要做去重判断。5.3 Trace 跨度关系混乱问题现象常见原因解决思路查询出的任务记录无法看出清晰的调用层级parent_span_id在创建时传错了检查run_subagent的parent_span_id参数主任务创建 span 后把root_span.span_id作为后续所有 subagent 的父 ID如果在同一个工作流中某个 subagent 又派生了子 subagent即 subagent 内部再调用其他 subagent则需要把当前 subagent 的span_id作为子任务的parent_span_id传下去而不是直接透传根任务的 ID。这样才能形成正确的树形追踪结构。5.4 任务记录中输出数据过大问题现象常见原因解决思路数据库查询越来越慢subagent 输出大量文本直接存到output_data字段导致单行数据过大对 output_data 做截断或压缩把大的输出放到对象存储数据库只保存引用路径在实际 LLM 应用中subagent 的输出可能是几百 KB 甚至几 MB 的文本。直接塞进 SQLite 字段不是好主意。更推荐的做法是输出的完整内容写入文件或对象存储数据库只记录 URI展示时按需读取。6. 最佳实践与工程建议6.1 持久化设计建议持久化层是 subagent 工作流的骨架设计时要注意几点。第一任务记录要支持幂等更新。同一个任务 ID 可能被更新多次存储层必须支持upsert存在则更新不存在则插入避免出现重复记录。第二给状态字段加约束。不要随意发明状态值。建议在设计阶段就确定状态机例如created → running → success ↘ failed ↗ retrying重试中每个状态之间的转换要可追踪最好在事件表中记录状态变更时间。第三定期清理历史数据。任务记录增长很快尤其是包含输入输出数据的表。可以按 trace_id 归档旧数据比如只保留最近 30 天的详细记录更早的归入冷存储。第四重启恢复要设计全局优雅退出。在生产环境中不建议直接 kill 进程恢复。通过接收 SIGTERM 信号让正在运行的 subagent 完成当前步骤并更新状态后再退出能最大程度减少恢复时的重复执行。6.2 追踪与可观测性建议追踪不只是保存几条日志而是一个完整的可观测性体系。首先统一 trace_id 的传递方式。如果 subagent 内部还要发起 HTTP 调用其他服务需要把 trace_id 放到 HTTP 请求头中例如X-Trace-Id让下游服务也可以关联起来。其次日志与追踪关联。每一条日志都应该包含 trace_id 和 span_id。推荐使用结构化日志JSON 格式这样在日志平台中可以直接按 trace_id 过滤出一次完整工作流的所有日志。再次与 OpenTelemetry 集成。如果你使用的是 Java、Go 或 Node.js可以直接使用 OpenTelemetry SDK 管理 Span原生支持导出到 Jaeger、Zipkin 等追踪系统。Python 也有对应的 OpenTelemetry SDK可以把自定义 Span 映射为标准 Span省去自建可视化面板的工作。6.3 多 Agent 协作的稳定性建议subagent 工作流稳定性差很多时候不是模型的问题而是编排层缺乏兜底。所有 subagent 的处理函数必须做异常捕获。在run_subagent中虽然加了重试逻辑但如果 subagent 内部抛出的异常类型不确定例如超时异常、API 限流异常、JSON 解析异常建议分类处理限流类重试解析类直接失败。外部 API 调用要设超时。subagent 调用 LLM API 时如果请求卡住工作流会一直阻塞。给每次外部调用加上超时时间超时后抛异常并触发重试。任务要支持取消。如果用户在中途发现输入错误需要能取消整个工作流。建议在工作流中设置一个取消标记每执行完一个 subagent 就检查一次标记。对最终结果要做校验。AI 生成的结果不保证格式正确。比如summarize_data返回的 JSON 可能不合法建议在 subagent 返回后立即做 JSON 校验而不是把脏数据传给下一个 subagent。6.4 成本与性能控制每次 subagent 调用都会产生模型费用和耗时工作流层需要做成本控制。一种方式是在调用前检查预算。如果输入数据量过大可以先让一个“评估 subagent”预估本次调用的 token 消耗超出预算则拒绝执行。另一种方式是结果缓存。对于输入相同、参数相同的 subagent 调用直接命中缓存而不是重新调用模型。缓存 key 可以设为task_name hash(input_data) model temperature。同时控制重试次数。重试虽然能提高成功率但每次重试都会重新消耗 token。推荐在编排层面做一次轻量级校验例如检查输入数据是否满足任务要求避免“明知输入不合法还反复重试”的情况。7. 总结与下一步学习建议本文以 subagent 工作流为对象完整演示了如何把“跑完就丢”的临时任务改造成“持久化 可追踪”的工程系统。核心要点可以概括为三条用 SQLite 或关系型数据库持久化任务记录让工作流不依赖进程内存崩溃后可从断点恢复。用 trace_id parent_span_id 建立树形追踪模型让每次执行、每个 subagent、每次重试都有据可查。用事件/状态机维护任务生命周期让重试、恢复、审计都变得标准化。如果你正在使用 LangGraph、CrewAI、AutoGen 等框架可以把本文的思路迁移过去实现一个自定义的持久化后端或者在每次节点执行时通过回调写入追踪记录。下一步建议探索的方向在 subagent 内部加入真正的 LLM 调用替换本文中模拟的process_func。引入异步执行asyncio支持多个 subagent 并行运行。将追踪数据接入 Jaeger 或 Grafana Tempo实现可视化链路查询。为工作流设计一个简单的 Web 管理面板提供任务列表、详情查看、手动重试等功能。下面留一个问题供你思考如果两个 subagent 需要在各自完成后把结果合并给第三个 subagent你的持久化模型需要怎么调整想清楚这个问题你对“持久化”的理解会更进一步。