subagent工作流如何设计持久化与可追踪机制

1. subagent 工作流是什么，为什么需要 Persistent + Trackable

1.1 从单 Agent 到多 Agent：subagent 的出现

过去一年多，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 运行单元、每一个工具调用做全链路记录。记录内容包括：

• 本次执行的整体 ID（trace_id）。
• 每个 subagent 任务的 ID 和父级 ID（span_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.py

import 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_thread=False) 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.py

from 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_factory=lambda: 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_id=self.trace_id, parent_span_id=parent_span_id, span_id=self._gen_id("span"), name=name, ) 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.py

import 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_id=task_id, trace_id=self.tracer.trace_id, parent_span_id=parent_span_id, task_name=task_name, status="created", input_data=json.dumps(input_data, ensure_ascii=False), output_data=None, error_message=None, attempt=0, created_at=datetime.now().isoformat(), updated_at=datetime.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(f"subagent {name} 未注册") subagent = self.subagents[name] span = self.tracer.start_span(name=name, parent_span_id=parent_span_id) task_id = span.span_id # 创建初始记录 record = self.create_task_record( task_id=task_id, parent_span_id=parent_span_id, task_name=name, input_data=input_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_ascii=False)) # 更新为成功状态 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中体现了完整的持久化和追踪流程：

1. 创建一个 Span，以 span_id 作为 task_id。
2. 创建任务记录，状态为created。
3. 更新状态为running，持久化。
4. 调用 subagent 处理函数。
5. 成功则更新为success并保存输出数据。
6. 失败则重试，每次重试都记录 attempt；超过最大重试次数则更新为failed。

这样，无论工作流执行到哪一步，数据库中都留有一条完整的任务记录。即使进程崩溃，也能通过get_running_tasks()找到中断的任务。

4.5 运行与验证

现在写主程序，演示一个“数据收集 → 清洗 → 汇总”的三层 subagent 工作流。

文件路径：main.py

import 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": [f"source_{kw}" for kw in keywords]}, ensure_ascii=False) 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_ascii=False) 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_ascii=False) 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_id=root_task_id, ) # 第二步：清洗数据 clean_result = engine.run_subagent( "clean_data", collect_result["output"], parent_span_id=root_task_id, ) # 第三步：汇总数据 summary_result = engine.run_subagent( "summarize_data", clean_result["output"], parent_span_id=root_task_id, ) print("=" * 50) print("工作流执行完成") print(f"Trace 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

SQLite 适合单机、低并发的场景。如果你的 subagent 工作流要支持多个 subagent 并行运行，每个 subagent 都在更新数据库，就可能出现锁冲突。建议把写操作统一收口到一个队列，或者直接升级为 PostgreSQL。

5.2 进程崩溃后如何恢复？

恢复逻辑的伪代码如下：

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 跨度关系混乱

如果在同一个工作流中，某个 subagent 又派生了子 subagent（即 subagent 内部再调用其他 subagent），则需要把当前 subagent 的span_id作为子任务的parent_span_id传下去，而不是直接透传根任务的 ID。这样才能形成正确的树形追踪结构。

5.4 任务记录中输出数据过大

在实际 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，你的持久化模型需要怎么调整？想清楚这个问题，你对“持久化”的理解会更进一步。