企业官网建设流程全解析

1. 多 Agent 系统为什么会走向失控

1.1 从单体 Agent 到多 Agent 协作的演进逻辑

2024 年之前，大多数团队对 Agent 的理解还停留在“一个模型加几个工具函数”的阶段。一个 Agent 负责理解用户意图、调用 API、返回结果，链路短、状态少、出问题也好排查。但从 2025 年下半年开始，情况发生了明显变化：任务复杂度上来了，单 Agent 的上下文窗口和推理深度都不够用了，于是大家开始把任务拆开，让多个 Agent 各管一摊，再通过某种编排层把它们串起来。

这个思路本身没问题。一个负责检索、一个负责推理、一个负责执行、一个负责校验，听起来分工明确。但真正落地之后你会发现，多 Agent 系统的复杂度不是线性增长的，而是指数级膨胀的。两个 Agent 之间有 2 条通信路径，五个 Agent 之间就可能出现几十条交互链路，再加上共享记忆、工具调用、状态回传，整个系统很快就变成了一团看不清的乱麻。

我见过一个很典型的案例：某团队做了一套多 Agent 协作的代码审查系统，一个 Agent 负责读代码，一个负责找问题，一个负责写修复建议，还有一个负责最终汇总。上线第一周运行良好，第二周开始出现“修复建议和原始问题对不上”的情况，第三周直接演变成两个 Agent 互相引用对方的输出，陷入无限循环，把 API 额度烧光了才被发现。

这就是多 Agent 走向失控的典型症状：不是某个 Agent 坏了，而是 Agent 之间的交互关系坏了。

1.2 失控的六种典型表现

把过去一年我接触到的多 Agent 项目问题做个归类，基本逃不出下面六种情况：

这六种问题里，最隐蔽也最危险的是“记忆污染”。因为循环引用和资源争抢通常很快就能暴露出来，但记忆污染是慢性的——一个 Agent 在某次任务中产生了一条错误结论，这条结论被写进了共享记忆，下一个任务里另一个 Agent 读到这条记忆，基于它做出了错误决策，再写回记忆，错误就像滚雪球一样越滚越大。等你发现的时候，已经很难追溯到底是哪一步开始出的问题。

1.3 为什么传统的微服务治理经验不够用了

很多做后端出身的同学会觉得，多 Agent 治理不就是微服务治理那一套吗？服务发现、熔断降级、链路追踪，搬过来不就行了？

部分可以，但不够。微服务之间的调用是确定性的：A 调 B 的某个接口，传入参数，返回结果，行为可预期。但 Agent 之间的交互是非确定性的：同一个 Agent 在相同输入下可能产生不同输出，它可能会“自作主张”地调用一个你没预料到的工具，也可能在推理过程中产生一段看似合理但实际错误的中间结论。

更麻烦的是，微服务的服务边界是人为划定的，相对稳定；而 Agent 的角色边界往往是动态的，一个 Agent 可能在任务的不同阶段扮演不同角色。这就导致传统的基于固定拓扑的治理手段很难直接套用。

所以我们需要一套专门针对 Agent 特性的治理框架。这也是为什么 2025 年底开始，FCoP 这类面向 Agent 协作的协议开始受到关注——它们试图解决的正是“非确定性主体之间如何可靠协作”这个问题。

2. Agent 治理的核心框架与协议选型

2.1 FCoP 协议到底解决了什么问题

FCoP 全称是 Framework for Collaborative Protocol，是一套面向多 Agent 协作的通信与治理规范。它的核心思路可以用一句话概括：把 Agent 之间的每一次交互都变成可审计、可约束、可回滚的结构化事件。

传统多 Agent 系统里，Agent A 给 Agent B 发消息，可能就是一段自然语言文本或者一个 JSON 对象，B 收到之后自行理解、自行决策。这种模式灵活但不可控。FCoP 的做法是在消息层加一层“契约”：每次交互必须声明意图类型、期望的输出格式、超时时间、失败处理策略，以及这次交互属于哪个上层任务。

我举个具体例子。假设你有一个“数据分析 Agent”和一个“报告生成 Agent”。传统模式下，数据分析 Agent 可能直接甩一段文字给报告生成 Agent：“销售额下降了 15%，可能是因为季节性因素。”报告生成 Agent 拿到这段话，自行决定怎么用。

在 FCoP 模式下，这条消息会被结构化为：

{ "intent": "data_insight", "source_agent": "data_analyzer", "target_agent": "report_writer", "task_id": "task_20260412_001", "payload": { "metric": "sales", "change": -0.15, "hypothesis": "seasonal_factor", "confidence": 0.72 }, "constraints": { "timeout_ms": 5000, "max_retries": 2, "on_failure": "return_partial" }, "audit": { "created_at": "2026-04-12T10:23:45Z", "parent_event": "evt_20260412_0012" } }

这样做的好处是：第一，每个 Agent 收到的输入格式是确定的，减少了理解偏差；第二，超时和重试策略是显式声明的，不会出现无限等待；第三，每个事件都有 parent_event，可以完整回溯整条调用链；第四，confidence 字段让下游 Agent 知道这个结论的可信度，可以据此决定是否需要交叉验证。

2.2 协议选型：FCoP、MQTT 还是自研

在实际项目中，Agent 之间的通信协议选型是个绕不开的问题。我对比过几种常见方案：

我的建议是：如果你的 Agent 数量在 5 个以内，任务链路比较固定，自研一套轻量的 JSON 通信格式就够了，没必要上 FCoP。但如果 Agent 数量超过 10 个，或者任务链路是动态生成的，那 FCoP 这类结构化协议能帮你省掉大量排查问题的时间。

MQTT 在 Agent 场景下有个独特的优势：它的发布订阅模型天然适合“一个 Agent 的输出需要被多个 Agent 消费”的场景。比如一个“感知 Agent”产出的环境状态，可能同时被“规划 Agent”和“监控 Agent”需要。用 MQTT 的话，感知 Agent 只管发布，不需要知道谁在消费。但 MQTT 的缺点是缺少 Agent 语义，你需要在 payload 里自己定义意图、约束这些字段，相当于在 MQTT 之上再搭一层 FCoP 的壳。

2.3 治理层的三个核心模块

不管用什么协议，一个完整的 Agent 治理层至少需要三个模块：

第一个是注册与发现模块。每个 Agent 启动时向治理层注册自己的身份、能力、当前负载、健康状态。治理层维护一份实时更新的 Agent 目录。当 Agent A 需要调用 Agent B 时，先向治理层查询 B 的可用实例，而不是硬编码地址。这个模块解决的是“责任漂移”问题——每个 Agent 的能力边界是显式注册的，不会出现“我以为你会做”的情况。

第二个是调用链管理模块。每次 Agent 之间的调用都会生成一个 trace_id，所有相关事件都挂在这个 trace_id 下面。治理层可以实时监控每条调用链的深度、耗时、成功率。当发现某条链路的深度超过阈值（比如 10 层），或者某个 Agent 在短时间内被反复调用（可能是循环引用），治理层可以主动中断这条链路。这个模块解决的是“循环引用”和“级联失败”问题。

第三个是共享记忆管理模块。多 Agent 系统通常需要一个共享的记忆存储（常见的是 Redis），让不同 Agent 可以读写公共状态。但这个共享记忆必须有治理：写入时需要校验（这条记忆的来源是否可信、是否与已有记忆冲突），读取时需要标注新鲜度（这条记忆是什么时候写的、是否可能已过期），删除时需要审计（谁删的、为什么删）。这个模块解决的是“记忆污染”问题。

3. 实操：搭建一套可落地的 Agent 治理系统

3.1 环境准备与基础架构

下面我用一个具体的例子来演示怎么搭建 Agent 治理系统。假设我们要做一个“智能客服多 Agent 系统”，包含四个 Agent：意图识别 Agent、知识检索 Agent、回复生成 Agent、质量校验 Agent。

基础架构是这样的：

• 治理层：一个独立的服务，负责 Agent 注册、调用链管理、共享记忆管理
• 通信层：用 Redis 的 Pub/Sub 做消息传递，用 Redis 的 Hash 做共享记忆存储
• 监控层：用 OpenTelemetry 做链路追踪，数据写入 ClickHouse

为什么选 Redis 而不是消息队列？因为在这个场景下，Agent 之间的通信是低延迟、高频率的，Redis 的 Pub/Sub 足够快，而且 Redis 本身就可以兼做共享记忆存储，少维护一个组件。如果你们的场景对消息可靠性要求极高（比如金融交易），那还是上 Kafka 更稳妥。

环境准备的具体步骤：

# 启动 Redis（用于消息传递和共享记忆） docker run -d --name agent-redis -p 6379:6379 redis:7.2-alpine # 启动 ClickHouse（用于链路追踪数据存储） docker run -d --name agent-clickhouse -p 8123:8123 -p 9000:9000 clickhouse/clickhouse-server:24.3 # 启动治理层服务（假设你已经写好了治理层的代码） docker run -d --name agent-governance -p 8080:8080 --link agent-redis:redis --link agent-clickhouse:clickhouse your-governance-image:latest

治理层的核心配置：

# governance-config.yaml governance: agent_registry: heartbeat_interval_ms: 5000 unhealthy_threshold: 3 call_chain: max_depth: 10 max_duration_ms: 30000 on_depth_exceeded: "terminate_and_alert" shared_memory: backend: "redis" key_prefix: "agent_mem:" ttl_seconds: 3600 write_validation: true conflict_resolution: "latest_wins"

这里有几个参数需要解释一下。max_depth: 10意味着任何一条调用链最多嵌套 10 层，超过就强制终止。这个值怎么定？我的经验是：先统计你系统里正常任务的平均调用深度，然后乘以 2 作为阈值。比如正常任务平均 4 层，那设 8 到 10 比较合适。设太小会误杀正常任务，设太大起不到保护作用。

ttl_seconds: 3600是共享记忆的过期时间。这个值取决于你的任务周期。如果一次完整的客服对话平均 5 分钟，那 1 小时是合理的。如果任务周期很长（比如跨天的数据分析），那需要相应调大。

3.2 Agent 注册与心跳机制的实现

每个 Agent 启动时，需要向治理层注册自己。注册信息包括：

import requests import time import threading class AgentRegistration: def __init__(self, agent_id, capabilities, governance_url): self.agent_id = agent_id self.capabilities = capabilities self.governance_url = governance_url self.heartbeat_thread = None self.running = False def register(self): payload = { "agent_id": self.agent_id, "capabilities": self.capabilities, "endpoint": f"redis://agent-channel-{self.agent_id}", "max_concurrent_tasks": 5, "version": "1.0.0" } resp = requests.post( f"{self.governance_url}/api/v1/agents/register", json=payload, timeout=5 ) if resp.status_code == 200: print(f"Agent {self.agent_id} registered successfully") return True else: print(f"Registration failed: {resp.text}") return False def start_heartbeat(self): self.running = True self.heartbeat_thread = threading.Thread(target=self._heartbeat_loop) self.heartbeat_thread.daemon = True self.heartbeat_thread.start() def _heartbeat_loop(self): while self.running: try: requests.post( f"{self.governance_url}/api/v1/agents/{self.agent_id}/heartbeat", json={"status": "healthy", "load": self._get_current_load()}, timeout=3 ) except Exception as e: print(f"Heartbeat failed: {e}") time.sleep(5) def _get_current_load(self): # 返回当前正在处理的任务数 return 0

这段代码的关键点是心跳间隔设为 5 秒，治理层连续 3 次没收到心跳就标记为不健康。这意味着一个 Agent 从挂掉到被治理层发现，最多有 15 秒的窗口期。在这 15 秒内，如果有其他 Agent 向它发消息，消息会丢失。所以你的消息发送逻辑需要处理这种情况：发送失败时，先查询治理层获取该 Agent 的最新状态，如果确实不健康，就触发降级策略。

3.3 调用链追踪与循环检测

调用链追踪的核心是在每次 Agent 交互时传递一个 trace 上下文。我用 Redis 的 Hash 来存储调用链信息：

import redis import json import uuid import time class CallChainManager: def __init__(self, redis_client): self.redis = redis_client self.max_depth = 10 self.max_duration_ms = 30000 def create_trace(self, initiator_agent, task_description): trace_id = f"trace_{uuid.uuid4().hex[:12]}" trace_data = { "trace_id": trace_id, "initiator": initiator_agent, "task": task_description, "created_at": time.time(), "depth": 0, "events": [] } self.redis.setex( f"trace:{trace_id}", 3600, json.dumps(trace_data) ) return trace_id def record_event(self, trace_id, source_agent, target_agent, intent): trace_key = f"trace:{trace_id}" trace_data = json.loads(self.redis.get(trace_key)) # 检查深度 current_depth = trace_data["depth"] + 1 if current_depth > self.max_depth: self._terminate_trace(trace_id, "max_depth_exceeded") return False # 检查耗时 elapsed_ms = (time.time() - trace_data["created_at"]) * 1000 if elapsed_ms > self.max_duration_ms: self._terminate_trace(trace_id, "max_duration_exceeded") return False # 检查循环引用 event_signature = f"{source_agent}->{target_agent}:{intent}" recent_events = [e["signature"] for e in trace_data["events"][-5:]] if recent_events.count(event_signature) >= 3: self._terminate_trace(trace_id, "circular_reference_detected") return False # 记录事件 trace_data["depth"] = current_depth trace_data["events"].append({ "signature": event_signature, "source": source_agent, "target": target_agent, "intent": intent, "timestamp": time.time() }) self.redis.setex(trace_key, 3600, json.dumps(trace_data)) return True def _terminate_trace(self, trace_id, reason): trace_key = f"trace:{trace_id}" trace_data = json.loads(self.redis.get(trace_key)) trace_data["terminated"] = True trace_data["termination_reason"] = reason trace_data["terminated_at"] = time.time() self.redis.setex(trace_key, 3600, json.dumps(trace_data)) # 同时发送告警 self._send_alert(trace_id, reason) def _send_alert(self, trace_id, reason): alert = { "level": "warning", "trace_id": trace_id, "reason": reason, "timestamp": time.time() } self.redis.publish("agent_alerts", json.dumps(alert))

这段代码里有个细节值得注意：循环检测用的是“最近 5 个事件里同一个签名出现 3 次”作为判断条件。为什么是 3 次而不是 2 次？因为正常的重试逻辑可能导致同一个调用出现 2 次，如果设成 2 次会误杀正常重试。3 次基本可以确认是循环了。

另一个细节是_terminate_trace不仅标记了终止，还发了告警。这个告警会推送到一个专门的告警频道，运维人员可以订阅这个频道，实时看到哪些调用链被终止了、原因是什么。

3.4 共享记忆的写入校验与冲突解决

共享记忆是多 Agent 系统里最容易出问题的地方。我的做法是在写入时加三层校验：

class SharedMemoryManager: def __init__(self, redis_client): self.redis = redis_client self.key_prefix = "agent_mem:" def write(self, key, value, source_agent, confidence, ttl=3600): full_key = f"{self.key_prefix}{key}" # 第一层：来源校验 if not self._is_trusted_source(source_agent): return {"success": False, "reason": "untrusted_source"} # 第二层：置信度校验 if confidence < 0.5: return {"success": False, "reason": "low_confidence"} # 第三层：冲突检测 existing = self.redis.get(full_key) if existing: existing_data = json.loads(existing) if self._is_conflicting(existing_data, value): # 冲突解决策略：保留置信度更高的 if confidence > existing_data.get("confidence", 0): pass # 覆盖 else: return {"success": False, "reason": "conflict_kept_existing"} # 写入 memory_entry = { "value": value, "source": source_agent, "confidence": confidence, "written_at": time.time(), "version": (existing_data.get("version", 0) + 1) if existing else 1 } self.redis.setex(full_key, ttl, json.dumps(memory_entry)) # 记录写入审计日志 self._audit_log(key, source_agent, value, confidence) return {"success": True, "version": memory_entry["version"]} def _is_trusted_source(self, agent_id): # 从治理层查询该 Agent 是否在可信列表中 trusted = self.redis.smembers("trusted_agents") return agent_id.encode() in trusted def _is_conflicting(self, existing_data, new_value): # 简单的冲突判断：如果两个值都是数值型且差异超过 20%，视为冲突 old_val = existing_data.get("value") if isinstance(old_val, (int, float)) and isinstance(new_value, (int, float)): if old_val != 0: diff_ratio = abs(new_value - old_val) / abs(old_val) return diff_ratio > 0.2 return False def _audit_log(self, key, source_agent, value, confidence): log_entry = { "key": key, "source": source_agent, "value_preview": str(value)[:100], "confidence": confidence, "timestamp": time.time() } self.redis.lpush("memory_audit_log", json.dumps(log_entry)) self.redis.ltrim("memory_audit_log", 0, 9999)

这里的三层校验各有侧重。来源校验防止不可信的 Agent 写入垃圾数据；置信度校验防止低质量的推理结果污染记忆；冲突检测防止两个 Agent 写入矛盾的数据。冲突解决策略我选的是“保留置信度更高的”，但这不是唯一选择。有些场景下“保留最新的”更合适，有些场景下“保留并标记冲突，等待人工介入”更安全。具体选哪种取决于你的业务对数据一致性的要求。

4. 常见问题与排查技巧实录

4.1 Agent 无响应或响应超时的排查路径

这是最常见的问题。一个 Agent 发消息给另一个 Agent，等了半天没反应。排查路径应该是这样的：

第一步，查治理层的 Agent 注册表，确认目标 Agent 是否在线。如果不在线，说明它挂了或者网络断了，直接触发降级策略。

第二步，如果目标 Agent 在线，查它的当前负载。如果负载已经满了（正在处理的任务数达到 max_concurrent_tasks），那消息可能在队列里排队。这时候要么等，要么扩容。

第三步，如果负载正常但就是不响应，查调用链追踪。看看这条 trace 是不是已经被终止了（可能因为深度超限或循环检测）。如果被终止了，看终止原因。

第四步，如果 trace 正常但目标 Agent 就是不回，那可能是目标 Agent 内部卡住了。这时候需要查目标 Agent 的日志，看它卡在哪个步骤。

我整理了一个速查表：

这里有个经验：Redis 的 Pub/Sub 是不保证消息可靠投递的。如果订阅者当时不在线，消息就丢了。对于可靠性要求高的场景，应该用 Redis 的 Stream 或者上 Kafka。我一般会在 Agent 的发送逻辑里加一个确认机制：发送后等待接收方的 ack，如果 3 秒内没收到 ack，就重发，最多重发 2 次。

4.2 记忆污染导致的行为异常怎么定位

记忆污染是最难排查的问题，因为它的表现往往是“Agent 行为变得奇怪”，但你不容易想到是记忆出了问题。

我的排查方法是：在治理层加一个“记忆快照”功能，每隔一段时间（比如每 5 分钟）把共享记忆的当前状态 dump 一份到 ClickHouse。当发现 Agent 行为异常时，对比异常发生前后的记忆快照，看哪些 key 发生了变化。

具体操作：

-- 查询某个 key 在指定时间范围内的变化历史 SELECT timestamp, key, source, value_preview, confidence FROM memory_audit_log WHERE key = 'customer_intent:session_12345' AND timestamp BETWEEN '2026-04-12 10:00:00' AND '2026-04-12 10:30:00' ORDER BY timestamp;

如果发现某个 key 在短时间内被多个不同 Agent 反复写入，而且值在剧烈变化，那基本可以确定是记忆污染。解决方式是：第一，找出写入这些值的 Agent，检查它们的逻辑；第二，在治理层加一条规则，对高频变化的 key 加写入锁，同一时间只允许一个 Agent 写入；第三，如果已经污染了，手动清理这个 key，让系统重新开始。

4.3 多 Agent 协作中的“责任漂移”怎么防

责任漂移的典型表现是：一个任务在多个 Agent 之间转了一圈，最后没人真正处理。比如意图识别 Agent 把任务转给知识检索 Agent，知识检索 Agent 觉得这不是自己的事，又转回意图识别 Agent，意图识别 Agent 又转出去……

防责任漂移的核心是“每个任务必须有明确的 owner”。我的做法是在任务创建时就指定一个 owner Agent，这个 Agent 对任务的最终完成负责。其他 Agent 可以协助，但不能把责任推回给 owner。

具体实现上，在 FCoP 的消息结构里加一个task_owner字段。当 owner Agent 收到一个任务时，它必须做出决策：要么自己处理，要么委派给其他 Agent 并明确等待结果，要么标记为无法处理并返回失败。不允许出现“转给别人就不管了”的情况。

治理层可以监控每个任务的 owner 变更历史。如果一个任务的 owner 在短时间内变更超过 3 次，就触发告警，提示可能存在责任漂移。

4.4 性能优化：减少不必要的 Agent 间通信

多 Agent 系统跑起来之后，你可能会发现性能瓶颈不在单个 Agent 的推理速度上，而在 Agent 之间的通信开销上。每次通信都有序列化、网络传输、反序列化的成本，Agent 数量一多，这些成本累积起来很可观。

我的优化经验是：第一，合并可以合并的调用。如果 Agent A 需要连续调用 Agent B 三次，看看能不能改成一次批量调用。第二，缓存高频读取的共享记忆。如果某个 key 被多个 Agent 频繁读取但很少变化，可以在 Agent 本地加一层缓存，减少对 Redis 的访问。第三，异步化非关键路径。有些 Agent 之间的通信不是主链路必需的（比如日志记录、指标上报），可以改成异步，不阻塞主流程。

实测下来，这三招能把多 Agent 系统的端到端延迟降低 30% 到 50%。当然具体效果取决于你的系统架构和通信模式。

4.5 治理层本身的容错设计

最后说一个容易被忽略的问题：治理层本身挂了怎么办？

治理层是所有 Agent 的依赖，如果它挂了，整个系统就瘫痪了。所以治理层必须做高可用。我的做法是：治理层至少部署两个实例，前面挂一个负载均衡。治理层的状态数据（Agent 注册表、调用链信息）存在 Redis 里，两个实例共享同一份 Redis，这样任何一个实例挂了，另一个可以立刻接管。

另外，Agent 端要做降级处理。如果治理层暂时不可达，Agent 不应该直接崩溃，而是进入“降级模式”：使用本地缓存的 Agent 注册表（可能不是最新的，但至少能用），跳过调用链追踪（记录到本地日志，等治理层恢复后再上报），共享记忆读写直接走 Redis（绕过治理层的校验逻辑，但加一个标记，等治理层恢复后补做校验）。

这套降级机制我实际用过一次，当时治理层因为一个内存泄漏问题重启了大约 40 秒，因为有降级机制，业务侧完全没有感知到异常。如果没有降级，这 40 秒里所有 Agent 都会报错，影响面就大了。

5. 从治理到自治：多 Agent 系统的演进方向

5.1 当前治理框架的局限性

说实话，现在这套治理框架本质上还是“外部管控”的思路：在 Agent 之上加一层监管，约束它们的行为。这种方式能解决大部分问题，但有两个根本局限。

第一个局限是治理层可能成为瓶颈。所有 Agent 的注册、调用链记录、记忆写入都要经过治理层，Agent 数量一多，治理层的负载就上来了。虽然可以通过分片、缓存等手段缓解，但架构上始终多了一层。

第二个局限是治理规则是静态的。max_depth 设成 10 就是 10，不会根据实际情况动态调整。但实际系统中，不同任务的合理调用深度是不一样的。一个简单的查询任务可能 3 层就够了，一个复杂的分析任务可能需要 8 层。用同一个阈值去约束所有任务，要么误杀要么漏放。

5.2 让 Agent 自己治理自己

演进的方向是让 Agent 具备“自治”能力：每个 Agent 自己知道自己的边界在哪里，自己知道什么时候该拒绝、什么时候该转交、什么时候该升级。

这需要 Agent 在推理时不仅考虑“怎么做这个任务”，还要考虑“我该不该做这个任务”“我做完之后该不该把结果传下去”。这其实就是把治理逻辑内化到 Agent 的推理过程中。

实现方式上，可以在 Agent 的 system prompt 里加入治理相关的指令，比如：“如果你发现当前任务的调用深度已经超过 5 层，不要继续调用其他 Agent，而是直接返回当前结果并标记为 partial。”或者：“如果你要写入共享记忆，先检查这个 key 是否在最近 10 分钟内被其他 Agent 写入过，如果是，不要覆盖，而是追加并标记冲突。”

这种方式的好处是治理逻辑和 Agent 逻辑融为一体，不需要额外的治理层。但缺点是治理规则变成了 prompt 的一部分，修改起来不如配置文件方便，而且不同模型的遵循程度可能不一样。

5.3 一个值得关注的实践方向：A-MemGuard

2026 年初开始，A-MemGuard 这类“主动防御”框架开始受到关注。它的核心思路是在 Agent 的记忆写入和读取路径上加一层“守卫”，这个守卫不是简单的规则校验，而是用一个小模型来判断“这条记忆是否可信”“这次读取是否安全”。

比如，当一个 Agent 要写入一条记忆时，守卫模型会评估：这条记忆的来源是否可靠？内容是否与已有记忆矛盾？是否包含可能的注入攻击？如果评估不通过，写入被拦截，并记录一条审计日志。

读取时也一样：守卫模型会评估这次读取是否合理。如果一个 Agent 突然读取了大量与当前任务无关的记忆，守卫会拦截并告警，防止敏感信息泄露或记忆被恶意利用。

这种方式的优势是比静态规则更灵活，能应对更复杂的场景。但代价是引入了一个额外的模型推理，增加了延迟和成本。目前我看到的实践是在关键路径上用守卫，非关键路径还是用规则。

5.4 给正在做多 Agent 系统的团队的建议

如果你现在正在做多 Agent 系统，我的建议是：不要一开始就上全套治理框架。先用最简单的架构把业务跑通，等遇到具体问题了再针对性解决。

具体来说，第一阶段（Agent 数量少于 5 个）：用简单的 JSON 通信，加一个基本的超时和重试机制就够了。第二阶段（Agent 数量 5 到 15 个）：引入调用链追踪和共享记忆管理，开始关注循环引用和记忆污染问题。第三阶段（Agent 数量超过 15 个）：上完整的治理层，考虑 FCoP 这类结构化协议，做高可用和降级设计。

我见过太多团队在只有 3 个 Agent 的时候就花大力气搭治理框架，结果业务逻辑一变，框架全白搭。治理是为了业务服务的，不要本末倒置。

另外，治理层的日志一定要保留足够长的时间。我建议至少保留 30 天。因为很多记忆污染和循环引用问题是慢性的，可能运行了一两周才暴露出来。如果日志只保留 3 天，等你发现问题的时候，现场早就没了。

最后分享一个我在实际项目中总结的小技巧：在治理层加一个“模拟回放”功能。把某条出问题的调用链的完整事件序列导出来，在一个隔离环境里重新回放一遍，观察每个 Agent 的行为。这个功能对于定位复杂的多 Agent 交互问题非常有用，比看日志高效得多。实现上也不复杂，就是把 trace 里记录的事件按时间顺序重新发一遍，但把 Agent 的输出替换成当时记录的输出，这样就能精确复现问题现场。