企业官网建设流程全解析

这次我们来看一个最近在 Show HN 上出现的方向：让普通的 subagent 工作流变得持久化（persistent）且可追踪（trackable）。Show HN 是 Hacker News 上作者展示自己开源项目的栏目，所以这个标题翻译过来就是“给我那个普通子智能体工作流加上保存和追踪能力”。用大白话说，它解决的是 AI Agent 开发里一个特别常见的痛点：主 Agent 派了一堆 subagent 干活，任务执行过程中上下文、中间结果、调用记录全在内存里，进程一崩就全没了，出了问题也没法复盘是哪个子任务花了多少 token、哪一步失败的。

这个项目方向的三个关键词已经说明了核心卖点：ordinary，不要求你推倒现有 Agent 框架重来；persistent，任务状态、上下文、中间产物可以落盘恢复；trackable，每次 subagent 调用的输入输出、耗时、token 消耗都可以查询和审计。下面我会先从核心能力、适用场景开始，然后给出一个可落地的持久化模型设计，再带你把任务执行器、恢复机制、追踪查询 API、批量任务队列完整走一遍，最后补上常见的坑和工程化建议。

1. 核心能力速览

在直接看代码之前，先把这个项目方向能做什么、不能做什么放在一张表里。因为标题只给出了能力边界，没有具体的仓库地址和版本，所以表格里的参数我尽量写成通用范围，实际以你拿到源码后的文档为准。

这里需要强调一点：标题里的 persistent 和“persistent homology”没有任何关系，后者是拓扑数据分析里的持续同调概念，属于数学方向。这个项目里的 persistent 指的是任务状态在进程退出、崩溃、重启之后仍然存在，可以继续执行或回溯。

2. 适用场景与使用边界

这个方案适合谁？最典型的是已经在用 LangGraph、AutoGen、CrewAI 这类框架，或者自己手写 Agent 循环的团队。你现在的 subagent 流程大概率是这样的：主 Agent 拆解任务，创建 subagent，subagent 调用 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 -> RUNNING
• RUNNING -> WAITING / COMPLETED / FAILED / INTERRUPTED
• WAITING -> RUNNING
• INTERRUPTED -> 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，就是这个意思，不改框架，只加钩子。

通用检查清单如下：

注意，这里没有写死任何第三方库的精确版本号。实际项目里，依赖版本以项目 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_key=True) parent_task_id: Mapped[str | None] = mapped_column(String(64), index=True) agent_name: Mapped[str] = mapped_column(String(128), index=True) status: Mapped[str] = mapped_column(String(32), index=True) payload: Mapped[dict] = mapped_column(JSON, default=dict) result: Mapped[dict | None] = mapped_column(JSON, nullable=True) error: Mapped[str | None] = mapped_column(String, nullable=True) heartbeat_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class TaskEvent(Base): __tablename__ = "task_events" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) task_id: Mapped[str] = mapped_column(String(64), index=True) event_type: Mapped[str] = mapped_column(String(64)) payload: Mapped[dict] = mapped_column(JSON, default=dict) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class LLMSpan(Base): __tablename__ = "llm_spans" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) trace_id: Mapped[str] = mapped_column(String(64), index=True) span_id: Mapped[str] = mapped_column(String(64), index=True) parent_span_id: Mapped[str | None] = mapped_column(String(64), nullable=True) task_id: Mapped[str] = mapped_column(String(64), index=True) model_name: Mapped[str] = mapped_column(String(128)) prompt: Mapped[dict] = mapped_column(JSON, default=dict) response: Mapped[dict | None] = mapped_column(JSON, nullable=True) prompt_tokens: Mapped[int] = mapped_column(Integer, default=0) completion_tokens: Mapped[int] = mapped_column(Integer, default=0) duration_ms: Mapped[int] = mapped_column(Integer, default=0) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.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(f"task {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_id=task_id, event_type="status_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(seconds=timeout_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_id=task.id, event_type="recovered", 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 的全过程都写入了数据库。

操作步骤：

1. 启动数据库和 Worker。
2. 创建一个最简单 subagent 任务，输入一条 prompt。
3. 观察 tasks 表和 task_events 表的数据变化。

预期结果：

• tasks 表出现一条记录，状态从 PENDING 变为 RUNNING，最后变为 COMPLETED。
• task_events 表至少有三条 status_changed 事件。

判断成功标准：进程重启后，tasks 表里的 COMPLETED 记录仍然存在，查询接口可以正常返回结果。

6.2 验证崩溃恢复

测试目的：确认任务执行到一半进程被 kill，重启后能找回并重新入队。

操作步骤：

1. 创建一个 subagent 任务，在 subagent 函数里加一个 sleep，人为拉长执行时间。
2. 在 RUNNING 状态下直接 kill 掉 Worker 进程。
3. 修改心跳超时阈值为 10 秒，重启 Worker。
4. 查看恢复模块扫描结果。

预期结果：

• 重启后，这个任务的状态由 RUNNING 变成 PENDING。
• 日志里出现 recovered 事件。
• Worker 重新消费任务，最终进入 COMPLETED 或 FAILED。

常见失败原因：如果恢复后重新执行需要完整的上下文，但 context_messages 没有持久化，任务会在第二次调用时抛出“上下文不存在”的错误。这就是前面强调上下文必须持久化的原因。

6.3 验证调用链追踪

测试目的：确认每次 LLM 调用都记录了 trace_id、token 和耗时。

操作步骤：

1. 创建一个主任务，内部派发两个 subagent。
2. 每个 subagent 至少调用一次 LLM。
3. 查询 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 任务可以并发消费，失败任务不阻塞其他任务。

操作步骤：

1. 构造一个包含 10 个任务的批量提交。
2. 其中一条故意使用一个会报错的 prompt，模拟模型调用失败。
3. 观察队列消费结果。

预期结果：

• 9 个任务进入 COMPLETED，1 个进入 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_model=TaskOut) def create_task(body: TaskCreate): # 生成 task_id，写入 tasks 表，状态为 PENDING，投递到队列 ... @app.get("/api/tasks/{task_id}", response_model=TaskOut) 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": [f"file_{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. 常见问题与排查方法

这里整理几个最容易踩的坑和排查路径。

这里面最值得关注的是最后两个。持久化虽然能保存状态，但如果你不处理幂等性，恢复机制反而可能造成重复操作。非幂等任务要么不自动重试，要么在重试前做幂等检查，这是生产环境必须考虑的问题。

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 先跑通，验证完恢复和追踪的价值再上生产配置。