简介这是一份以“分布式人工智能与Agent”为主题的PPT教学资料适合人工智能、多智能体系统相关课程的学习者与授课教师使用。内容系统梳理了分布式人工智能DAI的三大分支——分布式问题求解DPS、多Agent系统MAS与并行人工智能PAI并详细讲解了Agent的强弱定义、思考型/反应型/混合型分类以及MAS的组织形成与协调机制。资源为单个pptx文件压缩包大小约295KB方便直接下载使用。目前已有112人学习浏览。通过这份PPT读者可以快速建立对DAI的整体认知掌握从传统AI到多Agent系统的核心概念与典型方法适用于课堂演示、自学复习或备课参考。1. 分布式人工智能与Agent先解决“分布式”还是先解决“Agent”面对《分布式人工智能与Agent.pptx》大部分IT从业者第一反应是去翻Agent框架但真正需要先想清楚的是你把Agent放在分布式系统的哪一个位置。分布式人工智能解决的是算力、数据、状态在多个节点上如何被统一调度Agent解决的是意图理解、任务拆解和工具调用。两者的交集不是“Agent本身”而是Agent运行时如何依赖无中心、可容错、可水平扩展的基础设施。若把Agent当成普通API直接暴露几百个并发请求就会打满单机内存若先把分布式底座堆起来却让每个Agent独占一个线程又会出现资源碎片化。常见做法是让分布式AI提供模型服务的弹性伸缩以及任务与状态的分布式编排。我会以Ray为落地示例从架构选型讲到可直接运行的Actor方案适合设计Agent平台、做技术评估或正在整理Agent开发学习路线的人。2. 分布式人工智能与Agent的核心架构Agent框架与通信底座怎么选2.1 分布式AI的两条主线训练与推理对Agent完全不同分布式AI的传统叙事围绕模型训练展开用数据并行把Batch打散到多张GPU用模型并行把Transformer层切到多卡再用流水并行让不同batch在不同卡上重叠计算。但在Agent项目里绝大多数情况并不需要重新训练模型而是让多个Agent共享一套已部署的模型服务。这时分布式AI的重心从训练侧转到推理侧模型副本按请求量扩缩容Agent在容器启动时通过服务发现拿到模型Endpoint而不是在代码里硬编码IP。这种差异直接影响agent架构。训练侧可以容忍较高的批处理延迟Agent更适合作为数据生产端异步把数据推给训练集群推理侧则要求低延迟和流式返回Agent需要维护连接池并把一个用户请求拆成多个并行子请求。所以做分布式AI与Agent方案时先画清楚你的模型是离线计算还是在线服务再选框架否则很容易把训练管线和在线推理混在一个Agent里。我自己一般用Ray Serve托管推理因为它直接提供HTTP和gRPC入口也支持把模型副本数做成自动扩缩。Agent端并不直接调用模型函数的Python对象而是通过一个小型Client读取启动时注入的Endpoint信息这样Agent的重启不依赖模型副本的生命周期。2.2 Agent框架为何必须依赖分布式底座现在主流的Agent框架解决的是编排问题把“规划-工具调用-记忆-反思”串成一个状态机。开箱即用时状态存在进程内字典工具调用是普通函数执行多Agent协作则通过asyncio事件循环模拟。也就是说单机Agent框架默认假设所有Agent节点都在同一个进程里。这个假设在原型阶段没问题一旦Agent数量超过几十个或者在路由识别节点上频繁做LLM调用立刻会遇到四个瓶颈GIL限制并发、进程内状态无法共享、单个Agent抛异常拖垮整个进程、无法利用多台机器上的异构资源。分布式底座给Agent框架补齐的正是这四块进程隔离让每个Agent拥有独立内存Object Store让大模型输入输出在节点间高效流动Actor模型让有状态的Agent可以被调度到任意节点命名空间和服务发现让多个Agent相互定位。一个比较务实的组合是Agent框架负责编排算法Ray负责Actor生命周期和调度Redis负责短期记忆同步Kafka负责事件广播。下表列出单体Agent和分布式Agent的关键差异方便做技术选型时直接引用。维度单体Agent分布式Agent并发上限受进程内线程与GIL限制受集群总CPU/内存限制故障范围一个未捕获异常可能中断所有Agent单Actor崩溃可自动重启状态保存进程内存重启丢失可外置到Redis或对象库调试难度本地断点即可需要追踪跨节点调用链模型调用函数内直接import通过Ray Serve或gRPC访问模型副本需要留意的是不要因为选了Ray就放弃Agent框架。现在的常见结构是Agent框架生成规划图随后把每个节点的执行体推给Ray Worker执行。这样Agent框架的循环、判断、条件分支仍然在编排层可见而真正的算力消耗被隔离到Worker里。很多团队把这一层称为“Agent执行器”它会从框架拿到节点ID和输入数据再通过Ray的ObjectRef获得上游结果。2.3 通信底座选型gRPC、Ray和消息队列的边界通信层是整个方案中最容易过度设计的地方。我的选型原则是控制面用gRPC数据面用Ray Object Store事件面用Kafka或Redis Stream。Agent心跳、任务分发、模型状态上报都用gRPC双向流因为这类消息量小但要求毫秒级响应消息队列则负责解耦不让每个Agent都持有下游Agent的地址大容量的图像、文档、模型输入输出放进Object Store避免在Agent之间复制整个对象。一个常见误解是把所有Agent通信都改成异步消息认为这样就不怕节点崩溃。其实异步消息会带来新的问题消息排序、重复投递和消费失败后的重试语义。Agent不像普通微服务那样天然幂等某个Agent处理任务的同时修改了外部系统重试可能导致双写。所以我在设计Agent通信时会给每条消息带task_id和幂等键接收端先查Redis中的去重集合再决定是否执行。如果计划直接用Ray搭建可以先在本地启动一个head节点观察资源视图# 启动Ray head节点--resources定义自定义资源标签 ray start --head --port6379 --dashboard-host0.0.0.0 --num-cpus4 --resources{agent_slot: 2}参数说明--port指定GCS服务端口客户端和Worker都通过它连接--num-cpus会覆盖物理核数用于测试资源调度策略--resources是自定义资源标签你可以给Agent和推理节点打上不同标签之后在num_cpus之外用resources{agent_slot: 1}约束任务落到指定节点。启动后访问dashboard-host对应的8265端口能看到每个Actor的CPU/内存占用这在排查Agent资源争用时必不可少。停止本地测试环境用ray stop。3. 用Ray搭建分布式Agent的首个可运行版本Actor模型与关键参数3.1 用Remote Actor定义AgentWorker明确了分布式底座后我们可以用Ray把Agent变成一个真正独立的分布式实体。Ray里对应Agent的原生抽象是Actor它和普通远程函数最大的区别是Actor拥有自己的状态每次方法调用都作用于同一个对象实例。这对于保存Agent会话记忆、工具调用次数、路由状态很有价值。下面这个脚本会创建4个AgentWorker每个Worker用列表保存自己的已处理任务然后并发处理8个任务import ray # 本地启动Ray等价于单机伪分布式 ray.init(namespaceai-agent) ray.remote(num_cpus0.5, max_restarts2) class AgentWorker: def __init__(self, agent_id: str): self.agent_id agent_id self.memory: list[str] [] def handle(self, task: dict) - dict: # 模拟一次Agent任务处理并更新本地记忆 result { agent: self.agent_id, task: task[name], status: done, } self.memory.append(task[name]) result[memory_size] len(self.memory) return result if __name__ __main__: workers [AgentWorker.remote(str(i)) for i in range(4)] tasks [{name: fjob-{i}} for i in range(8)] futures [ workers[i % len(workers)].handle.remote(task) for i, task in enumerate(tasks) ] print(ray.get(futures))代码逻辑分成三步AgentWorker.remote(str(i))实际上不是创建Python对象而是向调度器提交一个创建Actor的请求返回的workers[i]是一个ActorHandle随后调用.handle.remote(task)产生一个Future任务会被派发到该Actor的真实进程里执行修改的是Actor内部的self.memory最后的ray.get(futures)会阻塞直到所有任务返回。这里特别注意ray.init(namespaceai-agent)命名空间让多个Python进程可以按名字找到同一个Actor。如果集群中有多个业务比如数据分析Agent和画图Agent建议各自使用不同namespace避免路由识别节点把任务发给错误的对象。3.2 调好num_cpus、max_restarts和max_task_retriesRay Actor装饰器上的参数会直接影响Agent的密度和稳定性。很多Agent项目把num_cpus设为1但实际CPU消耗并不均匀空闲等待模型输出时CPU几乎为0解析大段JSON时会瞬间打满。建议根据任务类型调整配额而不要一律给1个CPU。参数默认值推荐值暴露出的问题num_cpus10.5或1值太大导致节点Actor数过少太小会被频繁调度num_gpus00Agent本身不占用GPU推理交给Ray Servemax_restarts02不设置则Actor崩溃后任务永久丢失max_task_retries01或3任务不幂时时重试会造成重复副作用max_pending_calls10010超过后新的调用会阻塞或报错lifetimeNoneNone会话型Agent可以设置idle_timeoutmax_restarts和max_task_retries是两个容易混淆的参数。max_restarts是Actor进程级重启进程崩溃后重新执行__init__所以Agent的self.memory会清空这要求Agent在初始化时从Redis恢复记忆max_task_retries是单个任务级别重试会重新执行handle方法如果handle里修改了外部数据库则需要保证幂等性。我的经验是Agent在处理外部系统调用前先记录任务状态处理完再更新这样即使崩溃重放也能识别出已完成任务。3.3 把Agent Skill封装成远程任务而不是内部函数所谓Agent Skill是指可以被多个Agent复用的能力单元例如查询数据库、生成图表、调用搜索引擎。单机开发时Skill只是import进来的函数但在分布式Agent里如果Skill仍作为Agent类内部方法那么每个Agent都要持有对应的依赖和连接池既是内存浪费也会让热更新变得困难。更合理的做法是把Skill拆成独立的Ray远程函数或独立服务Agent只是Skill的一个调用方。下面代码演示如何把SQL查询Skill放到远端ray.remote(num_cpus1, max_retries1) def run_sql_skill(conn_str: str, query: str) - dict: import sqlite3 conn sqlite3.connect(conn_str) try: rows conn.execute(query).fetchall() return {rows: rows[:100], count: len(rows)} finally: conn.close() # 在Agent内部通过ObjectRef获取结果 ref run_sql_skill.remote(local.db, SELECT * FROM tasks) rows ray.get(ref, timeout10)这段代码里conn_str和query都被序列化成任务参数意味着Skill可以运行在任意一台有数据库依赖的节点上而不是必须和Agent同一台机器。timeout10是防止SQL查询长时间占用Agent线程max_retries1表示如果节点被抢占任务会重新调度但这里也隐含风险——只读SQL可以安全重试写入型SQL必须带幂等ID并在Skill内部做去重。把Skill做成远程函数后另一个好处是可以用Ray的任务级超时控制单个工具调用避免Agent轮询等待不可达的第三方API。4. 多Agent协作与分布式状态同步memory、Skill和路由识别节点4.1 多Agent协作的两种主流编排集中式Router与SEDA多Agent协作的第一步是确定任务由谁处理。如果任务类型只有两三种常见做法是用一个路由识别节点做集中式分发。这个节点可以是一个精细编写的规则引擎也可以是一个调用LLM的Agent。集中式的好处是决策路径清晰所有任务只经过Router一个入口合规审计比较容易缺点是Router本身会成为单点和性能瓶颈一旦路由节点崩溃整个任务流就停摆。另一类是SEDA分阶段事件驱动架构常用于请求量大、处理流水线固定的场景。每个Agent只监听自己感兴趣的消息主题处理完把结果写回下游主题没有全局调度器。它的扩展性更好某个阶段实例挂了还有其他消费者接着消费缺点是缺少全局视图任务卡在哪一段需要通过消息积压量来推断。下面的表可以快速帮你取舍。方式优点缺点适用场景Router Agent流程可控路由规则可审计单点风险路由节点负载高任务分类明确流程可配置SEDA消息驱动高吞吐天然削峰调试难需要监控每个Topic多阶段流水线数据量大从我的实践经验看大多数Agent项目应该先使用Router方式把路由逻辑做成一个独立Actor并且把路由结果记录到Redis。等容量上来了再把规则按业务拆成多个专用Agent用消息队列连接起来。不要一开始就追求去中心化去中心化通常是把调试成本往后移。下面是一个最基础的路由识别节点用规则检测输入文本中的关键词class RouterAgent: def __init__(self): # 路由表key代表业务类型value是下游Agent名 self.route_table { data: data-agent, plot: plot-agent, review: review-agent, } def detect_task(self, text: str) - str: # 规则优先后续可以叠加分类模型做兜底 if 画图 in text or plot in text.lower(): return self.route_table[plot] if 查询 in text or select in text.lower(): return self.route_table[data] return self.route_table[review]这段代码可以作为路由识别节点的雏形但实际生产里关键词匹配很容易被说法变化击穿。我一般会保留这个规则作为快速通道同时再用一个小的分类模型或LLM函数调用兜底。重点在于route_table最好从配置中心或Redis读取这样调整路由不需要重启RouterAgent。4.2 Agent记忆的分布式存储从Actor内存搬到Redis Stream在上一章的AgentWorker里self.memory只属于单个Actor重启即丢失。多Agent协作时A Agent产生的中间结论往往需要被B Agent看到因此记忆必须放到外部存储。短期记忆用Redis Stream比较顺手它同时提供消息持久化、消费组和延迟读取和Agent的异步事件模型天然匹配。import redis r redis.Redis(hostredis-agent, port6379, decode_responsesTrue) def save_memory(agent_id: str, payload: str): # 每个Agent一个Stream限制长度防止无限增长 r.xadd(fmemory:{agent_id}, {data: payload}, maxlen500) def read_memory(agent_id: str, last_id: str $): # $表示只读取新写入的事件适合等待Agent记忆更新 events r.xread({fmemory:{agent_id}: last_id}, block0, count20) return eventsxadd把一条记忆追加到Stream末尾maxlen500是近似长度限制超过后会自动淘汰最旧的记录xread中的block0会一直阻塞直到有新的记忆到达。注意生产环境不要直接block0应该设置block3000毫秒以便Agent在等待期间可以做其他事情也避免连接长期被占用。长期记忆建议放进向量数据库短期上下文放进Redis两者之间用task_id关联。4.3 让Agent Skill与Memory串成一条可观测的执行链多Agent协作不能只考虑通信还要把记忆和技能串在一条请求链路里。我通常会让Router节点生成一个trace_id然后把这个ID传给后续所有Agent。每个Agent在调用Skill前读取共享记忆在得到结果后写回记忆最后把当前状态交给下游。这样即使某个Agent状态丢失也能根据trace_id从Redis恢复整个会话。async def run_agent_pipeline(user_input: str, trace_id: str): # 先路由得到目标Agent route router.detect_task(user_input) worker worker_pool[route] # 读取该Agent的近期记忆 before await read_memory(worker.agent_id) # 执行Agent主任务同时传入上游上下文和链路ID result await worker.run(taskuser_input, contextbefore, trace_idtrace_id) # 处理结果写回记忆方便后续Agent和排障使用 await save_memory( worker.agent_id, ftrace{trace_id} input{user_input} - {result}, ) return result这段伪代码展示了执行链的关键约束worker.run不只接收当前输入还接收上游记忆片段和trace_id写入记忆的结果包含trace标识方便后续排查“为什么A Agent给出了这个决定”。实际落地时worker_pool可以是一个dictvalue是Ray ActorHandle而worker.run内部再调用独立的Skill远程函数。这样Agent、Skill、Memory三者都是分布式对象任何一层都能独立扩展或替换。5. 生产环境验证用Ray命令和链路追踪给分布式Agent做检查5.1 从资源状态里确认Agent没有“幽灵占用”分布式Agent部署完先不要急着压测优先看Ray集群的资源分配情况。Ray自带三条命令分别检查节点状态、对象存储和Dashboard# 检查集群节点与Actor资源占用 ray status --address127.0.0.1:6379 # 检查ObjectRef在内存和磁盘上的占用 ray memory --address127.0.0.1:6379 # 打开可视化Dashboard ray dashboard --address127.0.0.1:6379ray status输出每个节点的CPU/GPU使用量以及当前排队的任务数ray memory能看到ObjectRef占用磁盘和内存的比例用于排查Agent生成的大对象是否被及时释放dashboard则能展示所有Actor的存活时间和调用次数。这三条命令暴露的常见问题有两个一是Actor设置了max_restarts但初始化失败进程反复重启却不退出表现为节点CPU异常升高二是Agent的返回值长时间持有大对象ray memory会持续增长。解决办法是给Agent的调用超时加一个上限并在返回给用户之前主动del大对象再gc.collect()。另外需要检查Agent的内存是否无界增长。很多Agent把对话历史放在list里每轮append最终导致OOM。可以用一个初始化时读取的max_memory_items来控制list长度同时在save_memory时记录最近N条摘要而不是保存全部原文。5.2 用trace_id把Agent调用链串起来想快速定位“任务在哪个Agent上卡住”需要给每个请求绑定trace_id。简单做法是在Router入口生成UUID用Python标准logging以结构化字段输出每个Agent在调用Skill前后打印同一trace_id的事件。下面的代码片段展示了如何让日志字段可检索import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(agent) def log_event(trace_id: str, agent_id: str, message: str): # 关键日志统一带trace_id按ID即可检索整条链路 logger.info( message, extra{trace_id: trace_id, agent_id: agent_id}, )配合日志平台后用trace_id即可把Router、Agent、Skill的日志拼成时间线。需要注意的是在Ray的异步环境中同一个ActorHandle的多个并发任务会交替写日志因此extra里的agent_id不建议固定在__init__里而是从task数据里取否则多个并发任务会混在一起。更稳妥的做法是把JSON Formatter接入logging让日志平台自动解析trace字段。分布式Agent的生产验证远不是压测通过就结束。更理性的做法是每次发布前跑一轮故意延迟注入比如让某个Skill随机sleep 2秒观察路由节点是否按预期做超时降级。Agent安全也要纳入验证不同Agent使用独立的访问密钥Skill服务端只允许最小权限访问数据库所有外部调用都要有直接责任人。把这类故障注入脚本加进CI流程每次Agent框架版本升级时自动跑一遍能在多Agent协作问题爆发前先把脆弱点暴露出来。本文还有配套的精品资源点击获取