从 ad-hoc 到 accountable把临时 subagents 改造成可持久、可问责的 AI 团队最近有一个来自 Hacker News 的 Show HN 标题让人印象很深“Turn ad-hoc subagents into durable, accountable AI teams”。如果你最近研究过多 Agent 协作、AI Agent 工程化应该能立刻意识到这个标题戳中了什么今天大量 AI Agent 应用都停留在“临时脚本”阶段——Agent 被临时组装、用完即弃、无状态、无记录、无恢复能力。Demo 阶段一切美好一旦放进生产环境问题就像潮水一样涌出来。有个很典型的场景。你开发了一个内容审核助手上一次运行它成功处理了上千条文本接口也接好了。过了一周业务方要求重跑你发现会话上下文早被模型服务端的清理策略删了中间分析结果分散落在临时文件里那个“AI 助手”已经完全忘了之前约定的输出格式。你想复盘上次任务里某个 Agent 为什么放行了一条高风险内容却发现根本没有决策日志。你终于意识到真正阻碍 AI Agent 落地的不是模型效果而是工程化缺失。这篇文章直接说清楚一件事临时 subagents 和持久化 AI 团队之间差的不是“更强的模型”而是任务状态、共享上下文、决策审计这三个工程维度。我会用一套最小可运行的状态存储 审计日志 断点恢复的代码示例带你搭出一个 durable、accountable 的多 Agent 团队雏形并给出生产环境的坑与应对。无论你是在做内容审核、自动化报告、代码审查还是企业知识库助手这套思路都能复用。1. 这篇文章真正要解决的问题1.1 临时子代理的三个通病先说 ad-hoc subagents 的典型表现。很多开发者在写多 Agent 应用时习惯在代码里动态创建一个 Agent让它执行某个子任务拿到结果后就把这个 Agent 丢弃。下一次需要同样能力再创建一次。这种模式看起来轻巧但它有三个非常影响生产的缺陷。第一个是状态丢失。Agent 执行过程中的中间产物、上下文摘要、已经完成的子任务结果全部保存在进程内存或随机的临时文件里。进程一重启会话一过期这些信息就找不回来了。第二个是上下文隔离。多个 Agent 虽然服务于同一个总任务但彼此不知道对方做过什么。Agent A 处理后的输出没有被结构化地放入共享工作空间Agent B 接手时只能重新从原始输入开始既浪费 token还容易出现结论冲突。第三个是决策黑盒。整个任务执行流水线没有“为什么”的记录。某个 Agent 采纳了一条数据、放弃了一条规则、做了某个判断事后完全无法追溯。一旦出了问题你连排查入口都找不到。1.2 谁的痛点最明显如果你正处于下面这几类处境这个问题尤其值得关注。第一类是正在把 Agent Demo 推向生产环境的开发者。你可能会发现单次效果再好把 Agent 放进一个需要持续运行、多人协作、异常频发的业务系统里稳定性反而成了最大的拦路虎。第二类是搭建了多 Agent 协作平台的人。平台允许用户自定义多个子代理但如果这些子代理没有统一的任务状态和审计机制平台就是一堆独立的 API 调用的集合根本称不上“团队”。第三类是有合规要求的组织。无论在金融、医疗还是内容安全领域AI 做出的决策必须被记录、被复核、被审计。没有 accountabilityAI 系统很难通过内部风控。1.3 这是工程问题不是模型问题我想先给出一个贯穿全文的判断Agent 工程化远不是“写 Prompt”的事情它是软件工程里非常经典的问题——状态管理、任务编排、可观测性、可恢复性。一个 durable 的 AI 团队和一个高并发的分布式系统在架构思维上是同构的都要有持久化状态、都要有故障恢复、都要有可追溯的事件日志。把思考框架从“写脚本”切换到“建设系统”是这篇文章帮助你完成的最重要转变。2. 理解 subagents 与 AI 团队的基本概念2.1 什么是 Agent 和 Subagent先明确定义。Agent智能体指的是以大模型为决策核心具备任务规划、工具调用和结果反馈能力的一个执行单元。它可以读取输入、调用函数、查询数据库、写文件、发请求在有限的人工干预下完成一个相对完整的任务。Subagent子代理是相对“总控 Agent”而言的。当任务复杂度超过单个 Agent 的处理边界工程上倾向于把总目标拆成多个专职子代理每个子代理负责一个明确的环节。比如“生成一份竞品分析报告”可以拆成资料收集子代理、数据分析子代理、报告撰写子代理。每个子代理有自己的系统提示词、工具集和输出格式多个子代理协同完成总目标。Ad-hoc 的意思是“临时的、即兴的”。所以 ad-hoc subagents 翻译过来就是“临时拼装、用完即弃的子代理”——它们每次根据上下文被动态组合任务结束后没有持久化的团队级状态。2.2 为什么需要多个子代理协同你可能会问一个超强的大模型把所有任务都塞进一个 Prompt 不就行了为什么要拆成多个子代理答案有三个层面。第一是上下文窗口约束。即便模型支持超长上下文把完整业务流程的所有规则、历史状态、中间产物一股脑塞进去既容易超出窗口限制也会让模型注意力分散。每个 Agent 只处理自己的窄任务Prompt 更短规则更清晰输出更稳定。第二是专业化和工具绑定。资料收集 Agent 需要绑定搜索 API数据分析 Agent 需要绑定 SQL 执行器报告撰写 Agent 需要绑定 Markdown 渲染。如果把这些工具全部塞给一个 Agent工具选择本身就变成了一种复杂度。第三是编排与管控。拆成多个子代理之后每个子代理的执行结果可以被单独校验、单独重试、单独审计。这在生产环境里意味着可控的故障边界。2.3 从“临时调用”到“团队治理”的差异用一个表格来对比两种模式的实质差异能帮你快速定位自己现在处于哪个阶段维度临时 subagentsad-hoc持久化 AI 团队durable AI team任务状态进程内存随会话失效持久化存储可恢复、可断点续跑中间产物散落各处的临时文件统一的工作空间artifactsAgent 间上下文基本隔离需要手工传递共享状态结构化管理决策记录无或仅普通日志审计事件记录输入输出与理由故障恢复整体重跑从失败步骤继续运行治理能力无法回答事后问题可追溯、可复盘、可审批运行特征像跑一个脚本像一个有状态的服务理解了差异之后再看“Durable”和“Accountable”这两个词就不再是抽象口号了。3. Durable 与 Accountable 到底指什么3.1 Durable不是加一个数据库而是有状态的工作流很多人一看“持久化”第一反应是“那我给 Agent 加个 MySQL 存储不就行了”。这只说对了一半。Durable 的核心不是“有地方存数据”而是“任务可以在任何中断点被恢复并且恢复后能够正确继续”。这要求你为任务设计一个稳定的状态机。一个任务至少要经历 PENDING待执行、RUNNING执行中、SUCCEEDED成功、FAILED失败、NEEDS_REVIEW待人工复核这几个阶段。每一次状态迁移都要能被持久化。当一个 Agent 步骤执行到一半进程崩溃重启后系统能读取持久化状态判断哪些子任务已经完成哪些还需要继续而不是把整个流程从头跑一遍。打个比方传统脚本像一次性纸杯喝完就扔持久化 AI 团队像一个有抽屉的文件柜你打开哪个抽屉都能看到上次整理到哪一步了。缺少这层Agent 永远只能活在“当下”无法成为可积累的团队资产。3.2 Accountable不是加日志而是能回答“为什么”如果说 Durable 解决“做了什么”的连续性Accountable 解决的是“为什么这么干”的追溯性。普通日志通常只记录“某某 Agent 在什么时间调用什么接口、返回什么状态码”。但 Accountable 更关心的是语义层面的决策记录这个 Agent 基于什么理由选择了这个工具、为什么采纳了这条数据、为什么放弃另一个方案。记录这些信息之后当业务方问你“为什么系统把这篇内容放行了”你能从审计事件里找出一整条链条谁处理的、看到了什么输入、做出了什么判断、输出了什么结果。这种能力不是锦上添花而是生产系统的底线要求。3.3 两层要求如何共同降低生产风险把 Durable 和 Accountable 放在一起看你会发现它们互补。持久化状态让系统可以长周期运行、可恢复审计日志让系统可以被审查、被修复、被优化。如果没有持久化审计日志的上下文会断裂如果没有审计日志状态恢复后你无法判断之前那次失败到底是怎么发生的。两者同时具备AI 团队才真正具备“可运维”的基本盘。这也是标题里“durable, accountable AI teams”的关键含义一个能持续工作、且能解释自己工作的团队才值得被放进业务流程里。4. 架构设计把团队思维落到系统分层4.1 三层架构编排层、执行层、基础设施层一个可落地的 AI 团队离不开三层结构。编排层负责任务的生命周期管理。它把业务需求解析成任务规划子代理执行顺序决定状态如何流转失败后如何重试。你可以把它理解为“项目经理”只掌握任务清单不负责具体干活。执行层是真正的子代理集合。每个子代理负责一个专业环节联结大模型服务和工具。它读取共享工作空间中的输入产生结构化输出并把关键决策上报给审计模块。基础设施层是团队的“肌肉记忆”包括任务状态存储、审计日志存储、共享工件仓库、消息队列以及人工审批接口。这一层与业务逻辑无关但决定了整个团队能不能长期可靠运转。4.2 核心数据流以自动化报告为例想象一个“每日竞品分析报告”系统。编排层每天定时创建任务任务初始状态是 PENDING。随后资料收集子代理从共享工作空间读取关键词调用搜索工具把收集到的资料清单写入工作空间数据分析子代理读取该清单产出结构化摘要报告撰写子代理读取摘要生成 Markdown 报告。整个过程每一步都会触发状态迁移和审计事件。如果资料收集子代理失败编排层根据状态存储中的记录只重跑这一个步骤而不是把三个 Agent 全部重新执行。如果报告内容需要人工复核任务状态会进入 NEEDS_REVIEW等待业务人员确认后再完成闭环。这就是一个典型的 durable、accountable 团队工作流。4.3 不引入重框架也能起步一些开发者看了复杂的 Agent 框架可能担心要上 k8s、要搭消息队列。其实生产者视角下Durable 的核心可以先用单一任务表加 JSON 文件起步Accountable 的核心可以先用一个 JSONL 审计文件起步。只有当任务量、并发数、团队协作需求上升时再逐步替换为 PostgreSQL、Redis Streams 等更重型基础设施。早期框架越轻你能理解得越深。5. 环境准备与基础配置5.1 运行环境下面演示的代码不需要特别重的运行环境。一个装有 Python 3.10 以上版本的 Linux 或 macOS 环境即可Windows 在路径稍作调整也能运行。代码没有绑定某个具体的大模型服务商模型调用的位置会留出抽象接口你只需替换成你自己项目里实际的模型 SDK。存储部分示例使用本地 JSON 文件生产环境替换为 PostgreSQL 即可。版本方面请以你实际的项目环境为准本文重点是演示通用思路而不是绑定某个特定版本。5.2 项目目录与依赖先创建项目目录mkdir ai-team-demo cd ai-team-demo python3 -m venv venv source venv/bin/activate pip install pyyaml建议的目录结构ai-team-demo/ ├── config/ │ └── team.yaml ├── src/ │ └── ai_team/ │ ├── __init__.py │ ├── state.py │ ├── audit.py │ └── runner.py ├── examples/ │ └── run_research_team.py ├── data/ └── logs/依赖很少核心就是 PyYAML 用来解析团队配置。如果你不想引入配置解析也可以直接用 Python 字典写死团队结构但用 YAML 的好处是后续可以把团队组成放到配置中心统一管理。5.3 用 YAML 声明一个 AI 团队团队配置本质上是要回答三个问题团队有哪些成员子代理、每个成员做什么、团队如何提供状态和审计能力。下面是一份示例配置# 文件路径config/team.yaml team: name: research-team coordinator: orchestrator agents: - name: data_collector description: 资料收集负责搜索、抓取和初筛 tools: [search_api, web_crawler] max_retries: 3 - name: analyzer description: 数据分析负责从资料中提炼结构化结论 tools: [sql_runner, chart_builder] max_retries: 2 - name: writer description: 报告撰写负责生成最终文档 tools: [markdown_writer] max_retries: 1 state_store: type: local_json path: ./data/team_state.json audit: enabled: true path: ./logs/audit.jsonl这段配置把“团队人员结构”和“团队治理基础设施”固定下来。实际开发中无论你用什么 Agent 框架这种“先声明、后执行”的思路都值得保留。团队结构一旦可配置调参、加角色、改流程都不需要动核心代码。6. 核心实现把临时 subagents 变成持久化团队6.1 状态存储让任务不“失忆”第一个核心模块是状态存储。它的职责非常简单根据 task_id 读取任务状态写入新的状态。但正是这个简单模块解决了 ad-hoc 模式的第一个致命伤——进程重启后的失忆。生产环境请把它替换为数据库并加上事务在演示代码里我用 JSON 文件实现方便你完整跑通。# 文件路径src/ai_team/state.py 任务状态存储把 AI 团队的运行状态落到本地文件。 生产环境建议替换为 PostgreSQL / Redis并开启事务。 这里只演示最核心的状态读写逻辑。 import json from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict class TeamStateStore: def __init__(self, storage_path: str ./data/team_state.json): self.storage_path Path(storage_path) self.storage_path.parent.mkdir(parentsTrue, exist_okTrue) self._states: Dict[str, Any] {} self._load() def _load(self) - None: if self.storage_path.exists(): with self.storage_path.open(r, encodingutf-8) as f: self._states json.load(f) else: self._states {} def save(self) - None: with self.storage_path.open(w, encodingutf-8) as f: json.dump(self._states, f, ensure_asciiFalse, indent2) def get_task(self, task_id: str) - Dict[str, Any]: return self._states.get(task_id, {}) def update_task(self, task_id: str, patch: Dict[str, Any]) - None: task self._states.setdefault(task_id, {}) task.update(patch) task[updated_at] datetime.now(timezone.utc).isoformat() self.save()update_task的设计要点是“部分更新”。每次只修补发生变化的状态字段而不是整个任务对象重写这样可以减少并发写冲突的窗口。get_task返回空字典而不是报错目的在于容忍不存在的任务。生产环境中你还可以给_states字段加上锁或者直接依赖 PostgreSQL 的事务与行锁。6.2 审计日志让决策可追溯第二个核心模块是审计日志。我选择 JSONL 格式因为每行一个事件追加写入性能高读取时也可以逐行流式处理不会因为文件行数增加而明显变慢。事件中一定要记录 timestamp、task_id、agent_name、action、reasoning以及输入输出的摘要。注意“摘要”这两个字——如果直接把完整输入输出写进审计日志很快会被 token 量淹没而且可能泄露隐私。# 文件路径src/ai_team/audit.py 审计日志把 Agent 的决策过程记录下来用于追溯和复盘。 import json from datetime import datetime, timezone from pathlib import Path from typing import Any class AuditLogger: def __init__(self, log_path: str ./logs/audit.jsonl): self.log_path Path(log_path) self.log_path.parent.mkdir(parentsTrue, exist_okTrue) def log_decision( self, task_id: str, agent_name: str, action: str, reasoning: str, input_summary: Any None, output: Any None, ) - None: record { timestamp: datetime.now(timezone.utc).isoformat(), task_id: task_id, agent_name: agent_name, action: action, reasoning: reasoning, input_summary: input_summary, output: output, } with self.log_path.open(a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n)reasoning是审计日志里最重要、也最容易被遗漏的字段。很多团队记录 Agent 的调用参数记得很勤却从不记录 Agent“为什么选择这个工具”“为什么得出这个结论”。缺少 reasoning 的审计日志只是工具调用流水账无法支撑复盘。6.3 任务执行器支持断点恢复与失败兜底有了状态和审计还需要一个“任务执行器”把两者串起来。它负责创建任务、按步骤执行 Agent、更新状态、写入审计事件。最关键的是执行器需要支持断点恢复如果某个步骤已经成功恢复后应该跳过它而不是重跑整条流水线。# 文件路径src/ai_team/runner.py 任务执行器。 run_task 负责按顺序执行一组 Agent 步骤并实现断点恢复。 import uuid from typing import Any, Callable, Dict, List, Tuple from .audit import AuditLogger from .state import TeamStateStore AgentRunner Callable[[Dict[str, Any], str], Tuple[Dict[str, Any], List[Dict[str, Any]]]] class TeamRunner: def __init__(self, store: TeamStateStore, audit: AuditLogger): self.store store self.audit audit def create_task(self, task_name: str, input_data: Dict[str, Any]) - str: task_id input_data.get(task_id) or uuid.uuid4().hex[:12] self.store.update_task(task_id, { task_name: task_name, input: input_data, status: PENDING, current_step: PENDING, artifacts: {}, error: None, }) self.audit.log_decision( task_id, orchestrator, task_created, 任务创建完成进入待调度队列。, input_summaryinput_data, ) return task_id def run_step(self, task_id: str, agent_name: str, agent_func: AgentRunner) - None: task self.store.get_task(task_id) if not task: raise FileNotFoundError(f任务 {task_id} 不存在) self.store.update_task(task_id, { status: RUNNING, current_step: agent_name, }) try: result, decisions agent_func(task.get(artifacts, {}), agent_name) artifacts task.get(artifacts, {}) artifacts[agent_name] result for decision in decisions: self.audit.log_decision( task_id, agent_name, decision.get(action, agent_step), decision.get(reasoning, ), decision.get(input), decision.get(output), ) self.store.update_task(task_id, { status: SUCCEEDED, current_step: COMPLETED, artifacts: artifacts, }) except Exception as exc: self.store.update_task(task_id, { status: FAILED, error: str(exc), }) self.audit.log_decision( task_id, agent_name, step_failed, fAgent 执行失败: {exc}, ) raise RuntimeError(fAgent {agent_name} 执行失败: {exc}) from exc def run_task( self, task_id: str, agent_funcs: Dict[str, AgentRunner], steps: List[str], ) - None: 按顺序执行一组 Agent 步骤具备基础的断点恢复能力。 for agent_name in steps: task self.store.get_task(task_id) if task.get(artifacts, {}).get(agent_name) is not None: print(f[skip] {agent_name} 已完成跳过) continue self.run_step(task_id, agent_name, agent_funcs[agent_name])断点恢复的逻辑全部在run_task里每执行一个步骤前先检查共享工作空间中是否有该 Agent 的产物。有说明这一步已经成功执行过跳过。没有则执行。这与大型工作流引擎的“幂等步骤”思想完全一致。当然这个实现还比较朴素。如果 Agent 执行成功但审计写入失败恢复后会重复执行 Agent这一点需要靠更严谨的事务机制解决后面会展开说。6.4 组装示例跑一个三步骤的 AI 团队下面用一个最小示例把上面三个模块串起来。示例模拟一个“生成 AI Agent 工程化简报”的三步骤团队第一个 Agent 收集资料第二个 Agent 分析资料第三个 Agent 写简报。模型调用部分用一个占位函数你替换成实际 SDK 即可。# 文件路径examples/run_research_team.py 通过一个最小的 AI 团队演示创建任务 - 连续执行 - 断点恢复。 import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), ..)) from src.ai_team.state import TeamStateStore from src.ai_team.audit import AuditLogger from src.ai_team.runner import TeamRunner def call_model(system_prompt: str, user_content: str) - str: 模型调用占位函数。 请替换为你在用的模型服务 SDK这里不绑定具体厂商。 raise NotImplementedError(请接入你的模型服务后运行完整示例) def create_agent_func(system_prompt: str): def agent_func(artifacts, agent_name): # 根据当前步骤构造模型输入 if agent_name data_collector: user_content 收集与“AI Agent 工程化”相关的 5 条资料输出 JSON 列表。 elif agent_name analyzer: collected artifacts.get(data_collector, {}) user_content f分析以下资料输出 3 条关键结论。\n{collected} else: user_content 基于前面的分析结果生成一篇 300 字的技术简报。 decision call_model(system_prompt, user_content) # 这里把模型输出包装成结构化结果和决策记录 result {summary: decision, raw_output: decision[:200]} decisions [ { action: agent_name _done, reasoning: 完成本步骤的核心产出。, output: result, } ] return result, decisions return agent_func def main(): store TeamStateStore(./data/team_state.json) audit AuditLogger(./logs/audit.jsonl) runner TeamRunner(store, audit) task_id runner.create_task( 生成 AI Agent 工程化简报, {topic: AI Agent 工程化, output_language: zh}, ) print(created task:, task_id) # 实际使用时请替换 system_prompt agents { data_collector: create_agent_func(你是资料收集专家。), analyzer: create_agent_func(你是一个严谨的分析师。), writer: create_agent_func(你是技术文档编辑。), } steps [data_collector, analyzer, writer] # 第一次执行整条流水线 runner.run_task(task_id, agents, steps) # 模拟进程重启后从持久化状态恢复并再次执行 # 因为前面步骤已成功run_task 会跳过已完成步骤 store2 TeamStateStore(./data/team_state.json) runner2 TeamRunner(store2, audit) runner2.run_task(task_id, agents, steps) print(task finished, final state -, store2.get_task(task_id)) if __name__ __main__: main()这段示例展示了两个关键能力。第一多个子代理共享同一个artifacts上一个 Agent 的产出自动成为下一个 Agent 的输入。第二第二次执行时run_task会逐步骤检查产物并跳过已完成的部分。这正是 durable AI team 最基本的工程形态。6.5 为什么要先设计状态再写 Prompt代码放到这里有一个顺序问题值得强调。很多开发者接到“多 Agent 协作”需求第一反应是埋头写 Prompt 调模型写完再想怎么装配。但如果你想实现 durable 和 accountable开发顺序应该反过来先设计任务状态模型再设计审计事件结构最后才是写具体的 Agent Prompt。因为 Prompt 可以随时调架构一旦定了就很难回头。如果先有状态和审计你会发现后续增加 Agent、调整流程、排查问题都变得有章可循。7. 运行验证与效果检查7.1 运行步骤替换call_model为实际模型服务调用后运行python examples/run_research_team.py首次执行时你会看到类似输出created task: a1b2c3d4e5f6 [skip] data_collector 已完成跳过 [skip] analyzer 已完成跳过 [skip] writer 已完成跳过 task finished, final state - { ... }第二次执行时三个步骤全部被跳过因为状态存储里已经保存了产物。7.2 如何判断“持久化”生效判断持久化是否生效最直接的办法是模拟进程重启。你可以先运行一次示例观察data/team_state.json中出现了任务记录和artifacts字段然后再次运行代码观察控制台输出是否出现 “跳过” 提示。如果三个 Agent 步骤被跳过说明状态存储成功恢复了之前的执行结果。如果没有跳过、三步全部重跑说明团队没有持有“自身状态”仍然停留在临时 subagents 阶段。另一个验证手段是直接读取状态文件找到当前步骤。生产环境中建议在编排层增加一个“任务状态查询”接口让业务方随时知道任务进行到哪一步。7.3 如何判断“可问责”生效打开logs/audit.jsonl你应该能看到类似这样的事件结构。这里的关键不是日志是否有数据而是你能否根据一次业务异常还原出一整条决策链。例如用户投诉“某条文本被错误放行”你应能查到风险检测子代理当时的输入摘要、判断理由和输出结果。如果你发现审计记录里只有step_failed之类的事件看不到 Agent 的 reasoning就说明审计设计还不达标。7.4 失败排查的优先级运行示例如果失败先按下面顺序排查。第一模型调用占位函数是否抛出了NotImplementedError。没接实际模型服务后面所有步骤都不可能继续。第二确认路径是否以项目根目录运行。在src/ai_team目录下直接执行state.py不会报错但跨模块导入可能失败。第三检查data和logs目录是否生成了文件。如果生成了但内容为空再检查代码中save和log_decision是否真的被调用以及权限是否足够。8. 常见问题与排查思路问题现象可能原因排查方式解决方案任务状态一直停留在 RUNNING某个 Agent 执行超时或未正常返回查看 task 状态和 audit 日志给 Agent 增加超时控制超时后标记 FAILED 并重试进程崩溃后状态文件丢失状态存储路径不可写或未调用 save检查 data 目录权限与 update_task 是否被调用配置正确路径生产环境换数据库多个任务并发导致状态互相覆盖单一 JSON 文件没有并发控制复现并发场景查看状态文件换用 PostgreSQL/Redis并增加事务Agent 输出格式不稳定后面 Agent 解析失败Prompt 没有明确输出格式约束查看审计日志中的 raw_output在 System Prompt 中强约束 JSON Schema并增加解析容错审计日志为空audit.enabled 未开启或日志路径不可写检查 team.yaml 与 logs 目录权限开启审计修正路径断点恢复时 Agent 被重复执行状态持久化与执行成功之间没有强一致事务核对产品字段与日志时间戳引入幂等键或把产物写入和状态提交放进同一个事务同一任务重试后出现重复产物子代理步骤不具备幂等性观察 artifacts 中某 Agent 产物是否重复为每个产物增加 steptask 唯一标识重试前先删除旧产物9. 最佳实践与工程建议9.1 状态机的定义要尽量简单清晰前面已经提到了 PENDING、RUNNING、SUCCEEDED、FAILED、NEEDS_REVIEW 几个核心状态。实际生产中你可能会增加 STARTED、APPROVED、CANCELLED、ARCHIVED 等状态。但要注意状态越多状态流转越复杂出错的概率越高。建议先定义一张状态流转表明确规定每个状态可以迁往哪些状态不允许任意跳转。例如 FAILED 可以到 PENDING重试但 SUCCEEDED 不应回到 RUNNING。9.2 审计记录要坚持“摘要完整、明文脱敏”原则审计事件里既要记录充分的信息又要避免因记录完整原始内容引入隐私风险和存储成本。给大模型输入前先做摘要把核心事实和决策依据浓缩成结构化字段将敏感字段如手机号、邮箱、身份信息提前脱敏。从设计第一天就遵循这个原则远比后期补审计要轻松。9.3 人工审批要作为安全边界存在Accountable 并不代表所有决策都由模型自主完成。在有可能产生业务风险的环节例如放行高风险内容、执行删除操作、对外发消息一定要把任务状态置为 NEEDS_REVIEW等待责任人审批后在系统外确认完成。你可以在模型执行结果里附带 confidence 分数低于阈值自动进入人工审批这能在自主性和安全性之间取得平衡。9.4 成本和性能需要显式管理drable AI 团队运行时间长、涉及多个 Agent 调用token 消耗很容易失控。建议在每个任务创建时设定 token 预算Agent 执行前估算并记录预估用量执行后对比实际用量。同时在编排层加限流防止大量任务并发导致模型服务配额被瞬间打满。你甚至可以把 token 消耗作为审计事件的一个字段用来追踪“哪类 Agent 最费钱”。9.5 渐进式落地先窄后宽不追求一步到位如果你的业务还没到必须上重型框架的阶段先跑通一个最小团队即可两个 Agent、一个任务、一个 JSON 状态文件。先验证状态恢复和审计追溯能力再逐步增加 Agent、接入数据库、接入消息队列。不要一开始就把流程设计得特别复杂否则你很可能在第一天被基建问题淹没最后什么都跑不起来。10. 总结把 AI Agent 当系统来建设回到开头的那个 Show HN 标题。它其实提出了一个很清晰的方向subagents 不应该只是调用大模型的一次性脚本而应该被建设成有状态、可追溯、能恢复的团队系统。你可以把这套改造拆成四个不可省略的转变。第一从“会话内存”转变到“持久化状态”。任务要有生命周期状态要落盘进程重启后还能续跑。第二从“各自为战”转变到“共享工作空间”。多个子代理的中间产物要进入统一的工作空间并按任务 ID 组织。第三从“调用流水账”转变到“语义化审计日志”。日志不只记录调用参数还记录 Agent 的判断理由与输入输出摘要。第四从“全量重跑”转变到“断点恢复与审批闭环”。失败只重试失败步骤高风险输出必须经过人工审批。这套改动看起来没有引入什么黑科技但它把 Agent 从“一个聪明的临时工”变成了“一个可靠的项目成员”。如果你正在做 AI Agent 工程化下一步可以先画一张自己当前系统的状态流转图有哪些状态、哪些步骤、哪些黑盒环节然后把本文的最小状态存储和审计日志落到你的项目里跑一遍。你会发现等状态和审计成为系统默认设施之后再让多个基层 Agent 协作会顺畅得多。建议收藏备用等你真正接到生产级多 Agent 项目时再回来对照这份工程清单。