1. 内容整体设计与思路拆解先聊个现象。最近总有朋友在后台问我说一个人真的能管过来五六个 AI 员工吗听起来像个噱头但事实是我已经在本地跑了大半年类似架构稳定程度远超预期也踩过不少坑。今天这篇就拿标题里那套“OpenClaw 多 Agent 协作”展开讲讲我实际怎么搭、怎么调度、怎么避免它们互相打架的。先拆字面。OpenClaw 是底层框架多 Agent 是运行模式协作是核心目标。说白了OpenClaw 提供了一个类似“容器化”的环境你可以把不同职能的 Agent 封装成独立服务再通过消息队列、共享存储和统一调度中心把它们串起来。而“一个人管理 5 个 AI 员工”意思是这 5 个 Agent 各有各的角色定位比如一个负责抓取资料一个负责整理摘要一个负责写初稿一个负责审核校对还有一个做定时发布推送。你不需要同时盯着它们只需要在关键节点做决策、给反馈日常跑批完全是自动化的。为什么非要选多 Agent 而不是一个大模型一把梭这是我最初踩的坑。我当时用单一 Prompt 让模型同时完成“检索→分析→写作→发布”全链路结果总是顾此失彼要么摘要质量差要么写作风格漂移。原因很简单一个大模型要同时维护多种上下文和目标token 消耗高推理又慢还容易在任务切换时“失忆”。多 Agent 把大问题拆成小问题每个 Agent 只需要维护一份精简的系统提示词和工具列表各自专注一个环节最后通过协议对接既降低单点复杂度又让每个环节可以独立升级、替换甚至跑在不同机器上。另一个核心思路是“异步协作”而非“同步调用”。我见过很多所谓多 Agent 系统实际还是同步请求Agent A 处理完把结果丢给 Agent B卡住就全卡住。真正稳定的协作应该像公司部门一样各干各的把成果丢到公共空间比如数据库表、消息队列下一个环节发现“有活儿”就自动开始干。这样一来单个 Agent 崩了其他 Agent 不受影响等它恢复后再接着消费积压任务。OpenClaw 这类框架的好处就是把这种异步模型内置了你只需要配置 任务队列 和 结果回调不需要重复造轮子。至于“一个人管理”核心不是写代码而是定流程。你要把工作流拆成多细的颗粒度每个 Agent 的输入输出是什么、失败重试多少次、什么情况要升级人工这些规则越清晰你越省心。我在实际项目中大多数时间花在定义任务状态机和异常处理策略上真正写业务逻辑反而很快。这跟带团队很像你给每个人的职责越明确他们协作越顺畅你越不需要介入具体事务。2. 核心细节解析与实操要点先说框架选型。OpenClaw 只是一个代号市面上类似产物包括一些开源的 Agent 编排框架它们普遍提供四件事Agent 生命周期管理、消息路由、工具挂载、状态持久化。你如果自己硬写也不是不行但到第三四个 Agent 的时候就会因为重复造轮子而崩溃。所以我的建议是既然目标是一个人管 5 个 Agent就不要省基建成本。2.1 Agent 角色划分把工作拆成五份工我的配置里五个 Agent 分别是角色职责输入输出资料官多源检索、网页抓取、文档解析任务关键词结构化条目清单分析官信息抽取、观点归纳、数据清洗原始条目主题摘要和关键事实写手按不同文体生成初稿摘要风格参数长文草稿审核官事实校验、去重、语气调整草稿修订后的终稿发布官格式转换、生成摘要、推送到目标平台终稿发布确认这五个角色不是拍脑袋定的。它的设计思路是每个角色的输出格式是明确的、可验证的。资料官输出的每条记录都包含标题、来源、时间、正文片段、置信度分析官必须输出固定 JSON写手输出的 Markdown 草稿带章节号审核官只能返回通过/不通过修改建议发布官只接收终稿并返回链接。这种强契约设计是减少协作混乱的第一道防线。实操的时候你不需要给每个 Agent 写“小作文”式的系统提示词。我建议每个 Agent 的系统提示词控制在 200 字以内重点声明三件事你是谁、你只能做什么、你的输出必带什么标记。比如审核官的提示词大致是“你是审核官。你只负责检查输入文本中的事实错误、逻辑冲突和格式问题。如果通过输出以 PASS 开头如果不通过输出以 REVISE 开头并给出具体修改点清单。”你给模型讲太多剧情反而会带偏。2.2 消息与任务流转别让它们直接对话这可能是最关键的经验。5 个 Agent 之间千万不要直接互相调用 API也就是 A 内部用代码去请求 B 的接口。一旦这么干耦合度瞬间拉满改一个 Agent 就得改其他四个。我用的是统一任务表 轮询消费的模式。具体来说每台机器上运行一个 PostgreSQL 实例建一张 tasks 表。表里的核心字段包括task_id、type表示由哪个 Agent 处理、statuspending、processing、done、failed、payloadJSON 内容、resultJSON 内容、retry_count重试次数、created_at创建时间和 updated_at更新时间。每个 Agent 是一个常驻进程每隔几秒就查一次“有没有我这个类型的 pending 任务”取出来处理写回结果再更新状态。举个例子资料官跑完一批抓取任务后把几十条结构化记录合并成一个大 JSON插入到 tasks 表type 填analyzerstatus 填pending。分析官轮询到新任务开始干活。这样写的好处是你一端看到 tasks 表就能完整掌握现在工作流卡在哪个环节哪个任务出错了。排障变得极其直观。有人会问既然用轮询实时性会不会差我的场景是内容批量生成延迟两三秒完全无所谓。如果你是强实时交互那建议换 RabbitMQ 或 Redis Stream。但对于个人管理 5 个 AI 员工的场景数据库表是最高性价比的选择因为它天然带着事务和回滚能力哪怕崩了重启还能继续跑。2.3 Prompt 参数与管理模板化 变量注入多 Agent 协作里的 Prompt 管理比单 Agent 场景要复杂一个量级。我这里用的是“基础模板 上下文变量”的方式所有 Prompt 都放在一个prompts目录里按角色命名比如writer.md。模板里不要写死具体的任务用占位符代替例如{summary}和{style}。运行时Agent 进程读取模板并替换变量然后再发给模型。这里有个细节变量内容里经常包含特殊字符比如 JSON 里的引号、换行直接拼进 Prompt 容易导致格式错乱。我先对变量做一次序列化转义再塞进模板。还有一个老坑——模型输出偶尔会重新包装 JSON把字段固定在output节点里。我的解决方案是让解析器做“宽容解析”优先找output节点找不到就尝试整个文本 json.loads还不行就触发一次重试并附带一条修正指令“请重新输出不要包含多余解释”。这套机制把写手和审核官的输出解析失败率从 20% 降到了 3% 以下。另外每个 Agent 至少要配置两个模型参数temperature和max_tokens。资料官和分析官用低 temperature比如 0.2确保提取结果稳定写手用 0.8 左右保留一定的文采变化审核官建议用 0.0严格按规则打分。这些参数写成配置文件就不要在代码里硬编码了。后期换模型提供方时只改配置不改逻辑。3. 实操过程与核心环节实现这一节我给你一个可以抄作业的落地全过程。我用一个模拟项目 X一个“每日行业情报摘要”自动生成系统为例完整展示从目录结构到运行监控的所有关键步骤。这个模拟项目 X 就是典型的“一个人管 5 个 AI 员工”场景——每一天5 个 Agent 协同产出 5 份不同领域的简报全程无人值守。3.1 环境准备与配置初始化环境层面我建议用 Docker Compose 而不是裸机部署。因为 5 个 Agent 的 Python 依赖版本不同容易出现冲突。Docker 能隔离环境还能统一调度资源。我的 Compose 文件里包含四个容器postgres数据库、redis缓存用于放去重标记和锁、worker同时运行五个 Agent 进程用 supervisor 管理、scheduler定时触发每日任务。这里注意我没有把五个 Agent 拆成五个容器是因为它们都是轻量级长驻进程拆开反而增加网络开销和调度复杂度。如果你想水平扩容某个 Agent再单独把它拆出来也不迟。端口和卷的映射我是这样配的version: 3.9 services: db: image: postgres:15 environment: POSTGRES_USER: agent POSTGRES_PASSWORD: agent_pass POSTGRES_DB: agent_flow volumes: - ./pgdata:/var/lib/postgresql/data ports: - 5432:5432 worker: build: . depends_on: - db environment: TASK_DB_DSN: postgresql://agent:agent_passdb:5432/agent_flow LLM_API_KEY: ${LLM_API_KEY} LLM_BASE_URL: ${LLM_BASE_URL} volumes: - ./prompts:/app/prompts - ./output:/app/output scheduler: build: . command: python scheduler.py depends_on: - db配置文件我用 YAML 写在config.yaml里内容包括五个 Agent 各自对应的模型服务、Prompt 模板路径、任务类型映射。核心片段agents: researcher: task_type: research prompt_file: prompts/researcher.md model: gpt-4o-mini temperature: 0.2 analyzer: task_type: analyze prompt_file: prompts/analyzer.md model: gpt-4o-mini temperature: 0.2 writer: task_type: write prompt_file: prompts/writer.md model: gpt-4o temperature: 0.8 reviewer: task_type: review prompt_file: prompts/reviewer.md model: gpt-4o-mini temperature: 0.0 publisher: task_type: publish prompt_file: prompts/publisher.md model: gpt-4o-mini temperature: 0.0这套配置是目前我在多个项目里反复调整后沉淀下来的。模型选型上没有所有 Agent 都用最强模型因为成本会爆炸。写手和审核官这类关乎最终质量的环节用能力强的模型抓取、解析、发布这些机械性工作用小模型完全够用。3.2 核心代码骨架Agent 基类与任务循环五个 Agent 不必各写一套运行逻辑。我抽象了一个BaseAgent类子类只需要实现process(payload)方法其他轮询、状态更新、错误重试全部复用。下面是简化版基类代码重点看任务循环怎么写import time, json, yaml from sqlalchemy import create_engine, text class BaseAgent: def __init__(self, config, agent_name): self.agent_name agent_name self.task_type config[task_type] self.prompt_template open(config[prompt_file]).read() self.engine create_engine(config[db_dsn]) self.retry_limit 3 def run_forever(self): while True: task self._fetch_pending_task() if task is None: time.sleep(2) continue try: result self.process(task[payload]) self._complete_task(task[id], result) except Exception as e: self._fail_task(task[id], str(e)) def _fetch_pending_task(self): sql text( SELECT id, payload FROM tasks WHERE type :task_type AND status pending ORDER BY created_at ASC FOR UPDATE SKIP LOCKED LIMIT 1 ) with self.engine.connect() as conn: row conn.execute(sql, {task_type: self.task_type}).fetchone() if row is None: return None # 将任务状态改为 processing避免重复消费 conn.execute(text(UPDATE tasks SET statusprocessing, updated_atnow() WHERE id:id), {id: row.id}) conn.commit() return {id: row.id, payload: json.loads(row.payload)} def _complete_task(self, task_id, result): with self.engine.connect() as conn: conn.execute(text( UPDATE tasks SET statusdone, result:result, updated_atnow() WHERE id:id ), {id: task_id, result: json.dumps(result)}) conn.commit()这里有个细节查询 pending 任务时用了FOR UPDATE SKIP LOCKED。因为五个 Agent 进程同时在跑如果不用锁两个轮询可能拿到同一个任务重复处理。SKIP LOCKED让数据库自动跳过被其他事务锁定的行天然解决了并发消费问题。这是我在并发测试踩坑后学到的一开始没加锁偶尔会出现同一篇简报生成两次。process方法的内部逻辑是从 payload 读取输入数据从模板渲染 Prompt调用大模型接口再按角色要求做后处理。写手 Agent 的process大致长这样class WriterAgent(BaseAgent): def process(self, payload): summary payload[summary] style payload.get(style, default) prompt self.prompt_template.format(summarysummary, stylestyle) response call_llm(prompt, temperature0.8, max_tokens3000) markdown_text extract_markdown(response) # 校验标题和章节是否缺失 if not markdown_text.strip(): raise ValueError(Writer output empty) return {draft: markdown_text}call_llm是我封装的一个函数统一了不同模型的调用协议。它会先读环境变量里的 API Key 和 Base URL然后构造请求。这里还要做一个简单限流防止五路并发把账户额度打爆import time, threading _rate_lock threading.Lock() _last_request_time {} def call_llm(prompt, **kwargs): with _rate_lock: now time.time() last _last_request_time.get(kwargs.get(model, default), 0) wait max(0, 1.1 - (now - last)) if wait 0: time.sleep(wait) _last_request_time[kwargs.get(model, default)] time.time() # 真正请求逻辑区域 return request_llm(prompt, **kwargs)这个锁是为了保证同一模型的请求间隔不低于 1 秒。别小看这个限流我曾经因为没加某天批处理 200 条任务直接被限频整个流程中断。加了这个全局锁之后虽然速度慢了点但胜在稳定。3.3 调度器与任务初始化每日自动开工调度器负责生成最初的任务种子。场景是每天 9 点系统自动创建 5 个research类型任务每个任务对应一个行业关键词。我写了一个简单的scheduler.pyfrom datetime import datetime from sqlalchemy import create_engine, text import json TOPICS [新能源电池, 大模型推理优化, Web3钱包安全, 生物医药AI, 自动驾驶传感器] def create_daily_seed(): engine create_engine(postgresql://agent:agent_passlocalhost:5432/agent_flow) now datetime.now().isoformat() with engine.connect() as conn: for topic in TOPICS: conn.execute(text( INSERT INTO tasks (type, status, payload, created_at, updated_at) VALUES (research, pending, :payload, :now, :now) ), {payload: json.dumps({topic: topic, requested_date: now}), now: now}) conn.commit()如果你不想定时器依赖 crontab可以把这个脚本挂到一个常驻调度循环里每隔 60 秒检查当前时间是否到了 9 点且今天的任务还没建就执行一次。我用的是 APScheduler 库比 crontab 更容易动态修改时间而不需要重启容器。调度器只创建任务种子后续流程像流水线一样自动推进资料官消费 research 任务产出原始条目分析官消费 analyze 任务产出摘要写手消费 write 任务产出草稿审核官消费 review 任务产出终稿发布官消费 publish 任务产出发布链接。每个环节完成后都会把结果写回 tasks 表然后在下游插入一条新任务。这样一来整个工作流的每一步都有记录你随时能翻账。3.4 失败重试与升级机制如何让它自己站起来一个人管理多个 Agent最大的困扰就是半夜任务挂了没人处理。我的做法是分级失败处理机制。每个任务都有retry_count默认为 0。处理任务时如果抛出异常_fail_task会做判断如果retry_count 2就把状态重置为pending并且将retry_count加 1如果已经重试两次则状态改为failed并生成一条alert任务交给“通知 Agent”去推给 IM webhook。通知 Agent 很简单就是一个发 HTTP 请求的机器人负责把失败的 task_id、错误信息发送到你的飞书或钉钉群。这样你不需要盯着控制台只需要在群里等消息。下面这段代码就是_fail_task的关键逻辑def _fail_task(self, task_id, error_msg): with self.engine.connect() as conn: row conn.execute(text(SELECT retry_count FROM tasks WHERE id:id), {id: task_id}).fetchone() if row.retry_count self.retry_limit: conn.execute(text( UPDATE tasks SET statuspending, retry_countretry_count1, last_error:err, updated_atnow() WHERE id:id ), {id: task_id, err: error_msg}) conn.commit() else: conn.execute(text( UPDATE tasks SET statusfailed, last_error:err, updated_atnow() WHERE id:id ), {id: task_id, err: error_msg}) conn.commit() self._notify_alert(task_id, error_msg)这套重试机制要特别注意不是所有异常都应该重试。比如因为 API key 过期导致的认证失败重试多少次都没用反而会浪费调用额度。所以我在异常处理里会区分异常类型对于配置型错误直接标记失败并告警对于临时网络错误、超时才进入重试流程。具体做法是自定义一个NonRetryableError类抛这个类的异常时不重试。3.5 审计与追溯所有数据可回看多 Agent 协作经常会遇到“这篇稿子为什么写得这么烂”的疑问。如果系统没有审计能力你根本不知道是哪一环节出了问题。我的方案是tasks 表里每一个任务都保留原始 payload 和 result同时在当天的任务批次上挂一个batch_id这样你可以把所有相关任务串联起来。我在模拟项目 X 的数据库中每天会自动生成一个 batch_id格式是20250321_0900。调度器建种子任务时把这 5 个任务都打上同一个 batch_id。后续项目产生的所有任务也都继承这个 batch_id。前端想追溯某一篇简报的生成记录时只要按 batch_id 查询就能看到这一天的资料官抓了哪些链接、分析官抽了什么要点、写手出了什么草稿、审核官改了哪些语句、发布官在哪个平台发布了什么版本。整个链路透明化这对后期优化单个 Agent 非常有帮助。如果你嫌数据库查询不够直观我还在服务端加了一个简单的管理页面。用一个轻量级 Web 框架展示 tasks 表按状态和类型筛选点击某条任务可以看到 payload 和 result 的 JSON。这个页面不用做得复杂关键是可以快速定位“哪个环节卡住了”。偶尔某个任务一直 pending很可能是上一次进程崩溃导致状态变成了 processing 但没有恢复。我加了一个兜底脚本定期把处理时间超过 30 分钟的 processing 状态任务重置回 pending这招解决了绝大多数卡死问题。4. 常见问题与排查技巧实录这个架构跑了几个月确实积累了不少真实场景下的故障。我挑几个最常见的分享同时把排查思路和解决步骤一起列出来方便你遇到类似问题时少走弯路。4.1 任务一直 pending没有 Agent 消费第一次遇到这问题我以为代码写错了其实大多情况下是任务类型与 Agent 的 task_type 对不上。比如调度器插入了typeanalyse拼写错误而分析官只监听analyze自然没人消费。排查技巧是登进数据库执行一条查询看看都有哪些 type 值以及各自的状态SELECT type, status, count(*) FROM tasks GROUP BY type, status ORDER BY type;如果发现某 task_type 完全没配置对应 Agent那问题就清楚了。还有一个可能Agent 进程崩了并且没有被 supervisor 自动拉起。检查容器状态时用docker compose ps看 worker 是否是 Exited 状态如果是就看日志里退出原因。我在 Compose 文件里给 worker 加了restart: always并且加了 healthcheck这样进程异常退出时能自动重启。4.2 并发导致同一个任务被处理两次这个问题起初很隐蔽。表现为偶尔出现重复发布的简报。排查发现原因是两个 Agent 实例同时查到同一个 pending 任务。我的数据库查询语句开头如果没写FOR UPDATE SKIP LOCKED两个连接都会读到同一行处理完毕后都更新为 done但其中一个任务实际上被处理了两次。解决方法是给查询语句加FOR UPDATE SKIP LOCKED这属于 PostgreSQL 的语法在 MySQL 里你需要换一套方案推荐用 Redis 分布式锁。另外我还在消费端做了一个幂等保护任务结果写入下游任务之前先检查该 batch_id 和 task_type 是否已经存在 done 结果如果存在就直接跳过。双保险下来重复任务基本绝迹。4.3 输出内容不稳定偶尔出现空结果或脏数据写手 Agent 偶尔会返回空字符串或者输出一大段“抱歉我无法完成这个任务”。虽然 Prompt 已经规定了输出格式但大模型本质上还是概率输出。我采用了两层兜底方案。第一层是解析前校验如果响应为空或长度少于 50 个字符直接抛出异常触发重试。第二层是后处理清洗把模型输出里多余的可疑语气词如“好的这是您需要的稿子”等前缀去掉。这里我写了一个函数clean_model_output()它支持配置化替换列表例如默认替换掉“好的”“以下是”“根据您的要求”等常见前缀。def clean_model_output(text): for prefix in [好的, 以下是, 根据您的要求, 当然可以]: if text.startswith(prefix): text text[len(prefix):] return text.strip()这个方法简单但有效。真正难处理的是审核官“过于苛刻”的问题——它经常把写手的稿子退回导致重写循环。后来我调整了审核官的 Prompt让它只检查硬错误事实日期、来源缺失、格式错误对于风格和语气只给建议不给 REVISE任务流转立刻顺畅了。这也提示我们Agent 之间的主观判断必须降到最低能用规则判定的事情绝对不要让模型自由发挥。4.4 数据链路长查问题太慢怎么办五个 Agent 串行完成任务链路最长的时候可能涉及十几个步骤。如果整批数据质量有问题可以参考我维护的 Batch 追踪方法。每一条任务里都包含了batch_id所以先用一句话查询把整个 batch 的日志拉下来SELECT id, type, status, created_at, updated_at, last_error FROM tasks WHERE batch_id20250321_0900 ORDER BY created_at;这个结果会展示整个流水线的时间轴。你会发现哪一步耗时最长或者哪一步的 last_error 有记录。我之前排查过一次“简报内容过时”的问题就是看到资料官任务完成时间比发布时间晚了三小时最后发现问题出在资料官抓取某个网站时被反爬拦截改用模拟浏览器并设置随机间隔后解决。4.5 成本控制5 个 Agent 会不会烧钱会但可以算清楚。我按 token 来预估成本资料官和分析官每天总共消耗约 30 万 token写手消耗 10 万审核官消耗 2 万发布官消耗 1 万。按一个中位数价格算每日成本大概几块钱人民币一个月下来远低于请人写稿的费用。但要注意的是即使只有 5 个 Agent如果某个步骤出现重试 3 次token 消耗可能翻倍。所以我给每个 Agent 设置了每日 token 上限达到上限就暂停消费新任务并推送告警。这个上限值写在配置里比较容易调整。模型路由我做了简单的负载策略简单的机械任务用价格最低的模型需要语言创作的任务用性能好的模型而审核任务用价格适中但输出格式非常稳定的模型。这套路由不是死的你可以按实际效果随时切换。4.6 个人经验先跑通最小闭环再铺终端场景最后分享一个心得体会也是我建议你参考的路径不要一开始就上 5 个 Agent。最初我只用两个 Agent资料官 写手先跑通“检索→生成”这个最小闭环。确认稳定后再加入分析官把两者串联成“检索→分析→生成”。每加入一个角色都要单独测试它的输入输出兼容性并观察一段时间。这样做的原因是多 Agent 系统的问题往往不在单个 Agent 质量而在于任务交接的格式和语义是否匹配。你一次引入太多角色出了问题根本不知道是谁的锅。另一个经验是Agent 数量不是越多越好。我见过有人把一个大任务拆成 15 个小 Agent结果大部分时间都花在任务切换和数据传递上效果反而变差。正确的做法是“按需拆分”如果一个 Agent 能稳定完成某个环节且不需要其他能力就不要再拆了。5 个 Agent 对于内容生产类型的工作流来说是一个效率与复杂度平衡得比较好的规模。如果你正准备入坑多 Agent 协作我建议你把重点放在任务状态设计和消息契约设计上这两个做好了后面就是堆业务逻辑的体力活。不要先纠结模型多聪明先保证流程不断、数据不乱AI 员工才能真正为你所用。