去中心化多智能体系统：基于共享上下文的协同AI架构设计与实战

1. 从“单打独斗”到“群策群力”为什么我们需要去中心化多智能体系统如果你最近关注AI领域尤其是大模型应用可能会发现一个趋势单个AI模型的能力再强也总有它的边界。让它写代码可以但让它同时监控服务器、分析日志、再根据分析结果去自动扩容可能就力不从心了。这就像让一个全栈工程师去同时处理运维、开发和产品设计效率必然打折。于是一个更“社会化”的AI架构思路开始流行让多个专门的AI智能体Agent协同工作共同完成复杂任务。这就是多智能体系统Multi-Agent Systems, MAS。但传统的多智能体系统往往需要一个“中央大脑”来指挥调度。这个中心节点负责任务分解、分配和结果汇总。这种架构在实验室里跑Demo没问题一旦放到真实、动态、甚至网络不稳定的环境中问题就来了中心节点一旦宕机整个系统瘫痪所有通信都经过中心容易成为性能瓶颈和单点故障更重要的是智能体之间缺乏直接的、灵活的上下文理解协作显得僵硬。“Decentralized Multi-Agent Systems with Shared Context”去中心化多智能体系统与共享上下文这个概念就是为了解决这些问题而生的。它描绘的是一种更接近真实人类团队协作的AI组织方式没有唯一的领导每个智能体都是平等的参与者它们通过一个共享的“上下文”或“工作记忆”来对齐认知、协调行动。这个共享上下文就像是团队共用的白板、项目看板或者群聊记录记录了任务目标、当前状态、已完成的工作、遇到的障碍以及临时的决策。每个智能体都可以去查看、更新这个共享空间从而知道自己该做什么也知道队友在做什么。这种架构的价值在于韧性、可扩展性和灵活性。没有单点故障系统更健壮智能体可以动态加入或退出易于扩展共享上下文使得智能体间的协作不再是简单的“请求-响应”而是基于共同认知的、更有机的配合。这对于构建复杂的AI应用如自动化运营、游戏NPC生态、供应链协同优化、甚至科研探索都提供了全新的可能性。接下来我们就深入拆解这个系统的核心组件、实现原理以及在实际搭建时会遇到的那些“坑”。2. 系统基石拆解“去中心化”与“共享上下文”的核心组件要理解并构建这样一个系统我们首先得把两个核心概念——“去中心化”和“共享上下文”——拆解成具体的技术组件。这不仅仅是概念而是需要落地的工程模块。2.1 “去中心化”的通信层告别中心调度器去中心化的核心是通信模式。我们不能再依赖一个中心化的任务调度器Orchestrator。取而代之的是一种对等网络Peer-to-Peer, P2P或发布-订阅Pub/Sub的通信模型。智能体发现与寻址在一个动态系统中智能体可能随时上线或下线。我们需要一个轻量级的服务发现机制。这不一定需要一个复杂的服务注册中心如Consul、Etcd可以更简单。例如采用基于Gossip的协议让智能体定期广播自己的存在和元数据能力、状态或者在系统初始化时通过一个配置文件或环境变量注入已知的同伴地址列表。对于中小规模系统后者更简单直接。消息传递协议智能体之间如何交换信息直接使用HTTP/REST API在动态P2P网络中并不方便因为每个智能体都需要充当服务器。更常见的做法是使用消息队列或事件总线。例如让所有智能体都连接到一个共用的Redis Pub/Sub频道、Apache Kafka主题或者使用专门为分布式应用设计的框架如ZeroMQ。智能体将消息发布到特定的主题如task.announce,result.report其他关心该主题的智能体就会接收到。这种方式天然解耦了发送者和接收者。消息格式标准化为了确保智能体之间能互相理解必须定义一套统一的消息信封Envelope。一个典型的消息结构可能包括{ msg_id: uuid_v4, timestamp: 2023-10-27T10:00:00Z, sender: agent_coder, recipients: [agent_tester, agent_deployer], // 可以是广播null或指定列表 topic: code.review.request, payload: { code_snippet: def calculate(...), file_path: /src/utils.py, context_ref: shared_context_id_123 // 关联到共享上下文中的某个条目 } }这个格式定义了谁发的、发给谁、关于什么、以及负载是什么是系统内沟通的“普通话”。2.2 “共享上下文”的存储与同步构建团队的共同记忆共享上下文是整个系统的“灵魂”。它不是一个简单的数据库而是一个动态的、结构化的、可被所有智能体并发访问的“协作空间”。存储后端选型我们需要一个能够支持快速读写、有一定数据结构能力、并且方便多节点访问的存储。键值存储Key-Value Store是一个很好的起点因为它简单、高效。Redis是绝佳的选择它不仅支持简单的键值还提供List、Set、Hash、Sorted Set等丰富数据结构非常适合存储对话历史、任务队列、状态标记等。此外它的Pub/Sub功能可以直接用于我们上面提到的消息通信一举两得。如果对持久化和更复杂的查询有要求也可以考虑使用文档数据库如MongoDB但需要处理好实时同步的挑战。上下文的结构化设计不能把所有信息都堆在一个键里。我们需要设计一个命名空间Namespace来组织上下文。例如context:session:session_id: 存储本次协作会话的元数据如创建时间、参与智能体列表、最终目标。context:session:session_id:history: 一个List按时间顺序存储所有重要的消息和事件。context:session:session_id:artifacts: 一个Hash存储协作产生的中间产物如{“draft_design”: “设计文档内容”, “approved_api_spec”: “API规范JSON”}。context:session:session_id:state: 一个Hash存储当前任务状态如{“current_phase”: “testing”, “blocker”: “集成环境宕机”, “owner”: “agent_ops”}。并发控制与一致性这是最大的挑战之一。当两个智能体同时想去更新“当前阶段”state时会发生什么我们需要乐观锁或悲观锁机制。在Redis中可以使用WATCH/MULTI/EXEC命令来实现乐观锁。智能体在修改前先WATCH相关的键如果在事务执行前该键被其他客户端修改则本次事务失败需要重试。这保证了在并发写入时的数据安全。上下文的生命周期与清理共享上下文不能无限增长。我们需要为每个会话上下文设置TTL生存时间在会话明确结束后或超时后自动清理释放资源。这可以通过Redis的EXPIRE命令轻松实现。注意共享上下文的设计直接决定了系统的智能程度和协作效率。设计得太简单智能体缺乏足够的背景信息设计得太复杂又会带来巨大的同步开销和认知负担。初期建议从最核心的“任务目标”、“当前状态”、“历史动作”三个维度开始随着场景复杂再逐步扩展。3. 智能体设计范式从“功能模块”到“自主协作者”在这样一个去中心化系统中单个智能体的设计思路与传统的微服务或函数有本质不同。它不再是一个被动的、等待调用的服务而是一个具有一定自主性、能感知环境、并能主动采取行动的协作者。3.1 智能体的核心循环与决策逻辑一个典型的去中心化智能体其内部运行着一个持续的核心循环可以概括为“感知-思考-行动”感知Perceive智能体持续监听消息总线如Redis Pub/Sub上自己关心的主题。同时它也会定期或由事件触发去读取共享上下文中与自己相关的部分。例如一个“代码审查智能体”会订阅code.*相关的主题并关注上下文中state.current_phase是否为 “development”。思考Think/Deliberate接收到信息后智能体需要理解当前状况。这通常通过以下步骤完成上下文构建将收到的消息与从共享上下文中提取的相关历史信息组合起来形成一个完整的、用于决策的“情境提示”Context Prompt。意图推理将这个情境提示输入到大语言模型LLM如GPT-4、Claude或本地部署的模型中让LLM分析当前情况并决定下一步该做什么。LLM的提示词Prompt需要精心设计以明确智能体的角色、职责、可用动作以及协作规范。动作规划LLM的输出应该是一个结构化的动作决策例如{“action”: “review_code”, “parameters”: {“code_ref”: “xxx”}, “next”: “notify_agent_tester”}。行动Act根据决策执行具体操作。这可能是内部计算运行一段代码逻辑。调用工具调用一个外部API如执行单元测试、调用GitHub API创建PR。更新共享上下文将行动的结果、产生的新数据或状态变更写入共享上下文注意使用并发控制。发送消息向消息总线发布一个新消息通知其他智能体。例如审查完成后发布一个code.review.complete消息。这个循环使得智能体能够自主运行无需外部指令驱动。3.2 角色定义与能力封装每个智能体应该有清晰的角色边界。例如在一个软件研发自动化系统中我们可以定义产品经理智能体负责解析原始需求将其拆解为用户故事和任务并初始化共享上下文中的任务目标。架构师智能体监听任务创建负责进行高层设计产出技术方案草图并写入上下文。开发智能体根据技术方案和具体任务编写代码运行基础静态检查并将代码片段和检查结果提交到上下文。测试智能体监听代码提交负责编写测试用例、运行测试并将测试报告和覆盖率写入上下文。运维智能体监听部署就绪的消息负责将应用部署到沙箱环境并进行健康检查。每个智能体的“能力”就是它能调用的工具集Tools和内部逻辑。我们需要为智能体提供一套安全的、可描述的工具调用框架。例如使用LangChain的Tool抽象或类似自定义框架让LLM能知道“我现在可以调用哪些函数”并生成正确的调用参数。3.3 容错与自我修复机制在去中心化环境中智能体必须更加“健壮”。需要设计以下机制心跳与存活检测智能体可以定期向一个特定的上下文键如registry:heartbeat:agent_id写入时间戳。其他智能体或一个简单的监控器可以检查这些时间戳发现失联的智能体。任务超时与重分配在共享上下文中每个被认领的任务都应有一个超时时间戳和所有者。如果任务超时且所有者心跳消失该任务状态应被重置为“待处理”以便其他具有相同能力的智能体重新认领。决策日志与追溯将所有智能体的关键决策特别是LLM的输入和输出记录到上下文的history中。当出现意外结果时可以通过回溯历史来诊断是哪个环节的决策出了问题这对于调试和优化系统至关重要。4. 实战构建从零搭建一个简易自动化运维协作系统理论说得再多不如动手搭一个。让我们设想一个场景构建一个去中心化的智能运维系统包含一个“监控告警智能体”和一个“故障修复智能体”。当监控 agent 发现服务响应时间超标时自动触发修复 agent 进行问题分析和尝试缓解。4.1 环境准备与依赖安装我们选择Python作为实现语言因为它有丰富的AI和网络库。核心依赖如下# 基础框架与通信 pip install redis4.5.0 # 用于共享存储和Pub/Sub pip install openai1.0.0 # 或 anthropic, 用于LLM调用 pip install langchain0.1.0 # 可选但能极大简化Agent框架搭建 pip install pydantic2.0 # 用于数据验证和设置管理 # 工具调用相关示例 pip install requests2.31.0 # 用于调用外部API如重启服务 pip install psutil5.9.0 # 用于获取本地系统信息模拟监控我们需要一个运行中的Redis实例。可以通过Docker快速启动docker run -d -p 6379:6379 --name dmas-redis redis:7-alpine4.2 定义共享上下文管理器首先我们创建一个类来封装所有与Redis的交互确保读写的一致性和安全性。import json import redis import uuid from typing import Any, Dict, List, Optional from datetime import datetime, timezone class SharedContextManager: def __init__(self, redis_url: str redis://localhost:6379): self.redis_client redis.from_url(redis_url, decode_responsesTrue) self.session_prefix context:session def create_session(self, goal: str, agent_ids: List[str]) - str: 创建一个新的协作会话上下文 session_id str(uuid.uuid4()) session_key f{self.session_prefix}:{session_id} now datetime.now(timezone.utc).isoformat() # 使用管道保证原子性 pipe self.redis_client.pipeline() pipe.hset(session_key, mapping{ id: session_id, goal: goal, created_at: now, status: active, agents: json.dumps(agent_ids) }) # 初始化历史记录和状态列表 pipe.rpush(f{session_key}:history, json.dumps({ event: session_created, data: {goal: goal}, timestamp: now })) pipe.hset(f{session_key}:state, mapping{ current_phase: initialized, last_updated: now }) # 设置整个会话24小时过期 pipe.expire(session_key, 86400) pipe.expire(f{session_key}:history, 86400) pipe.expire(f{session_key}:state, 86400) pipe.execute() return session_id def update_state(self, session_id: str, updates: Dict[str, Any]) - bool: 使用乐观锁更新会话状态防止并发冲突 state_key f{self.session_prefix}:{session_id}:state self.redis_client.watch(state_key) # 开始监视 current_state self.redis_client.hgetall(state_key) current_state.update(updates) current_state[last_updated] datetime.now(timezone.utc).isoformat() # 在事务中执行更新 pipe self.redis_client.pipeline() try: pipe.multi() pipe.hset(state_key, mappingcurrent_state) result pipe.execute() return bool(result) except redis.WatchError: # 如果键被其他客户端修改则捕获异常调用方可以选择重试 print(fState key {state_key} was modified, update aborted.) return False def add_to_history(self, session_id: str, event: str, data: Dict[str, Any]): 向会话历史添加一条事件记录 history_key f{self.session_prefix}:{session_id}:history entry { event: event, data: data, timestamp: datetime.now(timezone.utc).isoformat() } self.redis_client.rpush(history_key, json.dumps(entry)) # 每次添加都刷新过期时间保持活跃会话 self.redis_client.expire(history_key, 86400) def get_recent_history(self, session_id: str, limit: int 10) - List[Dict]: 获取最近的会话历史 history_key f{self.session_prefix}:{session_id}:history raw_entries self.redis_client.lrange(history_key, -limit, -1) return [json.loads(entry) for entry in raw_entries] # 初始化全局上下文管理器 context_manager SharedContextManager()这个管理器提供了创建会话、安全更新状态和记录历史的基本能力。update_state方法中的乐观锁是保证数据一致性的关键。4.3 实现监控告警智能体这个智能体模拟监控系统定期“检查”指标并在发现问题时发布告警消息。import time import random import threading from typing import Dict class MonitoringAgent: def __init__(self, agent_id: str, context_manager: SharedContextManager): self.agent_id agent_id self.ctx context_manager self.redis context_manager.redis_client self.pubsub self.redis.pubsub() # 订阅修复完成的消息以便关闭告警 self.pubsub.subscribe(**{alert.resolved: self.handle_resolution}) self.thread threading.Thread(targetself.pubsub.run_in_thread, daemonTrue) self.thread.start() def simulate_monitoring(self, session_id: str): 模拟监控循环 while True: time.sleep(5) # 每5秒检查一次 # 模拟获取响应时间指标 (毫秒) simulated_response_time random.randint(50, 300) print(f[Monitor {self.agent_id}] 检测到响应时间: {simulated_response_time}ms) if simulated_response_time 200: # 假设阈值是200ms alert_data { alert_id: str(uuid.uuid4()), session_id: session_id, metric: response_time, value: simulated_response_time, threshold: 200, timestamp: datetime.now(timezone.utc).isoformat() } # 1. 将告警记录到共享上下文 self.ctx.add_to_history(session_id, high_latency_alert, alert_data) # 2. 更新当前状态表明出现问题 self.ctx.update_state(session_id, {current_phase: alerting, issue: high_latency}) # 3. 发布告警消息到消息总线 self.redis.publish(alerts.high_latency, json.dumps({ sender: self.agent_id, session_id: session_id, data: alert_data })) print(f[Monitor {self.agent_id}] 已发布高延迟告警: {alert_data[alert_id]}) def handle_resolution(self, message): 处理修复完成的消息 data json.loads(message[data]) if data.get(resolution) success: session_id data[session_id] print(f[Monitor {self.agent_id}] 收到修复成功通知关闭告警状态。) self.ctx.update_state(session_id, {current_phase: normal, issue: None})这个智能体独立运行持续检查发现问题后通过“更新上下文”和“发布消息”两种方式通知系统体现了去中心化的通信。4.4 实现故障修复智能体这个智能体订阅告警收到后进行分析调用LLM并尝试执行修复动作。from openai import OpenAI # 假设使用OpenAI API class RemediationAgent: def __init__(self, agent_id: str, context_manager: SharedContextManager, llm_client): self.agent_id agent_id self.ctx context_manager self.llm llm_client self.redis context_manager.redis_client self.pubsub self.redis.pubsub() self.pubsub.subscribe(**{alerts.high_latency: self.handle_alert}) self.thread threading.Thread(targetself.pubsub.run_in_thread, daemonTrue) self.thread.start() def handle_alert(self, message): 处理高延迟告警 data json.loads(message[data]) session_id data[session_id] alert_data data[data] print(f[Remediator {self.agent_id}] 收到告警: {alert_data[alert_id]}) # 1. 从共享上下文中获取相关历史信息构建完整上下文 recent_history self.ctx.get_recent_history(session_id, limit5) # 将历史转换为LLM可读的文本 history_text \n.join([f{h[timestamp]}: {h[event]} - {h[data]} for h in recent_history]) # 2. 调用LLM进行分析和决策 prompt f 你是一个运维专家。系统当前出现高延迟告警。 告警详情{alert_data} 近期系统历史 {history_text} 请分析可能的原因并给出一个具体的修复动作。你的回答必须是严格的JSON格式 {{ analysis: 简短的原因分析, action: 要执行的动作名称只能是 [restart_service, scale_out, clear_cache] 中的一个, parameters: {{}} // 动作参数如服务名 }} try: response self.llm.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}], temperature0.2 ) decision json.loads(response.choices[0].message.content) print(f[Remediator {self.agent_id}] LLM决策: {decision}) except Exception as e: print(fLLM调用失败: {e}) decision {action: restart_service, analysis: Fallback due to LLM error} # 3. 执行决策 result self.execute_action(decision[action], session_id) # 4. 将行动结果记录到上下文 self.ctx.add_to_history(session_id, remediation_action, { agent: self.agent_id, decision: decision, result: result }) # 5. 发布修复结果消息 self.redis.publish(alert.resolved, json.dumps({ sender: self.agent_id, session_id: session_id, resolution: success if result.get(success) else failed })) def execute_action(self, action: str, session_id: str) - Dict: 模拟执行修复动作 # 这里是模拟真实环境会调用Ansible、Kubernetes API、云服务商SDK等 actions { restart_service: {command: sudo systemctl restart myapp, success: True}, scale_out: {command: kubectl scale deployment myapp --replicas1, success: True}, clear_cache: {command: redis-cli FLUSHALL, success: True} } chosen actions.get(action, {command: unknown, success: False}) print(f[Remediator {self.agent_id}] 执行动作: {chosen[command]}) time.sleep(2) # 模拟执行耗时 # 更新状态表明正在修复 self.ctx.update_state(session_id, {current_phase: remediating, remediation_action: action}) return chosen这个智能体展示了如何利用共享上下文中的历史信息来辅助LLM决策并根据决策执行具体操作最后将结果反馈给系统。4.5 系统集成与运行最后我们编写一个主程序来启动整个协作系统。import signal import sys def main(): # 1. 初始化 ctx_mgr SharedContextManager() llm_client OpenAI(api_keyyour-api-key) # 请替换为你的API Key # 2. 创建协作会话 session_id ctx_mgr.create_session( goal自动监控并修复服务性能问题, agent_ids[monitor_01, remediator_01] ) print(f协作会话已创建ID: {session_id}) # 3. 启动智能体 monitor MonitoringAgent(monitor_01, ctx_mgr) remediator RemediationAgent(remediator_01, ctx_mgr, llm_client) # 4. 启动监控循环在新线程中 import threading monitor_thread threading.Thread(targetmonitor.simulate_monitoring, args(session_id,), daemonTrue) monitor_thread.start() print(系统已启动。监控Agent正在运行修复Agent已就绪。) print(按 CtrlC 终止程序。) # 保持主线程运行 def signal_handler(sig, frame): print(\n正在关闭系统...) sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.pause() if __name__ __main__: main()运行这个程序你会看到监控智能体模拟检测当“响应时间”超标时它会发布告警。修复智能体接收到告警后会结合上下文历史询问LLM得到修复建议如重启服务并模拟执行。整个过程两个智能体通过Redis的共享上下文和Pub/Sub频道进行协作没有中心调度器。5. 避坑指南从Demo到生产的关键挑战上面我们实现了一个可运行的Demo但要将去中心化多智能体系统投入生产还有一系列严峻的挑战需要克服。这些“坑”是我在类似项目中真实遇到的。5.1 共享上下文的爆炸与信息过载在Demo中我们只记录了少量历史。但在真实场景中智能体间的对话、中间产物、状态变更会非常多。如果所有内容都无差别地存入共享上下文很快就会导致存储压力剧增Redis内存被快速撑满。检索效率低下智能体每次都需要从海量历史中筛选相关信息延迟高。LLM上下文长度限制在构建提示词时我们无法将全部历史喂给LLM。解决方案分层与摘要策略分层存储将上下文分为“热数据”和“冷数据”。热数据如当前任务状态、最近10条关键历史放在Redis中。冷数据完整的历史日志、大型产物文件转移到对象存储如S3或时序数据库中只在需要时按索引查询。自动摘要定期或当历史达到一定长度时启动一个后台的“摘要智能体”它的任务就是读取近期详细历史调用LLM生成一段简洁的摘要例如“过去一小时内团队完成了需求分析确定了API接口并开始了模块A的开发当前在联调阶段遇到环境配置问题”然后用这个摘要替换掉旧的详细记录并保留一个指向冷存储的指针。这样后续的智能体在理解上下文时可以先读摘要有必要再深入细节。基于向量的语义检索对于存储的对话历史、文档等文本内容可以使用向量数据库如Chroma、Weaviate进行嵌入存储。当智能体需要寻找相关信息时不是进行关键词匹配而是将当前问题也转化为向量进行语义搜索找出最相关的几条历史记录。这能极大提升信息检索的精准度。5.2 智能体间的冲突与死锁在去中心化系统中多个智能体可能对同一资源或任务产生竞争。例如监控智能体告警“数据库连接池满”修复智能体A决定“重启数据库”而修复智能体B同时决定“扩容连接池”。两者同时执行可能导致更严重的问题。解决方案分布式锁与协商协议基于共享上下文的分布式锁对于关键资源如“执行数据库修复”在行动前智能体必须尝试在上下文中设置一个锁。例如执行SETNX lock:database_remediation agent_id EX 30。如果设置成功获得锁可以执行操作如果失败则等待或放弃。操作完成后必须删除锁或等待其自动过期。简单的协商协议对于任务分配可以采用“投标-确认”机制。当一个任务出现在上下文中时多个有能力的智能体可以“投标”在任务下添加自己的ID和方案。一个简单的“协调者”智能体可以是轮值的或一套规则如选择最先投标的来决定由谁执行并将中标者信息更新到上下文中。这比完全无协调的竞争更有序。5.3 LLM决策的不可控与幻觉我们的修复智能体依赖LLM做决策。但LLM可能会产生不切实际、甚至危险的指令如“删除整个数据库以提升速度”。解决方案严格的行动沙盒与验证层工具权限管控不要给智能体所有工具的完全访问权限。根据智能体的角色定义其可用的工具白名单。例如“修复智能体”只能调用重启服务、扩容实例、清除应用缓存等安全操作绝不能直接获得rm -rf /或DROP DATABASE的权限。决策后验证在LLM输出决策后、执行前增加一个“验证”步骤。这个验证可以是一套简单的规则引擎检查动作是否在白名单内参数是否在合理范围也可以是一个专门的“审核智能体”进行快速复核。只有验证通过的决策才会被真正执行。人类在环Human-in-the-loop对于高风险操作系统不应自动执行而应将LLM的建议和上下文信息推送给人类操作员审批。可以在共享上下文中将任务状态置为awaiting_human_approval并发送通知。5.4 系统的可观测性与调试当多个智能体自主运行时系统会变得像一个黑盒。一个问题出现时很难追踪是哪个智能体、基于什么信息、做出了什么错误决策。解决方案贯穿始终的链路追踪与审计日志唯一链路ID为每一个用户请求或外部触发事件生成一个唯一的trace_id。这个trace_id需要贯穿整个系统生命周期被记录在每一条消息、每一次上下文更新、每一个LLM调用中。结构化日志每个智能体不仅要将关键动作写入共享上下文的历史还应输出结构化的本地日志可收集到ELK或Loki等系统包含trace_id、agent_id、action、input_snapshot、output、timestamp。可视化看板构建一个简单的看板实时展示所有活跃会话、智能体状态、共享上下文的关键字段以及最新的历史事件。这能让你一眼看清系统的整体运行状况。可以使用Grafana连接Redis数据源来实现。去中心化多智能体系统与共享上下文是一个强大但复杂的范式。它放弃了中心控制的便利换来了系统的韧性、灵活性和更接近生物组织的协作智能。从Demo到生产核心在于处理好“状态同步”、“冲突协调”和“决策安全”这三座大山。我个人的体会是起步时一定要克制从最小的可行场景就像上面的运维告警开始清晰地定义每个智能体的边界和协议并投入精力构建强大的可观测性工具。只有这样你才能在享受其自动化红利的同时不至于迷失在智能体们自己创造的混沌之中。