企业官网建设流程全解析

这篇我聊长任务和多 Agent。做 Agent 工程到现在，我自己感受最深的一个转折点：demo 里单 Agent 什么都能干，一旦塞进真实业务流程——几十个步骤、几十分钟执行、还要跟别的 Agent 协作——原先所有"看起来不错"的设计全部失效。第四篇就围绕两个核心问题写：一是任务一长，状态、上下文、恢复怎么做；二是 Agent 一多，编排、通信、容错怎么做。适合已经跑通单 Agent Demo、正在往生产形态推进的工程同学参考。

1. 为什么"长"和"多"是同一个问题

1.1 单 Agent 的"短命才完美"

很多人会把 Agent 想象成一个越聊越聪明、什么都能接的助手，实际跑过一次长任务就明白：单 Agent 天生是"短命"的。

原因很朴素。模型每次调用只能基于上下文窗口里的内容做决策，而我们的任务状态、工具返回、中间结论全都在往这个窗口里塞。一个需要调用 40 次工具的任务，跑到 20 次的时候，前面 15 次工具结果已经被截断了。你会看到模型开始"失忆"：它忘了最初的任务目标、忘了已经确认过的结论，甚至开始重复调用同一个工具。

我管这个叫"三分钟热度天花板"。单 Agent 的架构里，上下文既是工作台又是记事本又是仓库，三者的容量共享同一个窗口。任务一拉长，最先爆掉的就是这个共享空间。再加上长时间运行一定会遇到进程重启、超时、限流这类环境问题，没有外部状态支撑的 Agent 一崩就从头再来，成本直接翻倍。

1.2 什么样的任务才算"长任务"

并不是执行时间超过十分钟就算长任务，我习惯从三个维度判断：

三个维度只要沾上两个，基本就得放弃"一个 Agent 从头干到尾"的朴素方案。举一个身边最常见的例子：做一个"多源资料整理报告"的任务，光是原始材料的字符量就可能超过模型窗口，这还没算分析结论和草稿。此时讨论模型能力强不强已经没有意义，问题出在架构边界上。

1.3 多 Agent 真正解决的问题是"可控性"

先说一句可能让很多人意外的话：多 Agent 不是为了变强，是为了可控。

单 Agent 像一个全能选手，能力强但所有状态都在一个脑子里，出了问题你只能看到它"做错了"，说不清是哪一步、哪个输入、哪个决策导致的。拆成多个 Agent 之后，每个 Agent 的上下文变小、职责变清晰、输入输出变成结构化数据，你就可以在 Agent 之间放隔离层、检查点和人工审核点。

这也是为什么我强烈建议从"长任务"入手理解多 Agent：长任务迫使你把状态显式化，而多 Agent 正是承接显式状态的一种自然组织方式。两者不是两个独立话题，而是同一个架构问题的一体两面。

2. 长任务的正确打开方式

2.1 规划先行：把"一次性表演"改成"流水线"

单 Agent 处理长任务的第二个坑是计划丢失。模型会在第 5 步还记得"我们要分四阶段做"，到第 15 步就开始即兴发挥。我的解决办法是把计划当成一等公民，落成数据，而不是模型记忆。

具体做法很简单：任务开始前，先让规划 Agent 基于目标产出结构化步骤清单，每步包含目标、产出物、验收条件。执行 Agent 每完成一步，就去更新这个清单的状态。计划在外部存储里，任何一次模型调用都不是从零开始，而是"读取计划 + 看当前进度 → 决定下一步"。这样即使用户中途追加需求，也能通过修改步骤列表来让整个流水线自适应。

我用的是 plan-then-execute 模式。第一步只产计划不干活，第二步严格按计划执行；执行中如果发现某一步需要偏离计划，必须先修改计划再执行，不允许 Agent 自己"顺手"改方向。这个约束能显著减少长任务后期的失控概率。

2.2 状态落盘与断点续跑

长任务必须接受一个现实：进程一定会被打断。可能是模型服务超时，可能是宿主环境重启，也可能只是你手动停掉一个跑飞了的任务。没有断点续跑，前面所有成本白花。

一个可落地的状态设计长这样：

from dataclasses import dataclass, asdict, field from typing import Any @dataclass class TaskState: task_id: str plan: list[str] # 步骤清单 done: set[str] = field(default_factory=set) # 已完成步骤 outputs: dict[str, Any] = field(default_factory=dict) # 步骤产物 attempts: dict[str, int] = field(default_factory=dict) # 各步骤重试次数 def checkpoint(self, store) -> None: payload = json.dumps(asdict(self), ensure_ascii=False) store.put(f"tasks/{self.task_id}.json", payload) @classmethod def resume(cls, task_id: str, store) -> "TaskState | None": raw = store.get(f"tasks/{task_id}.json") return cls(**json.loads(raw)) if raw else None

每一步执行完之后立刻调用checkpoint，执行前先resume。恢复的逻辑是：检查 done 集合，跳过已经完成的步骤，只重跑没做完的。这里有一个容易被忽略的细节：attempts字段一定要单独记录，否则某一步反复失败时，重试次数会跟着状态丢光。

存储后端不用太复杂，对象存储、数据库表、甚至本地 JSON 文件都行。关键不是用什么存储，而是"每一步的输入输出都可追溯"。我习惯把每一步的产物单独存一份，目录结构是tasks/{task_id}/{step_id}.json，状态文件只保留引用关系，这样单步产物再大也不会拖垮整个状态文件。

2.3 上下文瘦身四板斧

状态问题解决了，上下文怎么控制？我常用的手段按成本从低到高排列：

• 滚动摘要：每完成一个里程碑，让一个轻量 Agent 把已完成部分浓缩成 500 字以内的摘要，后续步骤只加载摘要，不加载原始对话。
• 上下文压缩：把早期已经被消费过的大段工具返回折叠成结构化结论，只在需要细节时才重新加载原文。
• 外部记忆：把资料切片后放进检索工具，Agent 需要哪段就调工具取哪段，而不是开始就把全部资料塞进窗口。
• 按需加载：超过阈值的数据全部走文件系统或数据库，Agent 永远只面对"当前这一步需要的最小数据"。

预算分配上，我的习惯是：假设模型窗口 200K，系统提示加工具描述不超过 30K，单次调用的输入上下文控制在 120K 以内，留出生成空间。超过这个阈值就主动压缩，而不是赌模型能扛住 200K 塞满的情况。

提示：摘要不是每一步都做，而是里程碑式做。每一步都摘要既费 token 又会丢失细节，我在实操中最优频率是每 5~10 步或每次切换阶段时做一次。

3. 多 Agent 的架构选型与分工

3.1 三种主流编排方式

多 Agent 最容易踩的坑，是一上来就设计一个"Agent 联邦"。我的建议是先从最保守的拓扑开始，按任务的流动方式选型：

从我自己的项目经验看，90% 的业务场景用 Supervisor 就够了。流水线适合纯数据加工流程，对等协作除非你有非常明确的终止条件（比如最大轮数、达成共识的判据），否则别碰。

3.2 职责边界怎么切

多 Agent 分工的原则可以用一句话概括：每个 Agent 只握一份领域上下文、一套工具子集、一个明确产出物。

很多失败案例都是因为把全部工具都给了每个 Agent。表面看是"每个 Agent 都很全能"，实际是模型不知道该用哪个工具，两个 Agent 还会为了同一个参数的取值互相推诿。这就像一家公司里所有人都是"总监"，反而没人干活。

我切分职责的做法是看工具而非看角色。先把工具按领域分组，比如"资料检索组""数据分析组""文件输出组"，然后让每个 Agent 只对应其中一组。执行 Agent 的 system prompt 里明确写清三件事：我是谁、我负责什么、我交什么格式的产物。产出物一定要有 schema，哪怕是几行 JSON，也远比一句"整理好报告发回来"可靠。

3.3 Agent 之间的"普通话"：消息协议

多 Agent 之间的通信不是聊天，是协议。没有协议约束的 Agent 对话会变成一场双方都在自说自话的灾难。

我建议所有 Agent 间交互都走结构化信封：

{ "message_id": "msg_20240101_001", "task_id": "task_20240101_001", "sender": "researcher", "receiver": "supervisor", "thread_id": "topic_017", "payload_type": "topic_note", "payload": { "topic": "agent memory", "conclusion": "...", "sources": ["..."] }, "created_at": "2025-01-01T10:00:00Z" }

payload_type决定了消息体的结构；thread_id用于多轮协商时保持线索完整；状态更新永远放在 payload 里，而不是靠 Agent 用自然语言描述"我完成了"。

注意：不要让两个 Agent 直接用自然语言无限对聊。无论用 Supervisor 还是流水线，我都要求每个 Agent 一次调用必须产出可校验的结构化结果，直接写入状态。让模型"描述结果"和"输出结构化结果"的成本差不了多少，但后者让验收、恢复、审计都变得可行。

4. 实操：拆一个"多源资料整理报告"任务

4.1 原始单 Agent 的问题

为了把上面的方法串起来，我用一个真实场景演示：收集 20 篇不同来源的资料，产出一份带章节结构的专题综述。

最初我用单 Agent 跑，问题很典型：资料检索阶段返回的内容全部堆在上下文里，到写综述时模型已经开始"选择性记忆"，引用的数据张冠李戴；最难受的是每次检索失败重跑，前面检索的成果全部清零。

4.2 改造为 Supervisor + 3 个 Worker

我把任务拆成四个 Agent：规划 Agent 负责产出章节计划；检索 Agent 按主题逐个检索，一次只处理一个主题，输出结构化笔记；写作 Agent 基于全部笔记生成综述；审校 Agent 检查引用完整性与章节结构并给出修改意见。

核心骨架如下：

workers = { "planner": create_agent(planner_system_prompt, planner_tools), "researcher": create_agent(researcher_system_prompt, researcher_tools), "writer": create_agent(writer_system_prompt, writer_tools), "reviewer": create_agent(reviewer_system_prompt, reviewer_tools), } async def run(task: TaskRequest, store) -> str: state = TaskState.resume(task.task_id, store) or TaskState(task_id=task.task_id, plan=[]) if not state.done or "plan" not in state.outputs: plan = await call(workers["planner"], task.description) state.plan = plan.get("steps", []) state.outputs["plan"] = plan state.done.add("plan") state.checkpoint(store) notes = [] for step in state.plan: if step["id"] in state.done: notes.append(state.outputs[step["id"]]) continue note = await call(workers["researcher"], step, resources=task.resources, state=state) notes.append(note) state.outputs[step["id"]] = note state.done.add(step["id"]) state.checkpoint(store) for round_no in range(MAX_REVISION): draft_key = f"draft_{round_no}" if draft_key in state.outputs: draft = state.outputs[draft_key] else: draft = await call(workers["writer"], notes) state.outputs[draft_key] = draft state.checkpoint(store) verdict = await call(workers["reviewer"], draft) state.outputs[f"verdict_{round_no}"] = verdict state.checkpoint(store) if verdict.get("approved"): return draft notes.append({"type": "revision_feedback", "feedback": verdict.get("feedback")}) return state.outputs[f"draft_{MAX_REVISION - 1}"]

这段代码把前文的原则都落实了：状态在外部存储、每一步产出落盘、断点续跑无需重来、修订轮次有上限兜底。真实的调度系统里，call()内部会走队列加异步消费，这里为了演示只写了顺序调用。

4.3 优先级：并行不等于盲目并发

如果你是让 20 个检索 Agent 同时开跑，大概率几分钟后就会撞上模型的速率限制。并行度必须手动控制。

我用asyncio.Semaphore限流，并给每种 Agent 配置独立的并发上限：

async def call(agent_name: str, payload: dict, *, sem: asyncio.Semaphore) -> dict: async with sem: for attempt in range(MAX_RETRY): try: return await run_agent_once(workers[agent_name], payload) except TransientError: await asyncio.sleep(2 ** attempt) # 指数退避 except PermanentError: raise

调度层面的重试和业务层面的重试要分开。业务层面（比如某篇资料本身无法解析）重试 3 次无效就该标记失败并跳过，不能让它卡住整个任务。调度层面只处理瞬时错误，比如网络抖动、限流、超时。

4.4 容错与人工兜底

多 Agent 任务里最坑的一种情况是某个子任务反复失败但 Agent 一直假装成功。所以我在每个 Agent 的调用结果里强制校验产物 schema，schema 不对的直接视为失败，并给出失败原因，不允许模型用一句"已完成"蒙混过关。

同时我设置了错误边界：子任务连续失败 3 次即标记为skipped，在最终报告里保留一个"未能覆盖的主题"清单。这样不会因为一个子项挂了导致整个报告交付不了，而且人工能清楚地看到哪些部分需要补。带着警告交付，远好过交付一份看似完美实则缺料的文档。

读到这里，你会发现多 Agent 的核心工程其实不是"让模型更聪明"，而是建立一套流程确保即使某个环节不聪明，整体依然可控。

5. 避坑指南与高频问题实录

5.1 高频问题速查表

把最近几个项目里反复出现的问题整理成表，按出现频率排序：

5.2 一次真实返工复盘

印象最深的翻车事故，是在一个调研项目里我给六个 Agent 设计了对等协作模式。它们需要围绕技术选型互相提意见，我天真地以为模型会自动收敛。结果跑了三十分钟，两个 Agent 为了某篇资料的可信度反复辩论，上下文全浪费在这上面，最终因为 token 超限整个任务报废。

排查之后我做了两处修改：一是把对等协作改成 Supervisor 模式，收敛权交回给主管 Agent；二是要求所有中间讨论都以"结论 + 理由"的结构化形式落盘，一句话总结一条，不允许长篇大论。修改后再跑，同样的任务三分钟出结果，而且每一步都可以回溯是谁、在什么输入下得出这个结论。

这个教训我后来反复讲：多 Agent 的架构能力取决于它的约束能力。你给 Agent 的自由越多，它越容易把时间和 token 浪费在低价值的来回拉扯上。

5.3 上线前的五项检查

最后给一份我每次都要过一遍的清单：

• 预检最大 token：最坏情况下本次任务至少需要多少上下文，预留了多少余量。
• 断点续跑可测：手动杀掉运行中的任务，重新启动后能恢复进度。
• 单 Agent 先于多 Agent：一个任务拆成多 Agent 之前，先确认单 Agent 确实失败，而不是"觉得它不行"。
• 最小人工介入点：至少在最终交付前保留一个审阅入口。
• 成本可观测：每次 Agent 调用都记录输入输出 token 与耗时，否则上线后成本失控都追不到源头。

我个人在实际操作中的体会是：所谓长任务和多 Agent 的最佳实践，本质就一句话——把 Agent 不可控的部分用工程手段锁死。状态落盘锁住记忆，结构化协议锁住沟通，重试和熔断锁住故障，人工兜底锁住质量。不要一上来就设计复杂拓扑，先加状态和断点，再加 worker 隔离，最后才考虑对等协作。多 Agent 不是炫技，是你对可控性的补偿。后面我会继续把调度策略、记忆管理和评估体系这几块的实战经验整理出来。