Skip to content

多 Agent、Memory 与 Loop

主题定位| 本文不把“角色多、消息多、循环多”当成高级 Agent。真正的专题主线是:如何划分协作责任,如何让事实跨步骤和跨会话安全流动,以及如何用确定性 Runtime 让嵌套循环最终收敛。

目录

1. 学习目标

学完后应能做到:

  • 用一句话区分多 Agent、角色扮演、并行工具调用和普通 Workflow;
  • 区分 StateContextCheckpointEvent Log、短期 Memory、长期 Memory 与知识库;
  • 解释单 Agent 内循环、多 Agent 调度外循环和 Memory 整理循环如何嵌套;
  • 设计 Manager-as-Tool、Handoff、Group Chat、Blackboard 与 Map-Reduce 等协作拓扑;
  • 用 Schema、作用域、版本、TTL、来源和删除策略管理 Memory;
  • 用步数、时间、Token、费用、重复动作和结果验收让 Loop 收敛;
  • 从 Trace、Memory 命中和最终状态定位循环风暴、上下文丢失、记忆污染和重复副作用。

2. 面试结论

2.1 30 秒回答

多 Agent 本质上是把一个目标拆给多个具有独立上下文、能力或权限边界的 Agent,并通过明确协议协调控制权与产物;Memory 是对可复用信息执行选择、写入、检索、注入、更新和删除的生命周期,而不等于聊天历史;Loop 是 Runtime 围绕“决策、执行、观察、状态更新、终止判断”运行的受控状态机。三者组合后能处理可并行、跨专业或长时任务,但也会增加上下文损失、冲突、污染、延迟与成本,所以必须先用单 Agent 或固定 Workflow 建立基线,再用端到端结果和故障测试证明拆分价值。

2.2 一分钟复述版

我会把系统拆成三层。第一层是协作拓扑:谁分解任务、谁拥有最终答案、控制权是委派后回收还是直接 Handoff;第二层是信息治理:当前执行事实进 State 和 Checkpoint,可复用经验经过来源、作用域、TTL 和冲突校验后才进入 Memory;第三层是循环控制:模型只提出候选动作,Runtime 负责工具执行、预算、幂等、重试、暂停和终止。

项目中我不会因为任务“看起来复杂”就直接上多 Agent。先跑同模型、同工具、同预算的单 Agent 基线;只有任务确实存在可并行子问题、上下文隔离、专门权限或独立验证需求,而且质量或墙钟时间收益覆盖协调成本时,才引入多 Agent。验收同时看最终任务成功率、证据覆盖、冲突率、Handoff 信息损失、Memory 误注入、循环重复率、延迟和费用。

面试停止点| 第一轮说清“协作拓扑、信息生命周期、受控状态机”后停止;被追问时再展开具体模式、Schema 和故障闭环。

3. 面试官为什么问

这组题通常不是在考框架 API,而是在判断候选人能否处理分布式系统与概率模型叠加后的复杂性:

考察维度面试官真正关心的问题低质量回答信号
概念边界多个 Prompt 是否就算多 Agent只按“角色数量”定义
架构能力控制权、上下文和责任如何流动罗列框架名,不画数据与控制边界
状态治理重启、并发、版本冲突后怎样恢复把 Context 当数据库
可靠性循环、重试和副作用怎样收敛只在 Prompt 写“不要死循环”
安全性Memory 是否会跨用户泄漏或被污染把所有历史直接向量化
评测意识多 Agent 是否真的优于单 Agent用 Demo 流畅度代替受控基准
项目表达是否有 Trace、Schema、失败样本和验收虚构准确率、效率或线上事故

中高级回答的区分度在于:不仅能描述“几个 Agent 怎么聊天”,还要说明谁拥有最终责任、哪些信息可以共享、每层循环由谁终止,以及失败后如何恢复到确定状态。

4. 概念与边界

4.0 小白先这样理解:一支跨科室急诊会诊队

深夜里,一名病人同时出现胸痛、呼吸困难和药物过敏史。值班医生没有把所有人叫来围着病床自由讨论,而是先做分诊:心内科查看心电图,呼吸科分析影像,药师核对禁忌;每个人把结论写进同一份结构化会诊单,主治医生对冲突证据复核后决定治疗。病历保存经确认的过敏史,但不会把每次随口猜测都写成永久事实;如果检查结果迟迟不回,系统会提醒、升级或停止等待,而不是让医生无限讨论。

对应关系如下:

生活场景技术对象
主治医生分诊和收口Orchestrator / Manager Agent
心内科、呼吸科、药师具有不同上下文、工具或权限的 Worker Agent
会诊单结构化 Task / Result Envelope 与共享 Artifact
当前生命体征和检查进度Run State 与 Checkpoint
经确认的过敏史有来源、作用域和生命周期的长期 Memory
检查、回报、再决策Agent Loop 中的 Action、Observation 与状态更新
超时升级、停止等待Runtime 的超时、预算、暂停与终止策略

回到专业机制:多 Agent 的价值不是“多几个人说话”,而是将目标、上下文、权限和验证责任分成可管理边界;Memory 的价值不是“记得越多越好”,而是在正确作用域内召回可信且相关的信息;Loop 的价值不是“持续思考”,而是让每次外部观察都推动状态向可验收终点变化。

类比边界: 医生具有稳定身份、专业训练和现实责任,LLM Agent 没有。多个 Agent 也可能共享同一个基础模型,因此“不同角色”不等于真正独立的知识来源;会诊结论还需要真实工具、数据、权限与人工负责,不能靠多数投票把同源幻觉变成事实。

图:教学插图|多 Agent、Memory 与受控 Loop 的空间关系

替代文本: 深色任务控制室中,中央 Orchestrator 把一个目标分派给三个具有独立上下文、工具入口和局部循环的 Worker;三个 Worker 产生不同颜色的结构化 Artifact,并统一汇聚到右侧 Verifier。画面下方以工作托盘、时间切片保险库和带访问锁的长期档案柜区分 Working State、Checkpoint 与 Long-term Memory;外围的锁、预算仪表、计时器和停止闸门表示 Runtime 对权限、成本、超时和终止的约束。

多 Agent、Memory 与受控 Loop 的任务控制室教学插图

读图结论: 多 Agent 的重点不是让三个角色自由聊天,而是由一个责任中心分派任务、让局部 Loop 在隔离边界内产生 Artifact,再经过统一验证;Memory 提供分层状态支撑,Runtime 护栏保证系统不会无界执行。

这幅教学图用于建立组件的空间直觉:上层表现“谁负总责、谁并行执行、谁统一验证”,下层表现“当前工作、恢复快照和长期复用信息不应混成一层”。精确的 Memory 读取、验证后写入、冲突重试与降级时序仍以第 6.5 节技术调用流程图为准,生成图片不替代接口、方向和失败分支的工程证据。

4.1 什么是多 Agent

本文把 Multi-Agent System 定义为:至少两个可独立运行决策循环的 Agent 实例,具有可区分的指令、上下文、工具、权限或状态边界,并通过显式消息、共享 Artifact 或 Handoff 协议共同完成一个目标。

这个定义强调三个条件:

  1. 独立决策边界:每个 Agent 至少能根据自己的输入选择动作;
  2. 协作协议:任务、结果、控制权和失败必须可传递;
  3. 共同验收:最终由某个责任主体对系统目标收口。

以下情况不自动构成多 Agent:

  • 同一个 Agent 连续使用多个工具;
  • 同一模型一次生成“产品、研发、测试”三段角色意见;
  • 固定 DAG 中多个普通函数并行执行;
  • 为增加答案长度,让多个相同 Prompt 重复采样后简单投票。

这些做法可能有用,但没有独立 Agent Loop 或明确协作边界时,更准确的名称是并行工具、角色化 Prompt、Workflow 节点或 Self-Consistency。

4.2 多 Agent 的五种常见拓扑

拓扑控制权上下文流动适合场景核心风险
Manager-as-ToolManager 始终负责最终答案Worker 接收裁剪后的任务,返回结构化结果可分解、需并行、责任要集中Manager 成为瓶颈,摘要丢证据
Handoff / Swarm当前 Agent 把主导权移交给下一个 Agent接收方获得全部或过滤后的历史客服分流、阶段性交接责任漂移、输入过滤错误
Group Chat调度器或模型选择发言者共享消息历史协商、辩论、创意探索Token 爆炸、重复发言、群体偏差
BlackboardAgent 读写共享任务板或 Artifact共享事实与中间产物,不必共享全部对话工程任务、研究证据汇总并发覆盖、陈旧读取、所有权不清
Map-Reduce代码分片并行,Reducer 聚合每个 Worker 只看分片和契约文档批处理、独立样本分析分片遗漏、聚合丢少数证据

最稳妥的默认不是自由群聊,而是确定性 Orchestrator + 有界 Worker + 结构化 Artifact + 独立 Verifier。只有控制权确实应移交、参与者需要互相回应时,才使用 Handoff 或 Group Chat。

4.3 Memory 到底是什么

Memory 不是某一种数据库,而是一条信息生命周期:

text
候选信息产生 -> 可信度与敏感性判断 -> 作用域和类型分类
-> 写入或合并 -> 按任务检索 -> 过滤与排序 -> 注入 Context
-> 使用结果反馈 -> 更新、过期、纠错或删除

从用途看,可以分为:

Memory 类型典型内容生命周期典型查询
Working Memory当前目标、计划、未解决问题单次 Run下一步还缺什么
Episodic Memory某次任务轨迹、失败与结果跨 Run,可过期上次类似事故怎么处理
Semantic Memory经验证事实、偏好、实体关系跨会话,需纠错用户确认的语言偏好是什么
Procedural Memory操作规则、Runbook、技能步骤版本化长期保存这个仓库怎样构建和验收

在工程上,还必须按作用域划分:

  • agent_private:单个 Agent 的私有草稿或专门经验;
  • run_shared:同一 Run 内所有参与者共享的已确认事实;
  • user / tenant:跨会话复用,但必须身份和租户隔离;
  • global:全局规则或公共知识,写入权限最严格。

一个 Memory Item 至少应带有:

json
{
  "memory_id": "mem_01",
  "scope": {"tenant_id": "t1", "user_id": "u7", "agent_id": "investigator"},
  "kind": "episodic",
  "content": "数据库连接池耗尽时先核对等待队列和慢查询",
  "source": {"run_id": "run_42", "artifact_id": "trace_9"},
  "confidence": 0.86,
  "status": "verified",
  "created_at": "2026-07-14T10:00:00+08:00",
  "expires_at": "2026-10-14T10:00:00+08:00",
  "version": 3,
  "supersedes": "mem_00",
  "acl": ["role:sre"]
}

这里的 confidence 不是模型说“我很确定”,而应来自来源等级、人工确认、工具证据和后续验证。敏感数据还要支持访问审计、更正、导出与删除。

4.4 State、Context、Checkpoint、Log、Memory 和知识库

概念核心问题是否直接给模型是否用于恢复典型边界
State系统现在处于什么确定状态只选必要字段结构化、版本化、可校验
Context模型这一轮能看到什么受窗口、相关性和安全约束
Checkpoint某个时刻怎样恢复执行通常不直接全量注入run_id、步骤和版本绑定
Event Log到底发生过什么按需摘要或查询可重建 State追加写、审计、不可静默改写
Memory哪些信息值得以后复用检索后选择性注入间接有来源、作用域、TTL、纠错和删除
Knowledge Base组织认可的外部知识是什么经 RAG 注入内容治理、ACL、版本、引用

关键边界是:Context 是视图,不是事实库;Checkpoint 是恢复材料,不等于长期 Memory;完整轨迹是审计证据,不应无界塞入模型;知识库是外部受治理知识,不应被 Agent 随手改写。

4.5 Loop 是什么

Agent Loop 可以抽象为:

atπθ(Context(g,st,mt,o<t))ot=Execute(at),st+1=T(st,at,ot)continuet+1=¬Stop(g,st+1,B,R)

其中:

  • g 是目标和验收条件;
  • s_t 是可持久化 State;
  • m_t 是检索并通过过滤的 Memory;
  • a_t 是模型提出的候选动作;
  • o_t 是工具结果、拒绝、超时或错误等 Observation;
  • T 是确定性的状态转移;
  • B 是步数、时间、Token、费用和并发预算;
  • R 是权限、风险、幂等和人工审批规则。

模型可以建议“完成”,但 Runtime 还要验证输出 Schema、目标断言和未决任务;模型可以建议“重试”,但 Runtime 要判断错误类别、副作用状态和剩余预算。

4.6 三类嵌套 Loop

完整系统通常不止一个循环:

  1. Worker 内循环:单个 Agent 决策、调用工具、接收 Observation;
  2. Orchestrator 外循环:分解任务、调度 Worker、合并冲突、决定是否追加工作;
  3. Memory 整理循环:从已完成轨迹提取候选记忆,验证、去重、过期和纠错。

三层不能共用一个含糊的 max_steps。应分配嵌套预算,例如:

text
run_budget = 60 秒 / 30k Token / 0.50 美元
orchestrator_rounds <= 4
worker_calls <= 6
worker_steps_each <= 5
memory_reads <= 8
memory_writes <= 2,且只在最终验收后提交

具体数字只是示例,必须通过任务基准和成本约束确定。重要的是预算要向下分配、向上汇总,子 Agent 不能绕过总预算。

5. 三个机制如何组成一个系统

5.1 从用户目标到任务契约

Orchestrator 首先不是“想办法”,而是把目标转为可验收的任务契约:

json
{
  "task_id": "task_db_001",
  "goal": "确认连接池告警的主因并给出证据",
  "inputs": ["metrics://pool", "logs://api"],
  "allowed_tools": ["query_metrics", "search_logs"],
  "output_schema": {
    "hypothesis": "string",
    "evidence_refs": ["string"],
    "confidence": "number",
    "unknowns": ["string"]
  },
  "deadline_ms": 12000,
  "max_steps": 4,
  "write_scope": "none",
  "idempotency_key": "run42:task_db_001"
}

契约让 Worker 知道任务边界,也让 Runtime 能在模型输出之外做确定性校验。没有 Schema、截止时间和权限的“帮我调查一下”会把协调风险推给模型自由发挥。

5.2 Context 传递应最小化

多 Agent 的上下文传递有三种常见策略:

  • 全历史转发:实现简单,但 Token、隐私和噪声成本高;
  • 摘要转发:成本低,但可能丢失来源、否定词和失败细节;
  • Artifact 引用:传结构化摘要和稳定 artifact_id,需要时再按权限读取原文。

生产系统优先使用“任务契约 + 结构化结果 + Artifact 引用”。摘要必须保留来源、时间、版本、不确定性和被排除假设,不能只保留一段自然语言结论。

5.3 Memory 的读路径

一次 Memory 读取建议按以下顺序:

  1. 用租户、用户、Agent、任务类型和时间做硬过滤;
  2. 用关键词、向量或结构化条件召回候选;
  3. 按相关性、可信度、时效性和注入成本排序;
  4. 去重、处理冲突和已废弃版本;
  5. 只把必要片段连同来源和时间注入 Context;
  6. 记录哪些 Memory 被使用、是否帮助结果。

可以使用启发式评分:

score(m,q)=wrRel(m,q)+wtTrust(m)+wfFresh(m)wcCost(m)wsRisk(m)

这不是通用标准公式。权重必须在固定任务集上校准;ACL、撤权和显式过期属于硬过滤,不能被高相关分抵消。

5.4 Memory 的写路径

Memory 写入比读取更危险。推荐把写入分成两阶段:

  1. 候选阶段:Agent 提交 memory_candidate,包含来源、作用域、置信依据和过期建议;
  2. 提交阶段:确定性规则、Verifier 或人工检查后,执行新增、合并、取代或拒绝。

以下内容默认不进入长期 Memory:

  • 模型未验证的推测;
  • 工具执行失败后的临时错误文本;
  • Secret、Token、完整隐私数据;
  • 只对当前 Run 有用的计划和草稿;
  • 与已有权威规则冲突但没有新证据的内容。

5.5 冲突与一致性

多 Agent 同时写共享 State 或 Artifact 时,应避免“最后写入者获胜”的静默覆盖。可选策略包括:

  • 追加写 Event,Reducer 按规则生成 Materialized State;
  • 使用 version 做乐观并发控制,版本不匹配则重新读取和合并;
  • 为任务、文件或实体指定唯一 Owner;
  • 对可交换结果使用集合并、计数器或其他确定性 Reducer;
  • 对不可自动合并的业务结论进入冲突队列或人工复核。

模型不能单独决定哪个冲突事实是真相。应比较来源等级、时间、工具证据和权威系统,必要时保留“当前无法确认”。

5.6 终止与收敛

Runtime 至少检查以下终止类别:

类别例子处理
成功终止所有必需 Artifact 已产生且验收通过返回结果、证据和终止原因
预算终止步数、时间、Token、费用或并发超限返回部分结果和未完成项
无进展终止动作指纹和 Observation 重复停止重试,进入降级或人工
安全暂停高风险工具等待批准持久化 Checkpoint,等待恢复
不可恢复失败权限拒绝、输入缺失、外部状态不一致记录证据,禁止盲重试
用户取消用户或上游撤销 Run传播取消,清理租约和待执行任务

“某个 Agent 输出 DONE”只能是软信号。最终完成必须由 Orchestrator 对目标断言、结果 Schema、证据引用和未决任务做验收。

6. 技术栈、实现选择与双图

6.1 概念与框架无关

多 Agent、Memory 与 Loop 是架构机制,不绑定 OpenAI Agents SDK、LangGraph、AutoGen 或其他框架。本文采用方案是“确定性 Orchestrator + 有界 Worker + Checkpoint/Store 分层 + 结构化 Artifact + 全链路 Trace”,用于说明生产参考设计,不代表所有项目都应选同一框架。

最小验证栈可以只用 Python 标准库、SQLite/JSONL 和伪 Agent;生产栈再按任务复杂度、恢复要求、团队语言、托管边界和合规需求选择框架、数据库、队列与可观测系统。

6.2 技术清单

技术点 ID技术点/环节类型采用方案链路职责版本/证据边界
TP-MAL-TOPOLOGY多 Agent 协作拓扑框架无关机制 + 编排实现确定性 Orchestrator + Agent-as-Tool + 独立 Verifier分解、并行、回收结果、冲突复核和最终验收参考架构;OpenAI Handoff、AutoGen Teams 等能力以 2026-07-14 官方文档为边界
TP-MAL-MEMORYState 与 Memory 持久化存储/状态管理Thread Checkpoint + Scoped Store + 追加式 Event Log恢复当前 Run、跨 Run 复用已验证信息、保留审计证据LangGraph 官方区分 Checkpointer 与 Store;具体数据库未在本仓库实测
TP-MAL-LOOP循环与终止控制Runtime/状态机确定性状态机 + 分层预算 + 无进展检测执行动作、更新状态、控制重试、暂停、恢复和终止OpenAI Agents SDK Runner Loop 可验证 Loop 与 max_turns;本文硬终止策略是工程参考
TP-MAL-PROTOCOLAgent 间通信与 ArtifactSchema/协议Task/Result Envelope + Artifact ID + 乐观版本最小化上下文、校验输入输出、避免静默覆盖本文示例 Schema;需按真实业务字段、权限和兼容策略验证
TP-MAL-OBSERVABILITYTrace、评测与回放可观测组件Run/Agent/Step Span + Event Log + 固定任务集重放定位路由、记忆、循环、工具与合并阶段的首次失败点框架通常提供部分 Trace;端到端字段和指标仍需应用侧补齐

6.3 横向选型对比

技术点 ID候选方案优点缺点/代价适用场景不适用场景选择结论与依据
TP-MAL-TOPOLOGYManager-as-Tool最终责任集中;Worker 可并行且上下文隔离Manager 可能成为瓶颈;任务摘要可能丢信息可分解研究、工程取证、多个只读专家需要连续直接服务用户的阶段交接默认基线;用并行收益、证据覆盖和冲突率验证
TP-MAL-TOPOLOGYHandoff / Swarm专家接管后交互自然;路由清晰时 Prompt 简洁控制权和历史传递复杂;责任易漂移客服分流、语言或权限域切换最终责任必须始终由一个控制器持有仅在确需转移主导权时使用,并测试输入过滤和回退
TP-MAL-TOPOLOGYGroup Chat / Selector支持多轮协商、批评和动态发言顺序Token 与协调成本高;易重复和群体偏差开放讨论、方案评审、红蓝对抗可被固定并行分解的独立任务先证明单 Agent/Manager 不足,再限定发言轮次和终止条件
TP-MAL-MEMORYCheckpoint + Scoped Store明确区分线程恢复与跨线程信息;易做 TTL、ACL需要维护两套模型和一致性长会话、HITL、跨会话偏好和经验一次性短任务生产参考方案;按作用域、来源和删除要求建模
TP-MAL-MEMORY仅追加全量消息历史实现最简单;审计原始对话方便无界增长、噪声和隐私风险;不等于可检索 Memory原型、小型短对话长任务、多租户、敏感信息仅作短期原型,不作为长期 Memory 设计
TP-MAL-MEMORYEvent Sourcing + Materialized View审计和重放能力强;并发合并更明确存储、Schema 演进和运维复杂高审计、复杂长任务、需要时间旅行低风险简单聊天达到恢复和审计要求后采用;否则 Checkpoint + Store 更轻量
TP-MAL-LOOP手写 FSM / Workflow控制和测试最透明;高风险分支可确定开发量大;开放任务灵活性较低核心流程稳定、高副作用、强合规路径高度开放且难枚举高风险控制面优先,模型只进入有限决策节点
TP-MAL-LOOP通用 Runner Loop工具、Handoff、Guardrail、Trace 集成快框架语义和升级约束;仍需应用侧验收快速构建有界 Agent需要非常规调度或跨语言 Runtime原型和常规任务可用,必须补硬预算与结果断言
TP-MAL-LOOP显式状态图 / Durable Workflow暂停恢复、可视化和节点测试更强图和 State Schema 的治理成本更高长任务、HITL、可恢复流程极短、无状态请求当恢复、人工审批和分支审计成为硬需求时采用
TP-MAL-PROTOCOL自然语言消息灵活、启动快难校验、易丢字段和产生提示注入创意讨论、低风险探索需要稳定自动聚合和审计只用于解释性内容,关键控制字段不得依赖自然语言
TP-MAL-PROTOCOL结构化 Envelope + Artifact可校验、可版本化、便于最小权限和重放需要 Schema 设计与兼容治理工程任务、数据处理、生产协作完全开放且无法定义任何输出契约默认采用;自然语言说明作为受约束字段存在
TP-MAL-OBSERVABILITY仅应用日志成本低、接入快难还原跨 Agent 因果与 Token/工具层级单服务原型多 Agent 并发与长任务排障只做起步,不足以证明首次失败层
TP-MAL-OBSERVABILITYTrace + Event + 评测集可按 Run/Agent/Step 回放和分桶评测存储、脱敏和采样成本生产系统、复杂路由、Memory 和 Loop 优化没有隐私治理能力的场景生产默认;先定义字段、保留期和脱敏,再扩大采集

6.4 架构图

图:架构|多 Agent、Memory 与 Loop 的控制面和数据面

替代文本: 用户目标进入 Orchestrator;Orchestrator 从 Checkpoint 恢复 Run State、从 Scoped Memory Store 检索必要记忆,并通过任务队列把结构化任务分给多个隔离 Worker;Worker 只经工具网关访问外部系统并把 Artifact 写入存储;Verifier 检查证据与冲突,Runtime 根据预算和终止策略决定完成、追加任务或人工介入;Event Log 与 Trace 覆盖所有组件。

图表加载中…

读图结论: Orchestrator 持有总目标和最终责任,Worker 只在任务契约与最小权限内行动;Checkpoint 解决“从哪里恢复”,Memory Store 解决“以后复用什么”,Artifact 解决“协作交付什么”,Trace 解决“失败首先发生在哪里”。

这幅架构图刻意把控制面与数据面分开。模型可以存在于 Orchestrator、Worker 或 Verifier 中,但权限、预算、状态提交、Artifact 版本和最终终止仍由确定性组件掌握。

6.5 技术调用流程图

图:技术调用流程|多 Agent 调度、Memory 注入与失败收敛

替代文本: Orchestrator 恢复 Checkpoint 并查询作用域正确的 Memory,生成带预算和权限的任务;两个 Worker 并行执行只读工具并返回 Artifact;Verifier 检查 Schema、来源和冲突;通过时提交最终 State 并异步产生 Memory 候选,冲突时追加一次定向任务,超时或预算不足时返回部分结果并进入降级或人工处理。

图表加载中…

读图结论: Memory 只在身份和作用域校验后进入本轮,Worker 的失败先变成可分类 Observation,长期 Memory 只在最终验收后产生候选;冲突和超时都有有限分支,不会无限拉起新 Agent。

调用流程图显示了架构图没有表达的时间关系:先恢复 State,再加载 Memory;先校验任务预算,再并行执行;先验证 Artifact,再提交最终状态;Memory 写入被放在成功验收之后,避免把中间推测永久化。

7. 最小可运行实现

7.1 运行环境与目标

下面代码只使用 Python 3.11+ 标准库,模拟两个 Worker、一个共享 State 和一个 Orchestrator。它不调用真实 LLM,目的是验证三个不变量:

  1. 子任务有独立预算;
  2. 重复动作会被 Runtime 终止;
  3. 只有通过 Verifier 的结果才能写入长期 Memory。
python
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Callable


@dataclass(frozen=True)
class Task:
    task_id: str
    question: str
    max_steps: int = 3


@dataclass
class Result:
    task_id: str
    answer: str
    evidence: list[str]
    status: str


@dataclass
class RunState:
    run_id: str
    steps: int = 0
    action_fingerprints: set[str] = field(default_factory=set)
    artifacts: dict[str, Result] = field(default_factory=dict)
    memory_candidates: list[dict] = field(default_factory=list)


Worker = Callable[[Task, int], tuple[str, list[str]]]


def run_worker(task: Task, worker: Worker, state: RunState) -> Result:
    for local_step in range(1, task.max_steps + 1):
        state.steps += 1
        answer, evidence = worker(task, local_step)
        fingerprint = f"{task.task_id}:{answer}:{','.join(evidence)}"

        if fingerprint in state.action_fingerprints:
            return Result(task.task_id, answer, evidence, "NO_PROGRESS")
        state.action_fingerprints.add(fingerprint)

        if answer and evidence:
            return Result(task.task_id, answer, evidence, "SUCCEEDED")

    return Result(task.task_id, "", [], "BUDGET_EXHAUSTED")


def verify(results: list[Result]) -> tuple[bool, list[str]]:
    errors = []
    for result in results:
        if result.status != "SUCCEEDED":
            errors.append(f"{result.task_id}:{result.status}")
        if not result.evidence:
            errors.append(f"{result.task_id}:MISSING_EVIDENCE")
    return not errors, errors


def orchestrate(tasks: list[Task], workers: dict[str, Worker]) -> RunState:
    state = RunState(run_id="run_demo")
    results = []
    for task in tasks:
        result = run_worker(task, workers[task.task_id], state)
        state.artifacts[task.task_id] = result
        results.append(result)

    accepted, errors = verify(results)
    if accepted:
        state.memory_candidates.append(
            {
                "scope": "run_shared",
                "content": "两项证据均已通过验证",
                "sources": [ref for result in results for ref in result.evidence],
                "status": "candidate",
            }
        )
    else:
        state.artifacts["verification"] = Result(
            "verification", ";".join(errors), [], "FAILED"
        )
    return state


def evidence_worker(task: Task, step: int) -> tuple[str, list[str]]:
    return "连接池等待数上升", ["metrics://pool_waiting?p95"]


def risk_worker(task: Task, step: int) -> tuple[str, list[str]]:
    return "写操作应保持关闭", ["policy://incident/read_only"]


if __name__ == "__main__":
    state = orchestrate(
        [Task("evidence", "主因证据"), Task("risk", "操作边界")],
        {"evidence": evidence_worker, "risk": risk_worker},
    )
    assert all(x.status == "SUCCEEDED" for x in state.artifacts.values())
    assert len(state.memory_candidates) == 1
    print(state)

这段代码是“控制机制验证”,不是“智能效果证明”。接入真实模型后,还需要严格 Tool Schema、超时取消、并发安全、持久化 Checkpoint、租户 ACL、Token/费用统计、Trace 和失败注入。

7.2 建议补充的故障测试

python
def repeating_worker(task: Task, step: int) -> tuple[str, list[str]]:
    return "仍在查询", ["logs://same-result"]


state = orchestrate(
    [Task("evidence", "主因证据", max_steps=3)],
    {"evidence": repeating_worker},
)
assert state.artifacts["evidence"].status == "SUCCEEDED"

上面的 Worker 第一次就给出了非空证据,所以会成功,并不能触发无进展。这正说明测试必须定义“结果有效”的业务断言,而不能只检查字段非空。生产 Verifier 应验证证据是否新鲜、是否支持结论、是否属于正确租户和时间窗。

可继续设计以下注入:

  • 第一次外部动作成功但响应丢失,确认不会重复副作用;
  • 两个 Worker 基于同一版本并发写 Artifact,确认版本冲突可见;
  • Memory 命中另一租户的高相似记录,确认 ACL 在排序前拒绝;
  • Handoff 摘要漏掉“不允许写库”,确认接收方仍受服务端权限限制;
  • Worker 持续返回同一 Observation,确认无进展次数到阈值后终止。

8. 示例项目:生产事故调查协作 Agent

证据边界| 本节是可实现的示例设计,不是本仓库已上线系统;没有真实用户量、准确率、延迟、成本或事故收益数据。

8.1 业务目标与约束

目标是帮助值班工程师在告警后快速收集证据、形成候选根因、识别高风险操作,并输出可追溯的调查摘要。约束包括:

  • 默认只读,禁止 Agent 直接重启、扩容、改配置或执行 SQL 写操作;
  • 日志、指标和变更记录来自不同系统,时间窗和版本必须一致;
  • 多个假设可以并行调查,但最终报告只能由 Orchestrator 验收;
  • 事故中的猜测不能直接成为跨事故长期 Memory;
  • 超时后允许返回部分结果和未知项,不能用幻觉补齐。

8.2 角色与职责

角色输入输出权限
Orchestrator告警、服务拓扑、验收条件调查计划、最终报告、终止原因调度与只读查询授权,无直接变更权
Evidence Agent指定假设、时间窗、指标/日志入口证据 Artifact、来源、未知项只读日志和指标
Change Agent部署、配置和依赖版本变更时间线 Artifact只读发布与配置历史
Risk Agent候选操作、Runbook、当前状态风险等级、所需审批、替代止损只读策略库
Verifier所有 Artifact 与验收规则支持、反驳、冲突和缺失证据无外部写权限

8.3 关键决策

  1. 使用 Manager-as-Tool,而不是自由 Group Chat,因为调查子任务可以按假设并行,最终责任必须集中;
  2. Agent 间传递 Artifact ID、来源和结构化摘要,而不是共享全部内部推理;
  3. Run State 用 Checkpoint 保存,事故经验只有在人工复盘批准后进入长期 Memory;
  4. 操作建议与执行工具分离,真正变更走现有审批 Workflow;
  5. 评测以最终证据支持和正确系统状态为主,不要求模型复现唯一调查路径。

8.4 方案亮点与验证边界

原基线可以是“一名通用 Agent 顺序查询所有系统”。多 Agent 方案的潜在价值是并行缩短墙钟时间、按权限隔离工具和让 Verifier 独立检查证据;代价是更多模型调用、冲突合并和 Trace 复杂度。

亮点不能写成“排障效率提升 70%”。应通过同一批脱敏历史事故或故障演练,在相同模型、工具、权限、预算和时间窗下比较:

  • 是否找到正确证据;
  • 是否引用错误时间窗或错误租户;
  • 是否产生危险操作建议;
  • 端到端耗时、总 Token 和费用;
  • 是否在预算内给出明确未知项;
  • 单 Agent 与多 Agent 的成功率差异是否稳定。

8.5 生产风险演练

假设 Evidence Agent 查到数据库等待数上升,Change Agent 查到同一时段刚发布新版本,但二者使用的时区不同,Orchestrator 错把两个事件拼成因果关系。

排查闭环:

  1. 现象:报告声称“发布导致连接池耗尽”,但人工时间线不一致;
  2. 影响范围:错误根因可能诱导回滚,掩盖真实慢查询;
  3. 定位证据:按 run_id 检查 Artifact 的 observed_attimezone、查询时间窗、数据源版本和 Verifier 冲突记录;
  4. 根因:Task Envelope 未强制统一时间语义,Verifier 只检查字段存在,没有校验时间区间重叠;
  5. 临时止损:标记报告为未验证,关闭自动建议,要求人工核对时间线;
  6. 长期修复:统一 UTC 存储与显式时区展示,在 Schema 中增加 window_start/end,Verifier 检查重叠和因果证据;
  7. 回归验证:重放跨时区、夏令时、延迟日志和部署回滚样本;
  8. 监控与防复发:统计时间窗冲突率、无来源结论率和人工驳回原因。

这个演练说明,多 Agent 会把普通数据一致性问题放大成“多个看似专业的结论互相印证”。独立角色不是独立证据,必须用时间、来源和版本做机器校验。

9. 技术难点、方案亮点与生产经验

9.1 关键技术难点

难点场景与约束为什么难失败信号拆解或解决方式验证方式
任务可分解性子任务既要并行又不能丢全局约束边界过粗无并行价值,过细则协调成本失控重复查询、结果无法合并、Manager 成为瓶颈用输入、输出、Owner、权限、依赖和验收定义 Task Envelope单 Agent 基线对比、关键路径和重复工作率
上下文最小传递Worker 需要足够信息但不能看到全部历史摘要会丢否定和来源,全量会污染与泄漏Handoff 后目标漂移、重复询问、越权Artifact 引用、结构化摘要、服务端权限和接收方输入测试Handoff 保真集、敏感字段泄漏测试
Memory 可信写入模型产生的信息真假混合错误一旦跨会话复用会持续放大陈旧建议、跨租户命中、错误规则反复出现两阶段提交、来源、TTL、版本、ACL、纠错和删除Memory Precision、Stale Injection Rate、删除传播测试
嵌套循环收敛Orchestrator 和 Worker 都可能追加任务局部有进展不代表全局接近完成调用数飙升、重复 Observation、预算失控分层预算、动作指纹、无进展检测、目标断言循环风暴故障注入与预算上界证明
并发一致性多 Worker 同时写 State 或 Artifact顺序不可预测,模型难可靠合并结果丢失、版本回退、静默覆盖Event Log、版本比较、唯一 Owner、确定性 Reducer乱序、重复、重放和进程重启测试

9.2 方案亮点与证据

原问题或基线关键决策相比备选方案的价值验证证据或方法代价与适用边界
单 Agent Context 过长且串行按独立证据域并行 Worker隔离上下文并可能缩短关键路径同任务、同模型、同预算对比成功率与墙钟时间总 Token 通常上升;不可分任务收益有限
自由群聊难以收口Manager 持有责任,Verifier 独立验收控制权清楚,冲突可见任务完成率、冲突遗漏率、未决项完整率Manager 可能成为瓶颈
全历史等同 MemoryCheckpoint、Store、Event Log 分层恢复、复用和审计职责清楚重启恢复、跨线程读取、删除和回放测试数据模型与运维复杂度提高
模型自己决定停止Runtime 硬预算 + 目标断言 + 无进展检测可证明最坏调用上界,避免无限循环重复 Observation、超时和工具错误注入预算过紧会提前终止,需要标注集调优

9.3 常见误区与反模式

  1. 按组织架构建 Agent:产品、研发、测试每个角色一个 Agent,却没有独立工具、数据或验收边界;
  2. 共享所有历史:以为信息越全越好,结果 Token、隐私和噪声同步增长;
  3. 把向量库当 Memory:只完成存取,没有来源、作用域、TTL、冲突和删除;
  4. 让模型投票决定事实:多个同源模型可能一起犯错,多数票不替代外部证据;
  5. 每次失败都再开 Agent:没有错误分类和全局预算,形成指数级循环;
  6. 边执行边写长期记忆:中间假设和错误 Observation 被永久化;
  7. 只看最终答案:没有 Run/Agent/Step Trace,无法判断首次失败来自路由、Memory、工具还是合并;
  8. 把并行等于低成本:墙钟时间可能下降,但总 Token、请求数和协调成本通常上升。

9.4 生产环境易发问题与排障闭环

问题触发条件现象与影响定位证据根因临时止损长期修复回归验证监控与防复发
Loop StormWorker 失败后 Orchestrator 反复重派调用数、费用和队列长度激增run_id 下动作指纹、错误类别、父子预算全局预算未向子任务传播,失败被当成可重试取消 Run、限流、冻结新任务分层预算、无进展阈值、错误分类和熔断注入持续超时、相同 Observation 和取消信号重复动作率、每 Run 调用数、预算终止率
Memory Poisoning未验证推测写入长期 Store后续任务持续召回错误建议Memory 来源、写入 Run、使用链、纠错历史写入与最终验收未隔离禁用该 Namespace、回滚受影响版本两阶段提交、Verifier、TTL、撤销传播重放污染记忆、纠错和删除场景Memory Precision、Stale/Poisoned Injection Rate
Handoff 信息丢失摘要裁剪掉限制或证据接收方重复工作、越权或答非所问交接前后 Context Diff、Artifact 引用、权限拒绝无结构化交接契约,安全只写在摘要收回控制权、补充原始 Artifact必填字段、服务端权限、保真测试否定词、单位、时间窗和敏感字段用例Handoff 返工率、必填字段缺失率
并发覆盖多 Worker 写同一 State/Artifact某份结论消失或版本倒退Event 顺序、版本号、写入 Owner最后写入者获胜且无冲突检测暂停写入、从 Event 重建乐观锁、唯一 Owner、Reducer乱序、重复、网络分区和重启测试版本冲突率、丢失更新告警
重复副作用工具已成功但响应超时重复发消息、重复变更或重复扣费call_id、幂等键、外部任务状态把未知结果误判为失败并盲重试停止重试、先向外部系统对账幂等键、状态查询、Outbox/补偿“成功但响应丢失”故障注入Unknown Outcome 数、重复副作用数

9.5 排障顺序

出现“多 Agent 结果差、成本高或不结束”时,按以下顺序定位:

  1. 固定 run_id、模型、Prompt、工具、Memory、拓扑和预算版本;
  2. 检查任务是否正确分解,是否出现依赖子任务被错误并行;
  3. 检查路由或 Handoff 的输入、控制权和权限是否正确;
  4. 检查 Memory 的 Namespace、ACL、检索候选、版本、时间和实际注入片段;
  5. 检查每个 Worker 的动作、Observation、状态变化和错误分类;
  6. 检查 Artifact Schema、来源、版本和合并冲突;
  7. 检查全局与局部预算、重复动作和终止原因;
  8. 用相同失败轨迹重放,比较修复前后最终状态、质量、延迟和成本。

不要第一步就“换更强模型”或“多加一个 Critic Agent”。先确定首次错误发生在哪一层,否则新 Agent 只会放大成本和不确定性。

9.6 可复用经验与预防清单

  • [ ] 单 Agent 或固定 Workflow 基线已建立;
  • [ ] 每个 Agent 有独立职责、输入输出、工具、权限和 Owner;
  • [ ] 控制权转移和结果回收语义明确;
  • [ ] Agent 间使用结构化 Envelope 和稳定 Artifact ID;
  • [ ] State、Checkpoint、Event Log、Memory 和知识库职责分开;
  • [ ] Memory 有来源、作用域、ACL、TTL、版本、纠错和删除;
  • [ ] 总预算可向所有子 Loop 传播,取消信号可向下游传播;
  • [ ] 重试区分可重试、不可重试和结果未知;
  • [ ] 写操作有幂等键、状态查询、审批或补偿;
  • [ ] Trace 能从最终结果追到每个 Agent、Step、Memory 和 Tool;
  • [ ] 评测同时覆盖质量、延迟、成本、安全与故障恢复;
  • [ ] 没有真实数据时明确写“示例设计”或“故障演练”。

9.7 面试时怎么回答

技术难点: 这个项目最难的不是把任务分给多个 Agent,而是在并行和上下文隔离的约束下,同时保证事实一致、权限不扩大和循环可收敛。我把问题拆成任务契约、Memory 生命周期、Artifact 版本和分层预算,通过失败轨迹重放与单 Agent 基线验证;当前结论只覆盖固定任务集和既定工具权限。

方案亮点: 原来的基线是一个 Agent 顺序读取全部历史,主要问题是 Context 膨胀、证据混杂和长任务不可恢复。我选择 Manager-as-Tool、Checkpoint/Store 分层和独立 Verifier,而没有选择自由群聊,因为最终责任和审计边界更清楚。亮点需要用证据覆盖、冲突遗漏、墙钟时间和总成本证明;代价是调用数、Schema 和可观测治理成本增加。

生产问题: 在生产风险演练中假设出现子 Agent 反复重试,影响费用和队列稳定性。我先取消 Run 并限制新任务,再根据 run_id、动作指纹、错误类别和父子预算定位到全局预算没有传播。长期修复是分层预算、无进展检测和取消传播,通过持续超时与重复 Observation 注入验证,并新增每 Run 调用数和重复动作率告警。

可复用经验: 我沉淀的规则是“先证明边界,再增加 Agent;先治理信息,再讨论记忆;先定义终点,再启动循环”。它适用于开放、长时、可分解任务,但固定高风险流程不能直接套用。我把它固化为 Task Envelope、Memory Schema、预算测试和 Trace 验收清单。

10. 评测与验收

10.1 先定义指标口径

指标公式或口径说明
End-State Success Rate最终状态满足全部硬验收的 Run 数 / 总 Run 数优先于逐步模仿唯一过程
Evidence Support Rate有可验证来源直接支持的关键结论数 / 关键结论总数不能只检查“有引用”
Handoff Loss Rate交接后缺失必需约束或证据的次数 / Handoff 总次数按字段和风险等级分桶
Memory Precision被注入且实际有助于正确决策的 Memory 数 / 注入 Memory 总数需人工或规则标注帮助性
Stale Injection Rate已过期或被取代仍被注入的 Memory 数 / 注入 Memory 总数反映版本与 TTL 治理
Loop Repetition Rate重复动作或重复 Observation Step 数 / 总 Step 数与无进展检测直接相关
Conflict Resolution Accuracy正确识别并处理的冲突数 / 标注冲突总数不能用多数投票代替真值
Cost per Successful Run总模型、工具与基础设施成本 / 成功 Run 数防止只优化单次调用成本

指标必须说明统计窗口,并按任务类型、租户、拓扑、模型、Prompt、Memory 版本和工具版本分桶。平均值会掩盖少数超长 Loop,应同时看 P95/P99 步数、延迟和成本。

10.2 受控对比

至少比较三组:

  1. 固定 Workflow 或单 Agent;
  2. 多 Agent,但关闭长期 Memory;
  3. 多 Agent,启用通过治理的 Memory。

控制同一任务集、模型、采样参数、工具权限、外部数据快照、总预算和验收规则。多 Agent 只有在目标指标显著改善且成本、安全与稳定性仍满足约束时才成立;某个公开产品或框架支持 Multi-Agent,只能证明它可作为候选,不能证明它在本项目更优。

10.3 故障注入矩阵

注入预期行为验收证据
Worker 超时有界重试或返回部分结果终止原因、重试次数、Checkpoint
响应丢失但工具成功先查状态,不重复副作用幂等键、外部任务记录
跨租户高相似 Memory排序前被 ACL 拒绝候选过滤 Trace、零泄漏
两个 Artifact 版本冲突显式冲突,不静默覆盖Event、版本和冲突队列
Handoff 摘要缺关键限制服务端权限仍阻止动作拒绝 Observation、审计日志
所有 Agent 给出同源错误Verifier 要求外部证据或输出未知来源图、未确认标记
用户取消下游任务和租约被传播取消取消事件、队列和工具状态

11. 面试题与递进追问

完整 L1~L7 答案见 多 Agent、Memory 与 Loop 专项面试题。主文档建议先练以下问题:

  1. 多 Agent 与“一个 Agent 调多个工具”有什么本质区别?
  2. Manager-as-Tool、Handoff 和 Group Chat 的控制权有什么差异?
  3. 为什么 Context、Checkpoint 和长期 Memory 不能混为一谈?
  4. Memory 写入怎样避免把模型幻觉永久化?
  5. 多层 Loop 怎样证明不会无限扩张?
  6. 多 Agent 共享 State 如何处理并发冲突和重复副作用?
  7. 如何用受控实验证明多 Agent 比单 Agent 值得?

面试官继续追问时,优先回到四个句柄:task_idartifact_idmemory_idrun_id。它们分别帮助说明任务边界、产物证据、记忆来源和端到端轨迹。

12. 实践任务

  • [ ] 用 Python 实现 Manager + 两个 Worker + Verifier,禁止共享可变全局变量;
  • [ ] 给 Task/Result/Memory/Checkpoint 设计 JSON Schema;
  • [ ] 同一任务比较单 Agent、Manager-as-Tool 和 Group Chat;
  • [ ] 注入重复 Observation,验证全局和局部预算都能终止;
  • [ ] 注入跨租户 Memory,验证 ACL 在相似度排序前执行;
  • [ ] 注入并发写冲突,验证不会静默覆盖;
  • [ ] 记录 Run/Agent/Step/Tool/Memory Trace,并重放一次失败轨迹;
  • [ ] 用两分钟口述“为什么不是 Agent 越多越好”。

13. 相关知识与参考资料

13.1 相关知识

13.2 已核对的一手资料

以下动态框架能力访问日期均为 2026-07-14

  1. OpenAI Agents SDK - Running agents:Runner Loop、Tool Call、Handoff、max_turns 与会话状态策略;
  2. OpenAI Agents SDK - Handoffs:控制权转移、结构化 Handoff 输入、输入过滤与历史边界;
  3. LangGraph - Persistence:Checkpointer 的线程级 State 与 Store 的跨线程 Memory 区分;
  4. Microsoft AutoGen - Teams:Round Robin、Selector、Swarm 等团队形式及先优化单 Agent 的建议;
  5. Anthropic - How we built our multi-agent research system:Manager/Worker 并行研究、协调与评测经验;
  6. ReAct: Synergizing Reasoning and Acting in Language Models:Reasoning、Action 与 Observation 交替的经典 Agent Loop 研究。

事实边界| 官方文档能证明框架公开支持某种 Loop、Handoff、Team 或 Persistence 能力,不能证明该方案在本项目数据上质量、延迟和成本最优。本文的 Schema、指标、故障演练和生产参考架构属于工程建议,需通过真实实现验证。

14. 简明总结

一句话记忆: 多 Agent 负责划分协作边界,Memory 负责治理跨步骤信息,Loop 负责让决策在真实 Observation 上受控推进;三者最终都必须服从确定性 Runtime 的权限、状态、预算和验收。

  • 多 Agent 不是角色越多越好,而是职责、上下文、权限和产物边界足够清楚时才值得拆分;
  • Memory 不是全量历史或向量库,而是带来源、作用域、TTL、版本、ACL、纠错和删除的信息生命周期;
  • Loop 不是模型自由 while,而是动作、执行、观察、状态更新和硬终止组成的状态机;
  • 生产排障先固定 run_id 和版本,再沿任务分解、Handoff、Memory、Worker、Artifact、预算与终止逐层定位;
  • 面试最应讲清三点:谁拥有最终责任、哪些事实可以跨边界流动、系统怎样证明会停止且结果可验收。