企业官网建设流程全解析

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 处理）、status（pending、processing、done、failed）、payload（JSON 内容）、result（JSON 内容）、retry_count（重试次数）、created_at（创建时间）和 updated_at（更新时间）。每个 Agent 是一个常驻进程，每隔几秒就查一次“有没有我这个类型的 pending 任务”，取出来处理，写回结果，再更新状态。

举个例子，资料官跑完一批抓取任务后，把几十条结构化记录合并成一个大 JSON，插入到 tasks 表，type 填analyzer，status 填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_pass@db: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 status='processing', updated_at=now() 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 status='done', result=:result, updated_at=now() 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(summary=summary, style=style) response = call_llm(prompt, temperature=0.8, max_tokens=3000) 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.py：

from datetime import datetime from sqlalchemy import create_engine, text import json TOPICS = ["新能源电池", "大模型推理优化", "Web3钱包安全", "生物医药AI", "自动驾驶传感器"] def create_daily_seed(): engine = create_engine("postgresql://agent:agent_pass@localhost: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 status='pending', retry_count=retry_count+1, last_error=:err, updated_at=now() WHERE id=:id """), {"id": task_id, "err": error_msg}) conn.commit() else: conn.execute(text(""" UPDATE tasks SET status='failed', last_error=:err, updated_at=now() 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 对不上。比如调度器插入了type='analyse'（拼写错误），而分析官只监听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_id='20250321_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 员工才能真正为你所用。