Agent系统的＂自动驾驶＂:ADAS自动设计 + Temporal持久编排 + 生产级状态管理

开篇：从"能跑"到"能上生产"的最后一公里

前两篇我们搞定了两件大事：

• Vol.1 讲清楚了什么时候拆、怎么拆、拆完怎么协调
• Vol.2 讲清楚了Agent 之间用什么方式通信（A2A、消息中间件、Actor 框架）

但到这一步，你手上的多 Agent 系统充其量叫"能跑"。离"能上生产"，还差三个关键拼图：

1. Agent 设计本身能不能自动化？ 每次新增一个 Agent 都要人手设计、调 Prompt、测效果，太慢了
2. Agent 执行到一半机器挂了怎么办？ 没有"断点续传"能力的系统不配叫生产级
3. Agent 的各种"记忆"存哪里？ 短期记忆、长期知识、日志审计，存储方案完全不同

今天这篇终章，我把这三块硬骨头啃完。

一、ADAS：让 Agent 自己设计 Agent

1.1 一个让人兴奋又恐惧的想法

我先问你一个问题：我们每次设计一个新 Agent，本质上在做什么？

无非是这几步：

1. 定义角色和 System Prompt
2. 选择绑定的工具集
3. 设计推理策略（CoT？ReAct？Reflexion？）
4. 跑测试、调参数、迭代优化

这个过程，本身能不能交给 AI 来做？

答案是：可以。 而且已经有人做出来了。

2024 年 8 月，Shengran Hu、Cong Lu 和 Jeff Clune 提出了 ADAS（Automated Design of Agentic Systems，自动化Agent设计系统）。这篇论文在 NeurIPS 2024 拿到了 Outstanding Paper 奖，并在 ICLR 2025 正式发表。

核心思想一句话：用一个 Meta Agent（元智能体）来自动设计、评估和迭代改进其他 Agent。

1.2 Meta Agent Search：Agent 的"进化引擎"

ADAS 的核心算法叫 Meta Agent Search（MAS），运作方式像一台"Agent 进化引擎"：

┌────────────────────────────────────────────────────────┐│                  Meta Agent Search 循环                  ││                                                        ││  第1代：Meta Agent 生成 Agent_v1（基础 CoT 策略）        ││    ↓ 评估：在验证集上跑分 → 得分 72%                      ││    ↓ 归档：存入 Agent 档案库                             ││                                                        ││  第2代：Meta Agent 参考档案库，生成 Agent_v2（加了反思）   ││    ↓ 评估：得分 78%                                     ││    ↓ 归档                                              ││                                                        ││  第3代：Meta Agent 发明新策略，生成 Agent_v3              ││    ↓ 评估：得分 85%                                     ││    ↓ 归档                                              ││                                                        ││  ...迭代 25-30 代...                                    ││                                                        ││  最终：从档案库中选出最优 Agent 投入使用                   │└────────────────────────────────────────────────────────┘

MAS 的关键设计：

1. 代码即 Agent：每个 Agent 被定义为一段 Python 代码（一个 forward 函数），而非简单的 Prompt。由于代码是图灵完备的，理论上能表达任意复杂的推理策略
2. 归档机制：每个成功的设计都会被存入档案库，供后续迭代参考。就像生物进化的"基因库"
3. 自我反思：生成后有两步 Reflexion 优化——先反思新颖性（避免重复），再反思正确性（修 Bug）
4. 自动调试：如果代码报错，最多自动修 5 轮

来看 MAS 的核心实现框架：

classLLMAgentBase:"""Meta Agent 的基础类——用大模型生成和评估 Agent 代码"""def__init__(self, output_fields: list, agent_name: str,                 role='helpful assistant', model='gpt-4o',                 temperature=0.5):self.output_fields = output_fieldsself.agent_name = agent_nameself.role = roleself.model = modelself.temperature = temperature  # 温度略高，鼓励"创造性"defgenerate_prompt(self, input_infos, instruction, output_description):# 构建系统 Prompt：角色 + JSON 格式约束        system_prompt = f"You are a {self.role}.\n\n" + \                        FORMAT_INST(output_description)# 用户 Prompt：输入信息 + 设计指令        prompt = build_input_text(input_infos) + instructionreturn system_prompt, promptdefquery(self, input_infos: list, instruction, output_description,              iteration_idx=-1):"""调用大模型，生成结构化的 Agent 设计方案"""        system_prompt, prompt = self.generate_prompt(            input_infos, instruction, output_description        )# 返回 JSON 格式的设计结果        response_json = get_json_response_from_gpt(            prompt, self.model, system_prompt, self.temperature        )return parse_output(response_json, iteration_idx)

MAS 的搜索主循环——进化式迭代：

defsearch(args, task):"""Meta Agent Search 主循环：迭代进化 Agent 设计"""    archive = task.get_init_archive()  # 初始档案：基础 Agent 模板for generation inrange(args.n_generation):  # 通常 25-30 代# 1. 基于档案库，生成新的 Agent 设计        next_solution = meta_agent.generate_new_design(archive)# 2. Reflexion 两步优化#    - 第一步：检查新颖性（是否跟档案库里已有的重复了？）#    - 第二步：检查正确性（代码能不能跑？逻辑对不对？）        next_solution = reflect_and_refine(            next_solution, task.get_reflexion_prompt()        )# 3. 在验证集上评估        accuracy_list = evaluate_forward_fn(            args, next_solution["code"], task        )# 4. 计算适应度分数（带置信区间）        next_solution['fitness'] = bootstrap_confidence_interval(            accuracy_list        )# 5. 归档        archive.append(next_solution)print(f"第{generation}代 | 得分: {next_solution['fitness']}")defevaluate_forward_fn(args, forward_str, task):"""动态加载并评估生成的 Agent 代码"""    namespace = {}exec(forward_str, globals(), namespace)  # 动态执行生成的代码    func = namespace['forward']              # 获取 forward 函数# 加载验证数据    data = task.load_data(SEARCHING_MODE)    task_queue = task.prepare_task_queue(data)# 并行评估（多线程加速）with ThreadPoolExecutor() as executor:        acc_list = list(executor.map(lambda item: score(func, item), task_queue        ))return acc_list

1.3 ADAS 的实际效果有多惊人？

论文的实验结果几乎让所有手写 Agent 的开发者"虎躯一震"：

更惊人的是：MAS 设计的 Agent 具有跨域迁移能力。 在 ARC 上进化出来的 Agent，不做任何改动直接拿去跑 MMLU，依然表现优异。甚至换模型（从 GPT-3.5 到 GPT-4）也照样好使。

这说明什么？MAS 发现的不是针对特定任务的"应试技巧"，而是更通用的推理策略。

1.4 ADAS 的现实应用与风险

目前 ADAS 还主要停留在学术研究阶段，但它给我们的启示非常深刻：

可以借鉴的思想（现在就能用）：

• 系统化的 Agent 评估流程：每个 Agent 设计都要在标准化的验证集上打分，不能靠"感觉"
• 归档和复用：把成功的 Prompt 策略、工具组合沉淀成模板库
• 自动化调试循环：Agent 输出不对？让另一个 Agent 自动分析原因并修复

需要警惕的风险：

• 安全性：MAS 通过 exec() 动态执行生成的代码，这在生产环境中是高风险操作
• 可控性：自动进化出来的 Agent 策略，人类不一定能理解其推理过程
• 对齐问题：自我进化的 Agent 是否始终符合人类意图？这是 AI Safety 的核心问题

💡 务实建议：不必追求完全自动化的 ADAS。但你完全可以借鉴 MAS 的"评估-生成-反思-归档"循环，手动执行每一步。这本质上就是Agent 工程化的最佳实践。

二、Temporal：多 Agent 系统的"断点续传"

2.1 没有持久编排的 Agent 系统有多脆弱？

来设想一个供应链 Agent 系统的典型流程：

1. 库存 Agent 检查库存告急 → 生成补货计划         ← 3秒2. 等待人工审批补货计划                           ← 可能等 2 小时3. 物流 Agent 安排发货                            ← 5秒4. 供应商 Agent 确认供货能力                       ← 调用外部API，可能超时5. 合规 Agent 审查文件                            ← 10秒6. 汇总结果，通知相关人员                          ← 2秒

如果第 4 步的外部 API 超时了怎么办？如果第 2 步等审批时机器重启了怎么办？

没有持久编排引擎，你的选择是：

• ❌ 从头跑一遍（浪费已完成的工作）
• ❌ 手动记录进度并恢复（维护噩梦）
• ❌ 相关方根本不知道流程走到哪一步了

这就是 Temporal 要解决的问题——把工作流的每一步状态持久化，失败自动重试，中断自动恢复。

2.2 Temporal 的核心概念

2.3 用 Temporal 编排供应链 Agent 流程（实战）

来把前面的 6 步流程用 Temporal 实现——每一步都带自动重试，整体支持"断点续传"：

from datetime import timedeltafrom temporalio import workflow, activityfrom temporalio.common import RetryPolicy# === 第一步：定义 Activities（每个 Agent 的执行逻辑） ===@activity.defnasyncdefinventory_activity(params: dict) -> dict:"""库存 Agent：检查库存、生成补货计划"""    operation = params["operation"]    messages = params["messages"]# 调用库存专家 Agent 的核心逻辑    result = inventory_agent.process(operation, messages)return {"messages": result["messages"], "plan": result.get("plan")}@activity.defnasyncdeftransportation_activity(params: dict) -> dict:"""物流 Agent：安排发货"""    result = transportation_agent.process(        params["operation"], params["messages"]    )return {"messages": result["messages"]}@activity.defnasyncdefsupplier_activity(params: dict) -> dict:"""供应商 Agent：确认供货能力"""    result = supplier_agent.process(        params["operation"], params["messages"]    )return {"messages": result["messages"]}# === 第二步：定义 Workflow（完整的编排流程） ===@workflow.defnclassSupplyChainWorkflow:"""供应链多 Agent 协作工作流——支持断点续传"""    @workflow.runasyncdefrun(self, operation: dict,                  initial_messages: list) -> dict:# 定义通用重试策略：最多重试 3 次        retry_policy = RetryPolicy(            maximum_attempts=3,            initial_interval=timedelta(seconds=2),  # 首次重试等 2 秒            backoff_coefficient=2.0# 每次等待时间翻倍        )# Step 1: 库存管理（带重试）# 如果这步成功后面的步骤失败了，# Temporal 不会重新执行这一步        inventory_result = await workflow.execute_activity(            inventory_activity,            {"operation": operation, "messages": initial_messages},            start_to_close_timeout=timedelta(seconds=30),            retry_policy=retry_policy        )# Step 2: 等待人工审批（可以等数小时甚至数天）# Temporal 会持久化等待状态，机器重启也不影响        approval = await workflow.wait_condition(lambda: self.approved isnotNone,            timeout=timedelta(hours=48)  # 最多等 48 小时        )# Step 3: 物流安排        updated_messages = (initial_messages +                           inventory_result["messages"])        transportation_result = await workflow.execute_activity(            transportation_activity,            {"operation": operation, "messages": updated_messages},            start_to_close_timeout=timedelta(seconds=30),            retry_policy=retry_policy        )# Step 4: 供应商确认（外部 API 可能不稳定）        final_messages = (updated_messages +                         transportation_result["messages"])        supplier_result = await workflow.execute_activity(            supplier_activity,            {"operation": operation, "messages": final_messages},            start_to_close_timeout=timedelta(seconds=60),  # 给多点时间            retry_policy=RetryPolicy(maximum_attempts=5)    # 多重试几次        )return {"inventory": inventory_result,"transportation": transportation_result,"supplier": supplier_result        }# Signal handler：接收人工审批结果    approved = None    @workflow.signalasyncdefapprove(self, decision: str):self.approved = decision

Temporal 在这里做了什么？

场景1：Step 3 执行时 LLM API 超时  → Temporal 自动重试（按 retry_policy），最多 3 次  → 重试期间不会重新执行已完成的 Step 1 和 Step 2场景2：Step 2 等待审批时，机器重启了  → Temporal 自动恢复 Workflow 状态  → 继续等待审批信号，无需任何人工干预场景3：想查看流程进度  → 通过 Temporal Query 实时查询：     "当前在 Step 4，供应商确认中，已重试 2 次"

2.4 Temporal 与 Agent 框架的集成

2025 年底，Temporal 宣布与 OpenAI Agents SDK 官方集成。这意味着你可以用 Temporal 编排 OpenAI Agent 的 Handoff 流程，同时获得持久化和容错能力。

💡 什么时候该上 Temporal？

• 流程涉及外部 API 调用（LLM、支付、第三方服务）→ 需要重试和容错
• 流程需要人工介入（审批、确认）→ 需要持久化等待
• 流程跨越较长时间（分钟/小时/天）→ 需要状态持久化
• 系统需要审计追溯→ 需要完整事件历史

如果你的 Agent 流程是秒级完成的简单查询，不需要 Temporal，杀鸡别用牛刀。

2.5 国内的工作流编排替代方案

Temporal 是全球范围内的标杆，但国内也有一些值得关注的选项：

三、状态管理：Agent 的"记忆系统"存哪里？

3.1 三种"记忆"，三种方案

Agent 的状态管理不是一个单一问题。根据记忆的生命周期和用途，需要完全不同的存储方案：

3.2 存储方案选型

3.3 实战组合推荐

一个完整的生产级多 Agent 系统，通常需要混合使用多种存储：

┌──────────────────────────────────────────────────────────┐│                  多 Agent 状态管理架构                      ││                                                          ││  ┌──────────────────────────┐                            ││  │     情景记忆（短期）        │                            ││  │  Redis / 内存              │ ← 当前对话上下文、缓存        ││  │  TTL: 30分钟 ~ 24小时      │                            ││  └──────────────────────────┘                            ││                                                          ││  ┌──────────────────────────┐                            ││  │     语义记忆（长期）        │                            ││  │  向量数据库 + PostgreSQL   │ ← 知识库、用户画像、RAG       ││  │  生命周期: 永久            │                            ││  └──────────────────────────┘                            ││                                                          ││  ┌──────────────────────────┐                            ││  │     流程记忆（持久化）       │                            ││  │  Temporal / PostgreSQL    │ ← 工作流进度、审计日志         ││  │  生命周期: 随工作流         │                            ││  └──────────────────────────┘                            ││                                                          ││  ┌──────────────────────────┐                            ││  │     归档存储              │                            ││  │  OSS / S3 / COS          │ ← 历史日志、大文件、数据备份    ││  │  生命周期: 按保留策略       │                            ││  └──────────────────────────┘                            │└──────────────────────────────────────────────────────────┘

国内具体选型建议：

💡 一个常见的坑：很多团队把所有状态都塞进向量数据库，觉得"什么都能语义搜索"很酷。但实际上，结构化数据用关系型数据库查询效率高 10 倍以上，而且更可靠。向量数据库只用于真正需要语义搜索的场景（RAG 检索、相似文档匹配），不要滥用。

四、终极选型：多 Agent 系统技术栈全景图

写到这里，三篇文章覆盖的所有技术栈汇总成一张全景速查表：

五、全系列总结：从一个 Agent 到多 Agent 的进化路径

三篇文章，我们完成了一次从单 Agent 到生产级多 Agent 系统的完整旅程：

Vol.1 → 架构设计

• 单 Agent 够用时别拆，拆的判断标准是工具数和业务域
• 拆分遵循六大原则，经理制（Supervisor）先用起来
• 四种协调策略各有千秋，按场景选

Vol.2 → 通信基础设施

• A2A 和 MCP 互补，一个管 Agent 互通，一个管工具连接
• 消息中间件是异步解耦的标配，Redis Stream 起步、Kafka 兜底
• Actor 框架在有状态分布式场景是刚需

Vol.3 → 生产级能力

• ADAS 展示了 Agent 自动设计的可能性，MAS 思想值得借鉴
• Temporal 解决了"断点续传"这个上生产的关键问题
• 状态管理要分层——不同类型的记忆用不同的存储

最后送你一句话：

多 Agent 系统的真正挑战不是"把 Agent 跑起来"，而是"让 Agent 持续可靠地跑下去"。

架构设计决定能不能拆，通信基础设施决定拆完能不能协作，持久编排和状态管理决定能不能上生产。三层能力缺一不可。

2025 是 Agent 元年，2026 是多 Agent 元年。基础设施和最佳实践还在快速演进，但底层的架构思想——分工、协调、通信、容错——是跨越技术迭代的永恒命题。

掌握了这些，无论框架和模型怎么变，你都能从容应对。

全系列三篇完结。如果这个系列对你有帮助，帮忙转发一下，让更多正在搭建 Agent 系统的同学少踩点坑。有任何问题欢迎后台留言。

下一个系列，我们聊点更硬核的——Agent 的评估与测试体系：怎么量化一个 Agent"好不好用"？敬请期待。