企业官网建设流程全解析

1. 项目背景与需求拆解

1.1 为什么需要 24 小时不停机的多 Agent 集群

先说结论：单 Agent 在工作流自动化这件事上，瓶颈非常明显。比如让一个 Agent 同时处理数据采集、内容生成、定时发送三个任务，你很快会遇到上下文窗口被占满、任务互相阻塞、失败重试互相干扰这些问题。这不是模型能力不够，而是任务调度和资源隔离的架构问题。

我最初的需求其实很简单：需要一套系统，能够持续从多个数据源抓取信息，经过筛选和处理后生成结构化摘要，再推送到内部接口。由于数据源是全天候产生数据的，采集和生成任务也必须全天候运行。试过单 Agent 定时触发的方式，每次都要手动清上下文，任务一多就出现上下文污染，生成质量明显下降。后来干脆自己搭了一套多 Agent 集群，专治这类持续型任务。

这个方案的核心思路是：把不同职责拆成独立 Agent，每个 Agent 只做一件事，通过消息队列解耦，由调度中心统一管理生命周期。这样单个 Agent 挂掉不影响全局，任务积压时可以横向扩容，上下文互不干扰，模型调用成本也能按 Agent 维度精确统计。

这篇文章主要面向两类读者：一是正在用 Agent 做自动化任务，但被单点故障和并发问题卡住的人；二是想了解多 Agent 编排架构的人。我会把架构设计、部署细节、实际运行数据、踩坑记录都写出来，你可以直接拿来参考。

1.2 核心需求与功能指标

动手之前，我先列了一份需求清单，这个环节非常重要，决定了后续所有技术选型。不要脑子一热就开始写代码，先想清楚你要解决的到底是什么问题。

在这份需求里，还有一个隐性问题：任务质量如何保证。Agent 处理结果的准确率不可能 100%，特别是大模型生成的摘要，需要建立抽查和人工复核机制。我把这个也写进了需求里，放在数据质量模块中，后面会详细说明。

2. 整体架构设计思路

2.1 多 Agent 集群的拓扑结构

整个系统采用经典的协调者-工作节点模式。调度中心负责接收任务、拆分任务、分发任务，各工作节点（Agent）只负责执行。消息层用 Redis Stream 作为任务队列，状态层用 MySQL 记录任务元数据，执行层用 Docker 容器封装每个 Agent 的独立运行环境。

具体来说，有三种角色：

• 调度器（Scheduler）：定时扫描任务表，生成任务实例，投递到 Redis Stream。
• 执行器（Executor）：从 Redis Stream 中拉取任务，实例化对应 Agent 处理，返回结果。
• Agent 工作节点：每个节点是一个独立的 Python 进程，封装特定的处理逻辑，比如数据抓取、文本摘要、格式转换。

执行器本身是无状态的，挂掉了可以从任务表恢复未完成任务。Agent 工作节点也是容器的、可横向扩展的。调度器和执行器是一对多的关系，调度器负责分配任务 ID 和记录状态，具体执行完全交给执行器。

我自己最满意的设计决策是：Agent 之间不允许直接通信，所有交互都通过调度器或消息队列中转。这保证了系统不会因为某个 Agent 异常调用链而变成网状耦合，整个系统保持星型拓扑，线上排查问题非常方便。

2.2 任务编排与状态机设计

任务状态机是整个集群的核心，我把任务状态设计成七个：PENDING、DISPATCHED、RUNNING、SUCCESS、FAILED、RETRYING、TIMEOUT。状态流转逻辑如下：

1. 调度器生成任务，状态置为PENDING；
2. 投递到 Redis Stream，消费者拉取后更新为DISPATCHED；
3. Agent 开始执行更新为RUNNING；
4. 正常完成更新为SUCCESS，异常更新为FAILED；
5. 失败且重试次数未达上限，重新投递到延迟队列，状态为RETRYING；
6. 执行超过预设时间阈值，更新为TIMEOUT，由调度器重新投递或人工介入。

这个状态机的好处是：任何时候，你只要查任务表，就能知道系统里每个任务处于哪个环节，不需要翻日志。线上定位问题时，直接按status字段筛选，效率极高。

还有一个细节：任务重试不能硬重试，必须带退避策略。我用的指数退避，第一次重试延迟 1 秒，第二次 4 秒，第三次 16 秒，最多重试五次后进入FAILED。为什么要这样？因为你调用第三方 API 报错时，大概率是对方限流，立即重试只会加剧限流，延时重试反而成功率更高。

# 状态机核心实现片段 class TaskStateMachine: def __init__(self): self.valid_transitions = { "PENDING": ["DISPATCHED", "TIMEOUT"], "DISPATCHED": ["RUNNING", "TIMEOUT", "FAILED"], "RUNNING": ["SUCCESS", "FAILED", "TIMEOUT", "RETRYING"], "RETRYING": ["DISPATCHED", "FAILED"], "TIMEOUT": ["RETRYING", "FAILED"], "FAILED": ["PENDING"], # 手动重推 } def transit(self, current, next_state): if next_state not in self.valid_transitions.get(current, []): raise ValueError(f"非法状态流转: {current} -> {next_state}") return next_state

2.3 为什么选择 Redis Stream 而非 RabbitMQ 或 Kafka

任务队列选型的时候，我在 RabbitMQ、Kafka、Redis Stream 之间纠结了很久。最终选了 Redis Stream，原因有三：

第一，Redis 本来就在基础设施里，不需要额外引入中间件。当时这套集群跑在 4 台 Linux 服务器上，已有 Redis 在做缓存，Stream 是 Redis 5.0 自带功能，零成本接入。

第二，Redis Stream 的消费者组模型非常适合任务分发场景。多个执行器可以组成消费者组，同一个任务只会被一个执行器消费一次，这正好满足任务不重复执行的要求。

第三，调试非常方便。Redis Stream 本质上还是 Redis 数据结构，XRANGE、XLEN、XINFO命令可以直接查看队列长度和消费进度，不需要额外的管理界面和调试工具。

Kafka 不是不行，但它的运维成本对一个 5 个 Agent 的集群来说太重了。RabbitMQ 功能全面但还没用到它那么多路由特性。等到将来任务规模确实超过 Redis Stream 的能力范围，再迁移到 Kafka 也不迟，业务逻辑在状态机层面已经解耦，队列只是传输层。

3. 集群部署与核心实现

3.1 运行环境配置与容器化隔离

我用了四台 Linux 服务器，配置不算高：16 核 CPU、64GB 内存，系统盘加数据盘。部署方式全部走 Docker Compose，每个服务一个容器，包括调度器、执行器、Redis、MySQL、监控组件。

容器化隔离这个决定，原因是后期 Agent 版本升级太频繁。我最早直接在宿主机上跑 Python 进程，每次升级依赖都担心影响其他 Agent，比如 Agent A 要 pydantic 2.x，Agent B 还在用 1.x，pip 安装完 A 的依赖，B 就崩了。容器化之后，每个 Agent 独立镜像，依赖互不干扰，升级只需要重新构建镜像再滚动重启容器。

数据盘单独挂载，因为任务日志和模型输出都比较大，MB 级别的文本文件一天能积累几个 GB 下来。系统盘要是被日志撑满了，整个集群都会响应变慢，这是血泪教训。

容器资源配置我做了硬限制：每个 Agent 容器最多 2 核 + 4GB 内存，调度器容器 2 核 + 2GB，Redis 和 MySQL 各自 4 核 + 8GB。限制资源的目的是防止某个 Agent 出现内存泄漏时拖垮整台服务器。Docker 的--memory限制配合--memory-swap可以让容器内存超限直接被 OOM Kill，而不是把宿主机的内存耗尽。

四个服务器之间走了轻量级内网通信，任务数据不落地磁盘，全部通过内部端口传输。这里要特别强调的是，多 Agent 集群对网络稳定性要求很高，如果你的网络设备不稳定，建议加备用线路，这个后面实操小节会展开说。

3.2 调度器与执行器的代码实现

调度器的职责非常纯粹：扫描任务表、生成任务实例、投递队列。我把调度逻辑抽成了一个独立服务，没有混在执行器里，这样调度策略调整（比如改成 cron 触发、改成基于 crontab 的规则触发）不需要重启执行器。

调度器核心逻辑用 Python 的apscheduler库实现，支持 cron 表达式配置任务时间。这里有个坑：apscheduler的 cron 表达式默认基于 Asia/Shanghai 时区，但如果服务器时区不是中国标准时间，会出现定时偏差，所以容器环境变量里我特意设了TZ=Asia/Shanghai，并且每个定时任务配置文件里显式声明时区。

# 调度器定时投递核心逻辑 from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def dispatch_task(): # 从 DB 获取所有已启用的任务定义 tasks = get_enabled_tasks() for task in tasks: # 生成任务实例并写入 Redis Stream task_id = generate_task_id() task_meta = { "task_id": task_id, "task_type": task.type, "payload": task.payload, "priority": task.priority, } redis_client.xadd("task_queue", task_meta, maxlen=10000) db.update_task_status(task_id, "PENDING") print(f"[{datetime.now()}] 已投递 {len(tasks)} 个任务") scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job(dispatch_task, CronTrigger(minute="*/5")) scheduler.start()

执行器的实现逻辑是：循环从 Redis Stream 读取任务，用XREADGROUP消费，处理完之后XACK确认。如果处理过程中抛异常，把任务标记为FAILED并触发重试机制。如果执行器进程崩溃，Redis Stream 里的消息没有被XACK，重启后会被重新投递，这就保证了任务不丢失。

# 执行器消费逻辑 while True: # 阻塞读取任务，最多等待 5 秒 raw = redis.xreadgroup( group_name="executors", consumer_name=consumer_id, streams={"task_queue": ">"}, count=1, block=5000, ) if not raw: continue stream_name, messages = raw[0] msg_id, payload = messages[0] try: # 标记 RUNNING 状态 update_status(payload["task_id"], "RUNNING") # 根据任务类型选择对应 Agent 处理 result = route_to_agent(payload) store_result(payload["task_id"], result) update_status(payload["task_id"], "SUCCESS") except Exception as e: update_status(payload["task_id"], "FAILED", error=str(e)) retry_task(payload["task_id"]) finally: # 无论成功失败，都已处理完毕，确认消息 redis.xack("task_queue", "executors", msg_id)

Agent 工作节点自身的实现，其实就是带独立提示词模板和工具函数的 Python 类。每个 Agent 定义run(task)方法，内部调用大模型接口、读取外部数据、加工，最终返回结构化字典。不同 Agent 之间不共享任何对象，连模型 API Key 都是独立的子账户，方便做成本统计。

3.3 Redis Stream 消费者组的初始化细节

很多第一次用 Redis Stream 的人容易漏掉一个步骤：在开始消费之前，必须创建消费者组，否则XREADGROUP会直接报错。我刚开始也踩了这个坑，老是报NOGROUP错误，后来在启动脚本里加了一句初始化命令。

redis-cli XGROUP CREATE task_queue executors 0-0 MKSTREAM

这个命令的含义是：在task_queue上创建一个名为executors的消费者组，从0-0开始消费（从头开始），如果流不存在则自动创建。如果重复执行会报BUSYGROUP错误，所以我封装成了幂等逻辑：

def ensure_consumer_group(redis_client, stream, group): try: redis_client.xgroup_create(stream, group, id="0-0", mkstream=True) except Exception as e: if "BUSYGROUP" in str(e): pass # 消费者组已存在 else: raise

消费者组的 pending 列表也是一个必须了解的概念。正常情况下，消息被XACK之后会从 pending 列表移除。如果执行器处理消息过程中崩溃，消息会残留在 pending 列表，用XREADGROUP重新上线后不会自动处理这些滞留消息。我加了一个定时任务：每 5 分钟扫描一次 pending 列表，检查是否有矿工消息滞留超过 10 分钟，如果有就重新投递到主队列。

def requeue_stuck_messages(redis_client, stream, group): all_msgs = redis_client.xpending_range(stream, group, min=0, max=100000, count=100) for msg in all_msgs: consumer, attr_dict, elapsed_ms = msg # 简化表示 if elapsed_ms > 600000: # 超过 10 分钟 # 获取原始内容 raw = redis_client.xrange(stream, min=msg_id, max=msg_id) # 重新投递 redis_client.xadd("task_queue", raw[0][1], maxlen=10000) # 确认掉旧消息 redis_client.xack(stream, group, msg_id)

这段逻辑非常关键，它保证了整个系统的“at-least-once”语义。代价是极端情况下任务可能重复执行，但对数据采集和摘要生成这类任务来说，重复执行的副作用可以接受。

4. 稳定性保障与监控体系

4.1 健康检查与自动恢复机制

调度器、执行器、Agent 容器都有可能崩，所以必须做进程级别的健康检查和自动恢复。我用的是 Docker Compose 自带的healthcheck配置，加上外部 supervisor 守护进程双保险。

每个容器里都内置了健康检查接口，返回200 OK就表明进程活着。健康检查不只是检查进程存活，我会在接口里顺带检查数据库连接是否正常、Redis 连接是否正常、最近一次任务执行时间是否超过阈值。这样能及时发现进程活着但业务已经卡死的情况。

# docker-compose.yml 健康检查配置示例 services: scheduler: image: myregistry/scheduler:latest ports: - "8081:8081" healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8081/healthz"] interval: 30s timeout: 5s retries: 3 start_period: 20s restart: always # 进程退出时自动重启

restart: always是 Docker Compose 自带的进程守护，容器退出后会自动重新拉起。配合健康检查接口，如果进程只是处于假死状态（端口还能通，但业务不执行），就需要额外的报警来提醒人工介入。

外部 supervisor 守护是什么？我写了一个简单的轮询脚本，放在宿主机上每 60 秒访问一次每个容器的健康检查接口，连续失败 3 次就执行docker-compose restart <service>。这个脚本虽然简陋，但在遇到 Docker 守护进程卡死或网络抖动时能兜底，比纯靠 Docker 自愈更可靠。

网络层面的保障有必要单独说一句：多 Agent 集群对外部数据源的访问非常频繁，建议提前规划好网络可靠性，数据入口和数据出口走不同线路，这样一条线路抖动时另一条还能承载流量。我这边实测下来，备用线路在高峰期确实救过几次命，尤其是做信息采集任务的时候，数据源响应变慢很常见，网络线路一旦抖动会影响所有 Agent 的任务。

4.2 监控指标与告警规则

只是把集群跑起来不算完，必须能看清楚系统里发生了什么。我搭建了一套简易监控体系，采集三类指标：

• 任务指标：任务总数、成功率、失败率、平均耗时、队列积压数。这些是集群健康度的核心指标。
• 资源指标：CPU 使用率、内存占用、磁盘 IO、网络流量。资源瓶颈往往先于业务问题出现。
• 模型调用指标：API 调用次数、token 消耗量、单次调用延迟、请求失败率。这部分直接关系到成本。

采集方式没有过度工程化，就是每个服务通过 StatsD 协议把指标推送到一个中心指标服务，然后统一写入 MySQL 的监控表，再用一个 Python 脚本定时把最近一小时的指标聚合成图表，推送到 Web 面板。

告警规则我设置了三条核心规则：

告警阈值不是拍脑袋定的，我会先看一周的基线数据，正常情况下任务成功率在 95% 以上，队列积压数很少超过 50，所以阈值设置是高于基线但留有余地的。告警太灵敏会频繁打扰值班人员，告警太迟钝又失去了意义。

这里有个经验：不要只监控技术指标，还要监控“任务结果质量”。大模型生成的文本，偶尔会出现空内容、乱码、内容重复这些问题，从任务状态看是 SUCCESS，但结果实际不可用。我加了一个后置校验模块，对关键任务的结果做正则校验和长度校验，不符合预期的结果自动标记为 LOW_QUALITY 并重新生成。

4.3 数据持久化和备份策略

任务元数据和执行记录存储在 MySQL，每天凌晨两点自动全量备份到备份盘，同时保留最近 7 天的增量备份。Word 文档之类的二进制产品存在对象存储里，本地只保存访问链接。

备份的重要性我深有体会，有段时间因为数据库误删导致整个任务队列恢复不了，前后花了四个多小时才补回来。后来做了两件事：一是备份策略改为每日全量 + 每两小时增量，二是引入了数据库误操作前的闪回机制，从表里直接恢复被删的元组。成本不高，但关键时刻能救命。

Redis Stream 本身不是持久化存储，重启会丢数据。我把 Redis 的持久化策略设置为 RDB 快照 + AOF 每秒刷盘，这样至少能保证秒级数据不丢失。队列消息丢失的问题我用另一个方案兜底：任务状态都记录在 MySQL 里，就算 Redis Stream 全丢了，还可以从 MySQL 中恢复出所有未完成任务，重新投递到新队列。这种“双写状态 + 队列缓存”的模式，任务可靠性才真正有保障。

5. 实际运行中的问题与排查实战

5.1 任务积压导致的内存暴涨

集群稳定运行两周后，某天突然收到内存告警，监控面板显示执行器容器内存使用率从 40% 飙升到 95%。定位过程花了三个小时，但问题原因其实很简单：某个数据采集任务依赖的外部数据源响应变慢，单个任务执行时间从原来的 5 秒拉长到 30 秒。任务总耗时变长，队列积压越来越多，执行器消费速度跟不上生产速度，于是启动了多个并发消费线程，每个线程都会加载一次大模型客户端，内存就这样被吃满了。

解决方案分两步。第一步，临时措施：新增两台服务器，把执行器容器扩容到三份，并行消费队列压力。第二步，根因修复：在调度器侧增加熔断机制，如果检测到某个数据源连续 10 次请求超时，暂停投递该数据源相关任务 5 分钟，避免无效任务占满队列。

从这个案例里提炼出的经验是：任务积压和资源异常往往只是表象，真正要查的是数据源响应和任务耗时这两个上游指标。只看队列积压数来找问题，会走很多弯路。

5.2 上下文长度溢出与输出截断问题

大模型生成摘要时，偶尔会遇到输出 token 数超过模型最大限制，导致返回内容被硬截断，后半段文字缺失。这个问题在任务维度看是 SUCCESS 状态，但实际内容不可用，非常隐蔽。

我在 Agent 层加了输出长度预检查：调用模型前，根据提示词和输入文本长度，估算输出需要多少 token，如果超过模型最大输出限制，先对输入内容做分块处理，再逐块生成摘要，最后合并结果。这个策略有效地解决了长文本摘要任务中输出不完整的问题。

另外还需要处理上下文长度溢出。不要在大模型 API 调用时一次性把超长文本都塞进去，先按字符数或 token 数切片，并设置重叠区。我这里的经验是：切片大小设置为模型上下文窗口的 60%，重叠区设置 200 个 token 即可，既能保证信息连续，又不浪费 token。

5.3 调用第三方 API 的限流问题

多 Agent 集群接入外部服务时，最头疼的就是 API 限流。某个服务的免费额度是每分钟 60 次请求，但集群有五个 Agent 同时运行时，很容易就冲破这个限制，返回 429 错误。

第一版我直接在每个 Agent 里做简单的请求计数，结果多个 Agent 之间不同步，照样限流。后来把限流器抽成了独立服务，用 Redis 的滑动窗口算法做全局计数，所有 Agent 的 API 调用都经过限流器统一调度。实现并不复杂，但效果立竿见影：

# 基于 Redis 的滑动窗口限流 def allow_request(api_key, limit_per_minute): key = f"ratelimit:{api_key}" current_time = int(time.time()) pipeline = redis_client.pipeline() pipeline.zremrangebyscore(key, 0, current_time - 60) # 移除 60 秒前的记录 pipeline.zadd(key, {str(current_time): current_time}) pipeline.zcard(key) # 统计当前窗口内请求总数 _, _, count = pipeline.execute() return count <= limit_per_minute

滑动窗口比固定时间窗更平滑，不会出现“整点重置后瞬间打满”的毛刺问题。限流器返回allow=False时，Agent 会等待重试，不会直接报错。

5.4 网络线路抖动引起的连锁反应

网络问题是最难排查也最容易忽视的。刚开始运行的那一个月，出现过两次莫名的低成功率告警，排查到后期才发现不是程序问题，是网络抖动导致外部数据源连接超时，Agent 本身逻辑没问题，但所有依赖这个数据源的任务都失败了。

网络波动引起的连锁反应是：任务失败触发重试，重试导致请求量翻倍，请求量翻倍又加剧数据源压力，数据源继续超时，形成恶性循环。后来我在重试逻辑上加了一个限制：单任务最大重试次数从 5 降到 3，并且在数据源连续失败时启动全局熔断。数据源恢复后，熔断器自动放行。

这个案例给我的经验是：多 Agent 集群的稳定性不只是写好代码，还要关注上游依赖的健康度。第三方接口不可控，但必须保证自己的容错机制足够健壮，才不会因为上游抖动而雪崩。

5.5 常见问题速查表

5.6 一波三折的故障复盘：从崩溃到优化

有一次比较严重的故障，让我彻底重构了部分容错设计。情况是：执行器所在服务器宕机，所有进程全部退出，MySQL 连接数瞬间打满，导致调度器也连不上数据库，整个集群完全停摆。

复盘发现三个隐患：第一，所有服务共用一台数据库服务器，单点故障面太大；第二，连接池配置了最大连接数 200，但实际场景只需要 50 左右，打满原因是每个 Agent 都创建了大量短连接；第三，调度器自身没有本地缓存，数据库一挂调度器就完全瘫痪。

针对这三点做了改造：第一，数据库做了主从切换，宕机后备用实例自动接管；第二，统一连接池管理，所有服务复用同一个 Redis 连接池和 MySQL 连接池，设置合理上限；第三，调度器加了本地任务缓存，数据库不可用时暂停投递，但不至于进程崩溃。改造完成后，集群的可用性明显提升，这类故障再没出现过。

故障复盘的价值在于，每次事故都能暴露出设计缺陷，而这些缺陷往往在正常运行时根本看不出来。持续修复这些缺陷，系统的健壮性才是真正意义上的提升，而不是“跑起来了就算稳定”。

6. 性能调优与资源成本控制

6.1 并发参数与批量处理优化

多 Agent 集群的吞吐量受限于两个因素：单 Agent 的处理速度和并发数。并发数不是越大越好，因为大模型 API 是共享资源，并发过高会导致响应变慢，反而降低整体吞吐。

我做了几轮压测，最终把执行器并发线程数设置为 4，每个 Agent 的并发请求数限制为 2，整个集群并发模型调用数控制在 20 以内。这个参数在不同模型和不同任务类型下会变化，建议以实际压测数据为准，而不是凭感觉拍脑袋。

批量处理方面，有时间窗口内多个任务请求同一个数据源的情况，我会做一个聚合操作：把请求合并成一个批量请求，减少网络往返和资源消耗。比如数据采集任务每 15 分钟会抓取一批数据，总共 50 个 URL，串行抓取需要 5 分钟，用aiohttp并发抓取只需要 40 秒，效果非常明显。

另外还有一个值得优化的点：模型调用的输入压缩。在处理长文本时，摘要任务只需要关键信息，可以先用简单规则抽取核心段落，再做模型生成，减少 token 消耗。我实测下来，压缩后的 token 消耗能减少 30% 到 50%，成本节约非常可观。

6.2 模型选型与分级调度策略

不是所有任务都需要最强模型。我的集群里设计了三个模型档位：基础档用于文本分类、关键词提取这些简单任务；标准档用于摘要生成、内容改写；增强档用于复杂推理、长文本深度分析。调度器在生成任务实例时，会根据任务类型和输入文本复杂度自动分配模型档位。

这个策略反映出来的是“成本意识”，基础档和增强档的 API 价格差了十几倍，如果全部都用增强档，每个月成本直接翻几倍。分级调度后，整体成本下降了约 40%，而任务质量没有明显下降。

具体的分配逻辑是：任务类型携带model_tier字段，Agent 根据字段选择对应模型。如果是长文本摘要任务，输入长度超过 8000 字符就打标为tier2，否则tier1。最终模型选择不是写死在代码里的，而是通过配置中心动态调整，方便以后替换模型。

6.3 数据去重与任务幂等处理

多 Agent 集群处理数据采集任务时，最容易出现的问题是：同一个 URL 被多个 Agent 采集到，产生重复数据。尤其是采集任务按关键词匹配数据源时，不同 Agent 的关键词集合可能有重叠，同一篇内容会被不同 Agent 抓到。

我的解决办法是：在存储层增加唯一的 URL 指纹字段（对 URL 做 MD5 哈希），入库前查询是否已存在，存在则跳过。数据库层面用唯一索引做拦截，再配合 Redis 的 SETNX 做并发场景下的互斥，双保险确保不重复入库。

任务幂等性方面，我设计了task_instance_id + action作为唯一业务键，同一份数据重复执行任务时，会直接返回已有的结果缓存，而不是重新生成。这个逻辑在处理“队列消息重复消费”的场景时很关键。

缓存的有效期设置为 24 小时，超过缓存时间的任务需要重新执行。对实时性要求很高的任务不设置缓存，对历史数据补采任务设置缓存，因为同一时间窗口内重复调度概率很低，但缓存可以避免人工误操作造成重复触发。

6.4 夜间削峰与任务优先级队列

任务负载是有明显波峰波谷的。9:00 到 18:00 是任务高峰期，外部数据源活跃，用户触发的任务也多；凌晨 2:00 到 6:00 任务量骤降。如果按照峰值负载来配置资源，大部分时间资源都是浪费的。

我做了削峰策略：把不紧急的批量任务（比如历史数据归档、大文本的离线摘要）调度到凌晨低峰执行，紧急任务（用户实时请求、监控告警）优先抢占资源。实现方式是在任务表加priority字段和earliest_run_time字段，调度器生成任务时，如果任务是低优先级且允许延迟，就把其执行窗口挪到凌晨。

这种时间维度上的调度策略特别适合成本敏感型项目，不需要为了一天只有三小时的高峰期去额外买服务器。集群的资源利用率从原来的不够用，变成现在的稳定在 50% 到 70% 之间，算是一个比较健康的水平。

7. 多 Agent 集群的扩展方向与长期运营思考

7.1 从“能跑”到“好用”的能力演进

基础集群稳定运行之后，我开始考虑一个问题：单纯把任务跑通只解决了“有没有”的问题，要解决“好不好用”的问题，还需要哪些能力？

第一个方向是知识库的积累。Agent 每次执行任务都会产生处理结果、中间数据和错误信息，这些数据散落在各处。我建了一个统一的知识库，把成功任务的经验总结（比如数据源响应规律、文本特征、模型参数）沉淀下来，新任务执行时可以参考旧结果，减少试错成本。

第二个方向是 Agent 技能的模块化。目前每个 Agent 是独立进程，但很多底层能力是重叠的（比如数据清洗、格式校验、请求重试），这些完全可以抽成共享库。我目前正在做的是把公共能力做成插件包，每个 Agent 按需加载，减少冗余代码，也让新 Agent 的开发成本降低。

第三个方向是反馈闭环。Agent 生成的内容再怎么自动化，最终还是要服务于人。我加入了一个轻量级反馈界面，业务人员可以对 Agent 的输出打标、评分，这些反馈数据定期回流，作为任务参数和提示词优化的依据。

长期运营这套系统的体会是：不要把多 Agent 集群想象成一个“永久稳定”的成品，它更像是一个需要持续养护的系统。外部数据源会变、模型接口会变、业务需求也会变，运维的重心不是“搭建一次就完事”，而是建立一整套持续观测、反馈、优化的循环。

7.2 多 Agent 协作的边界与拆解原则

很多人一听多 Agent 集群，就以为是把所有任务交给一组 Agent 自由协作。但实际上，Agent 之间协作越自由，系统越难驾驭。我在实战中总结了几条拆解原则：

第一，单 Agent 职责必须单一。一个 Agent 只做一个类型的任务，不要试图让一个 Agent 既做数据采集又做内容生成又做格式转换。职责越单一，调试越简单，出问题时定位越快。

第二，Agent 之间的交接越简单越好。如果 Agent A 的输出要作为 Agent B 的输入，中间必须有一个结构化的数据格式约定，不建议直接传自然语言文本。我在中间层设计了统一的 JSON Schema，所有 Agent 输出都校验后再流转。

第三，不要引入复杂的对话协作机制。多 Agent 互相讨论、互相评审听起来很美好，但实际运行中会陷入无限循环、token 消耗失控、结果无法收敛这些问题。目前真正稳定可控的模式仍然是“流水线式”协作：任务一个大步骤拆成若干小步骤，每个小步骤对应一个 Agent，步骤之间顺序执行或者有条件的并行执行。

7.3 成本模型与容量规划

多 Agent 集群不是一次性投入，后续每个月都有模型调用和数据存储成本。我把成本拆成三块来管理：

• 模型调用成本：这是大头，通过分级模型、token 压缩、缓存命中来降低，上面已经提到过；
• 服务器成本：通过容器化弹性伸缩来控制，高峰期临时加容器，低峰期回收多余容器；
• 存储成本：定期清理过期中间数据、压缩日志，避免磁盘不断膨胀。

容量规划方面，我用的是“按任务类型预估 + 监控回测”的方法。先根据每个任务类型的历史日均执行量、平均单个任务耗时，估算出需要的执行器数量，再通过监控数据回测验证估算是否准确。规划结果不用太精确，但要有依据，这样才能避免临时加服务器的被动情况。

我踩过的一个坑是：服务器数量按任务量最大峰值来规划，结果平时 70% 资源闲置，成本居高不下。后来改成按基准负载 + 弹性扩容的方式，节省了大概三成开支。

8. 一些踩坑心得与改进建议

8.1 一定要记录任务级的“上下文”

前面提到过 Agent 之间共享上下文会导致污染，但这不是说完全不要上下文。我做的改进是：每个任务实例维护一份独立的上下文记录（包括输入、输出、中间步骤、关键参数），存在 MySQL 的一个 JSON 字段中。这样一来，任务可完整回放，出问题时可以精确定位是哪个环节出了错。

任务级上下文还有一个好处：它支持“人工接管”。当自动处理结果不符合预期时，运维人员可以直接从上下文里看到原始输入和 Agent 的推理过程，手动修正甚至重新提交，不用从头重新跑整个流程。这个设计让系统从“黑盒”变成了“灰盒”，可维护性大幅提高。

8.2 失败任务的人工复核流程

就算集群再稳定，也会有模型输出的“软错误”（格式正确但语义不对）。我建立了一个失败任务人工复核队列：自动处理失败标记的任务，按优先级排队展示给复核人员，复核通过后重新进入处理流程，复核不通过则进入人工处理通道。

这个流程的价值在于：它保证了任务处理结果的质量下限。自动化任务追求的是效率，但质量底线必须由人来守住。尤其是对外发布内容或数据质量要求很高的场景，人工抽检不能省。

我建议比较合理的比例是：高风险任务 100% 复核，中风险任务 30% 抽查，低风险任务 5% 抽检。高风险管理起来虽然麻烦，但长期来看是值得的。

8.3 集群扩容的正确姿势

扩容不是把容器数量调大就完事的。我刚开始扩容器时，直接升级了执行器容器副本数，结果多个执行器同时消费同一个 Redis Stream，任务冲突率高了不少。

Redis Stream 消费者组其实已经处理了“多消费者并行消费”的互斥问题，但前提是组内消费者数量不超过队列分区数。XREADGROUP的机制是：一个消费者组内的不同消费者会收到不同的消息，但同一个消费者组内所有消费者共享同一个 pending 列表。扩容前要检查队列是否足够多，否则新增的消费者收不到消息，白占资源。

正确扩容流程是：首先确认 Redis Stream 有足够的待消费消息，再增加执行器容器副本数，最后监控任务处理速度和任务耗时，验证扩容是否达到预期效果，而不是一上来就复制容器。

8.4 日志管理的几个建议

多 Agent 集群的日志量非常大，不管理好日志，排查问题时会陷入日志海洋。我的日志策略是：

第一，结构化日志。所有日志输出统一采用 JSON 格式，包含时间戳、任务 ID、Agent 名称、日志级别、消息体。这样日志可以方便地聚合检索，按任务 ID 串联所有相关日志，而不是靠 grep 猜关键字。

第二，分级存储。热日志（7 天内）存放在本地磁盘或热存储，访问速度快；冷日志（7 天以上）压缩后转存到低成本存储，定期清理超过 90 天的日志。

第三，日志采样。对于执行频率极高、日志量很大的任务，只记录错误日志和采样日志（比如每 100 次记录一次完整日志），避免单任务日志量太大影响磁盘空间。

经验是：日志宁可多写也不要少写，但关键是写入时做好分类，检索时才能高效。

8.5 一点运营层面的建议

最后聊点偏运营层面的体会。多 Agent 集群搭建起来之后，不要天天只盯技术指标，更要关注“任务结果是不是真的被使用”。我见过不少项目，集群任务完成率很高，但业务方反馈根本没用到结果，原因是任务目标本身定义得不对。

建议是：每两周和业务使用方回顾一次任务清单，确认每个任务是否仍然有存在价值，输出格式是否仍然符合预期。无效任务及时下线，不养无用负载。这些看似和技术无关的工作，其实才是保证系统长期有价值的关键。

跑通一套多 Agent 集群并没有想象中那么难，真正难的是让它稳定地、持续地、有价值地运行超过半年。如果你正在规划类似系统，记住我前面说的这些设计原则和踩坑记录，应该能帮你少走不少弯路。