企业官网建设流程全解析

1. 从单兵作战到团队协作：为什么Agent架构必须进化

过去大半年，我一直在折腾各种智能体项目，从最简单的问答机器人到能自动完成多步任务的个人助理，踩过的坑比写过的代码行数还多。最开始那会儿，每个人都在聊“Personal Agent”——你给它一个目标，它自己拆解步骤、调用工具、循环执行，最后把结果丢回给你。听起来很美好对吧？但真正落地的时候，你会发现一个致命问题：单个Agent的能力边界，其实比我们想象的要窄得多。

我举个自己遇到的真实场景。当时我想做一个能自动完成“行业调研报告”的智能体，流程大概是：先搜索最近一个月的相关新闻，然后提取关键数据，接着对比历史趋势，最后生成一份带图表的总结。单Agent方案跑起来是这样的：它先调用搜索工具，拿到一堆链接，然后试图逐个打开、提取内容。但问题来了——搜索回来的信息量太大，上下文窗口直接爆掉；提取数据的时候，它又需要同时做“筛选”和“计算”，注意力被严重分散，最后生成的报告质量惨不忍睹。

这就是单Agent架构的典型困境：它既要当指挥官，又要当士兵。规划、执行、校验、修正，所有环节都压在一个模型实例上，上下文越跑越长，指令越叠越多，最后模型自己都“晕”了。我试过用更长的上下文窗口去硬扛，也试过把提示词写得极其精细，但效果提升非常有限。根本原因在于，单个Agent的“工作记忆”是有限的，它没法像人类团队那样，把不同职能拆给不同角色去并行处理。

于是，“Agent-to-Agent”这个思路就自然而然地浮出水面了。简单说，就是不再让一个Agent包揽所有事情，而是设计多个各有所长的Agent，让它们之间互相通信、协商、协作，共同完成一个复杂任务。这就像从“一个人开一家公司”变成“组建一个团队”——有人负责市场调研，有人负责数据分析，有人负责撰写报告，还有人负责质量审核。每个Agent只专注自己最擅长的那一块，通过消息传递来同步进度和结果。

这个转变带来的好处是立竿见影的。首先，上下文压力被分散了。搜索Agent只需要关心“怎么找到最相关的信息”，分析Agent只需要关心“怎么从结构化数据里算出趋势”，写作Agent只需要关心“怎么把结论表达清楚”。每个Agent的提示词可以写得更聚焦，工具调用也更精准。其次，容错和校验变得可行。你可以专门设一个“审核Agent”，它的唯一任务就是检查其他Agent的输出有没有逻辑漏洞或事实错误。这种“交叉验证”在单Agent模式下几乎不可能实现，因为它自己检查自己，很容易陷入盲区。

当然，多Agent协作也不是没有代价。最直接的问题就是通信开销。Agent之间要传递消息，消息格式怎么定？谁来维护共享状态？如果A Agent等B Agent的结果，但B Agent卡住了怎么办？这些在单Agent模式下不存在的问题，到了多Agent场景全冒出来了。我刚开始尝试的时候，就因为没设计好消息协议，导致两个Agent互相等待，直接死锁，排查了半天才发现是消息队列的确认机制没写对。

所以，从Personal Agent到Agent-to-Agent，本质上是从“单体架构”向“分布式架构”的演进。它不是为了赶时髦，而是被实际需求逼出来的。当你发现单个Agent再怎么优化也突破不了能力天花板时，拆分成多个专职Agent、让它们通过标准化协议协作，就是最自然的解法。接下来，我会详细拆解这套架构的设计思路、核心组件、实操步骤，以及我在这个过程中踩过的那些坑。

2. 多Agent协作的核心架构与通信机制

2.1 角色划分：到底需要几个Agent才够用

刚开始设计多Agent系统时，最容易犯的错就是“过度拆分”。我见过有人把一个简单的文本摘要任务拆成六个Agent：一个负责读文件，一个负责分段，一个负责提取关键词，一个负责生成摘要，一个负责检查语法，还有一个负责格式化输出。结果呢？通信开销比实际计算还大，整体耗时反而比单Agent多了三倍。

我的经验是：从最小可行角色集开始，按需增加。对于一个典型的“调研-分析-生成”类任务，三个Agent通常就够了：一个规划Agent，负责理解用户意图、拆解任务步骤、决定调用哪些下游Agent；一个执行Agent，负责实际调用工具（搜索、计算、读写文件等）；一个审核Agent，负责检查执行结果是否符合要求，如果不符合就打回重做。

这三个角色的划分逻辑是这样的：规划Agent相当于项目经理，它不需要知道具体怎么搜索，只需要知道“这个任务需要先搜索再分析”；执行Agent相当于一线员工，它拿到具体指令后，专心把这一件事做好；审核Agent相当于质检员，它不关心过程，只看结果对不对。三者职责清晰，边界明确，消息传递也简单。

如果你做的任务更复杂，比如需要多轮迭代优化，那可以再加一个记忆Agent，专门负责维护对话历史和中间结果，避免每个Agent都自己存一份上下文导致不一致。但记住，每增加一个Agent，你就多了一条通信链路，系统复杂度是指数级上升的。我个人的建议是：能用三个解决的，绝不用四个。

2.2 通信协议：Agent之间到底怎么“说话”

Agent之间怎么传递消息，这是多Agent系统最核心的技术点。我试过三种方案，各有优劣，下面直接上对比表格。

我最终选择的是基于消息队列的异步通信，具体用的是Redis的Stream数据结构。为什么不用更简单的共享内存？因为共享内存虽然写起来快，但一旦Agent数量超过两个，状态同步就会变成噩梦。你永远不知道哪个Agent在什么时候改了哪个变量，调试的时候只能靠打印日志，效率极低。

用消息队列的好处是，每个Agent都是一个独立的消息消费者，它们只关心自己订阅的消息类型。比如规划Agent往“task.plan”频道发一条消息，执行Agent订阅了这个频道，收到后就开始干活，干完往“task.result”频道发结果，审核Agent再订阅“task.result”。整个流程是松耦合的，任何一个Agent挂掉，其他Agent不会立刻崩溃，只是任务会卡住，你可以很容易地定位到是哪个环节出了问题。

消息格式我推荐用JSON，结构大概长这样：

{ "message_id": "uuid", "sender": "planner_agent", "receiver": "executor_agent", "task_type": "search", "payload": { "query": "最近一个月AI行业动态", "max_results": 10 }, "timestamp": "2025-01-01T00:00:00Z", "reply_to": "task.result" }

这里有几个关键字段需要特别注意。message_id用于去重和追踪，reply_to指定了回复的频道，这样执行Agent干完活就知道把结果发到哪里。task_type是路由依据，执行Agent根据这个字段决定调用哪个工具。这套格式看起来简单，但实际跑起来非常稳，我用了大半年没出过消息错乱的问题。

注意：消息队列一定要设置过期时间，否则如果某个Agent挂了，消息会一直堆积，最后把内存撑爆。我一般设24小时过期，超过这个时间的消息基本没有处理价值了。

2.3 状态管理：谁负责记住“我们做到哪了”

多Agent系统里，状态管理是最容易被忽视但又最致命的一环。单Agent模式下，状态就是它的对话历史，模型自己就维护了。但多Agent模式下，每个Agent都有自己的上下文，它们之间的状态必须有一个“唯一真相源”。

我的做法是引入一个共享状态存储，用Redis的Hash结构来存。每个任务有一个唯一的task_id，对应的Hash里存了当前步骤、已完成步骤、中间结果、错误信息等。所有Agent在开始工作前，先从这个Hash里读状态，工作完成后，再写回状态。这样即使某个Agent重启了，它也能从共享状态里恢复上下文，继续干活。

举个例子，规划Agent拆解完任务后，往状态里写：

task:12345:status = "executing" task:12345:steps = ["search", "analyze", "write"] task:12345:current_step = "search"

执行Agent完成搜索后，更新为：

task:12345:current_step = "analyze" task:12345:search_results = [...]

审核Agent检查时，如果发现search_results为空，就知道搜索环节出了问题，可以直接把状态改回current_step = "search"，让执行Agent重做。这种“状态驱动”的设计，让整个系统的行为变得可预测、可回溯。

这里有个坑我踩过：状态更新一定要加锁。Redis本身是单线程的，但如果你用多个命令去更新同一个Hash，中间可能会有其他Agent插进来读到了不一致的状态。我后来改用Lua脚本，把“读-判断-写”打包成一个原子操作，问题就解决了。

3. 从零搭建一个多Agent协作系统的实操步骤

3.1 环境准备与基础依赖安装

动手之前，先把基础环境搭好。我用的技术栈是Python 3.11 + Redis 7.0 + OpenAI API（你也可以换成任何兼容的模型接口）。为什么选Python？因为生态最全，各种Agent框架和工具库都是Python优先。Redis用来做消息队列和状态存储，轻量且稳定，单机跑完全够用。

安装依赖：

pip install redis openai pydantic tenacity

redis是Redis客户端，openai用来调模型，pydantic做消息格式校验，tenacity做重试。这四个库基本覆盖了核心需求，不需要引入太重的框架。我试过一些现成的多Agent框架，但要么太黑盒不好调试，要么抽象层次太高，改起来反而麻烦。自己从零写，代码量其实不大，但可控性最强。

Redis的安装就不展开了，网上教程很多。启动后确认一下：

redis-cli ping # 返回 PONG 就说明OK

3.2 定义Agent基类与消息总线

先写一个Agent基类，把所有Agent共有的逻辑抽出来：连接Redis、订阅消息、发送消息、读写状态。这样每个具体Agent只需要实现自己的handle_message方法就行。

import json import uuid import redis from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, name, redis_url="redis://localhost:6379"): self.name = name self.redis = redis.from_url(redis_url, decode_responses=True) self.pubsub = self.redis.pubsub() def subscribe(self, channel): self.pubsub.subscribe(**{channel: self._message_handler}) def _message_handler(self, message): data = json.loads(message['data']) if data.get('receiver') != self.name: return self.handle_message(data) def send_message(self, receiver, task_type, payload, reply_to=None): msg = { "message_id": str(uuid.uuid4()), "sender": self.name, "receiver": receiver, "task_type": task_type, "payload": payload, "reply_to": reply_to or f"{self.name}.reply" } self.redis.publish(receiver, json.dumps(msg)) def get_state(self, task_id): state = self.redis.hgetall(f"task:{task_id}") return state def update_state(self, task_id, updates): self.redis.hset(f"task:{task_id}", mapping=updates) @abstractmethod def handle_message(self, message): pass def run(self): self.pubsub.run_in_thread(sleep_time=0.1)

这段代码的关键在于_message_handler里的过滤逻辑：每个Agent只处理receiver等于自己名字的消息。这样多个Agent可以订阅同一个频道，但只有目标Agent会响应，避免了消息被重复处理。

3.3 规划Agent的实现：任务拆解与调度

规划Agent是整个系统的入口，它接收用户请求，拆解成子任务，然后分发给执行Agent。它的核心逻辑是调用大模型做任务规划，然后把规划结果写入状态，并发送第一条执行消息。

class PlannerAgent(BaseAgent): def __init__(self, llm_client): super().__init__("planner_agent") self.llm = llm_client def handle_message(self, message): if message['task_type'] == 'user_request': task_id = message['payload']['task_id'] user_input = message['payload']['query'] # 调用模型拆解任务 plan = self._create_plan(user_input) # 写入初始状态 self.update_state(task_id, { "status": "executing", "steps": json.dumps(plan['steps']), "current_step": plan['steps'][0], "original_query": user_input }) # 发送第一条执行消息 self.send_message( receiver="executor_agent", task_type=plan['steps'][0], payload={"task_id": task_id, "query": user_input}, reply_to="executor.result" ) def _create_plan(self, user_input): prompt = f"""你是一个任务规划专家。请把以下用户请求拆解成3-5个步骤。 每个步骤必须是一个具体的动作，比如"search"、"analyze"、"write"。 用户请求：{user_input} 请以JSON格式返回，包含steps字段。""" response = self.llm.chat(prompt) return json.loads(response)

这里有个细节：规划Agent不直接调用执行工具，它只负责“想”，不负责“做”。这种职责分离让规划逻辑可以独立优化，比如你可以换一个更强的模型来做规划，而执行Agent继续用便宜的小模型，成本可控。

3.4 执行Agent的实现：工具调用与结果回传

执行Agent是真正干活的，它根据task_type决定调用哪个工具。我目前实现了三个工具：搜索、计算、文件读写。每个工具都是一个独立的函数，执行Agent根据消息里的task_type来路由。

class ExecutorAgent(BaseAgent): def __init__(self, tools): super().__init__("executor_agent") self.tools = tools # {"search": search_func, "analyze": analyze_func, ...} def handle_message(self, message): task_id = message['payload']['task_id'] task_type = message['task_type'] if task_type not in self.tools: self._send_error(task_id, f"未知任务类型: {task_type}") return try: result = self.tools[task_type](message['payload']) # 更新状态 self.update_state(task_id, { f"{task_type}_result": json.dumps(result), "current_step": task_type }) # 发送给审核Agent self.send_message( receiver="reviewer_agent", task_type="review", payload={"task_id": task_id, "step": task_type, "result": result}, reply_to="reviewer.result" ) except Exception as e: self._send_error(task_id, str(e)) def _send_error(self, task_id, error_msg): self.update_state(task_id, {"status": "error", "error": error_msg}) self.send_message( receiver="planner_agent", task_type="error", payload={"task_id": task_id, "error": error_msg} )

执行Agent的异常处理很重要。如果工具调用失败，它不能直接崩溃，而是要把错误信息写回状态，并通知规划Agent。规划Agent收到错误后，可以决定是重试、跳过还是终止任务。这种“错误上报”机制让系统有了基本的自愈能力。

3.5 审核Agent的实现：质量把关与打回重做

审核Agent的任务是检查执行结果是否合格。它的判断标准可以很简单，比如“搜索结果是否为空”、“分析结果是否包含数字”，也可以很复杂，比如调用模型来判断逻辑是否自洽。我目前用的是“规则+模型”混合策略：先跑一遍硬规则，如果规则通过，再让模型做一次快速审核。

class ReviewerAgent(BaseAgent): def __init__(self, llm_client): super().__init__("reviewer_agent") self.llm = llm_client def handle_message(self, message): task_id = message['payload']['task_id'] step = message['payload']['step'] result = message['payload']['result'] # 硬规则检查 if not result or (isinstance(result, list) and len(result) == 0): self._reject(task_id, step, "结果为空") return # 模型审核 review_prompt = f"""请判断以下{step}步骤的结果是否合格。 结果：{json.dumps(result)[:500]} 只需回答"合格"或"不合格"，并给出简短理由。""" review = self.llm.chat(review_prompt) if "合格" in review: self._approve(task_id, step) else: self._reject(task_id, step, review) def _approve(self, task_id, step): state = self.get_state(task_id) steps = json.loads(state['steps']) current_idx = steps.index(step) if current_idx + 1 < len(steps): next_step = steps[current_idx + 1] self.update_state(task_id, {"current_step": next_step}) self.send_message( receiver="executor_agent", task_type=next_step, payload={"task_id": task_id}, reply_to="executor.result" ) else: self.update_state(task_id, {"status": "completed"}) self.send_message( receiver="planner_agent", task_type="completed", payload={"task_id": task_id} ) def _reject(self, task_id, step, reason): self.update_state(task_id, {"last_error": reason}) self.send_message( receiver="executor_agent", task_type=step, payload={"task_id": task_id, "retry": True}, reply_to="executor.result" )

审核Agent的“打回重做”逻辑是整个系统的质量保障。如果没有它，执行Agent返回什么就是什么，错误会一路传递到最后。有了审核环节，每个步骤的输出都被检查过，最终结果的质量会稳定很多。当然，重试次数要设上限，我一般设3次，超过3次就标记为失败，避免无限循环。

4. 实际运行中遇到的典型问题与排查技巧

4.1 消息死锁：两个Agent互相等待怎么办

这是我遇到的第一个严重问题。当时规划Agent发了一条消息给执行Agent，执行Agent处理完后，把结果发给了审核Agent，审核Agent审核通过后，又发回给执行Agent让它做下一步。但执行Agent在等审核Agent的结果时，自己阻塞了，没法处理新消息，而审核Agent又在等执行Agent的确认，两边就卡死了。

排查这个问题的关键是看消息队列的堆积情况。我用redis-cli XLEN查看各个频道的消息数量，发现executor.result频道有消息堆积，但执行Agent没有消费。再一看执行Agent的日志，发现它卡在了一个同步等待上。

解决方案是所有Agent必须异步处理消息。具体来说，handle_message里不能有阻塞操作，如果确实需要等待某个结果，应该把“等待”变成“状态检查+定时轮询”，而不是直接阻塞。我后来把执行Agent改成了纯异步模式：收到消息后立刻返回，把耗时操作丢到后台线程池里，完成后通过回调发送结果。这样Agent永远不会阻塞，死锁问题就消失了。

实操心得：多Agent系统里，任何形式的同步等待都是定时炸弹。宁可多写几行异步回调代码，也不要图省事用同步调用。

4.2 上下文污染：Agent之间传递了错误的状态

有一次，我发现最终生成的报告里混入了上一个任务的数据。排查了半天，发现是共享状态没有清理干净。上一个任务完成后，task:{task_id}这个Hash没有被删除，下一个任务用了相同的task_id（因为我的ID生成逻辑有bug），导致读到了旧数据。

这个问题让我意识到，状态隔离必须从设计上保证。我的改进措施有三个：第一，task_id改用UUID，确保全局唯一；第二，任务完成后，把状态Hash标记为completed并设置24小时过期，而不是立即删除，方便事后排查；第三，每个Agent在读取状态时，先检查status字段，如果任务已经完成或失败，就不再处理。

def get_state(self, task_id): state = self.redis.hgetall(f"task:{task_id}") if state.get('status') in ['completed', 'failed']: return None # 任务已结束，不再处理 return state

4.3 模型输出格式不稳定：JSON解析失败怎么破

执行Agent和审核Agent都依赖模型返回结构化数据，但模型有时候会返回带Markdown代码块的JSON，有时候会多写几句解释，导致json.loads直接报错。我一开始用正则去提取，但正则写得太复杂，维护起来很痛苦。

后来我用了两个策略组合。第一，在提示词里明确要求“只返回JSON，不要任何其他文字”，并且给出示例格式。第二，用pydantic做校验和容错，如果解析失败，自动触发一次重试，重试时把错误信息也附上，让模型自己修正。

from pydantic import BaseModel, ValidationError from tenacity import retry, stop_after_attempt class PlanSchema(BaseModel): steps: list[str] @retry(stop=stop_after_attempt(3)) def parse_plan(response_text): # 尝试提取JSON text = response_text.strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0] try: data = json.loads(text) return PlanSchema(**data) except (json.JSONDecodeError, ValidationError) as e: raise ValueError(f"解析失败: {e}")

这套组合拳打下来，JSON解析成功率从最初的70%提升到了99%以上。剩下那1%基本是模型彻底跑偏了，重试也救不回来，那就只能标记失败，人工介入。

4.4 性能瓶颈：Agent数量多了之后延迟飙升

当我把Agent数量从3个增加到6个之后，整体任务耗时从平均15秒飙升到了2分钟。用Redis的MONITOR命令一看，发现消息在频道之间来回传递了几十次，每次传递都有网络往返开销。更糟糕的是，有些Agent处理一条消息要调用好几次模型，每次调用又是几秒的延迟，累积起来非常可观。

优化方向有两个。第一，合并消息。原来每个步骤完成都要发一条独立消息，后来我改成批量发送，比如执行Agent完成搜索和分析后，把两个结果打包成一条消息发给审核Agent，减少通信次数。第二，并行化。有些步骤之间没有依赖关系，比如“搜索新闻”和“搜索论文”可以同时进行，我就让执行Agent开两个线程并行处理，最后合并结果。

优化之后，6个Agent的任务耗时降到了25秒左右，虽然还是比3个Agent慢，但考虑到任务复杂度提升了，这个延迟是可以接受的。

5. 多Agent系统的扩展方向与个人经验总结

5.1 从固定流程到动态编排

我目前这套系统还是“静态编排”——规划Agent一次性拆解完所有步骤，然后按顺序执行。但实际任务往往需要根据中间结果动态调整。比如搜索发现信息不足，应该自动增加一轮搜索；分析发现数据异常，应该回溯检查原始数据。这些动态决策，静态编排做不到。

下一步我打算引入动态编排机制：规划Agent不再一次性输出所有步骤，而是只输出第一步，然后根据执行结果决定下一步做什么。这相当于把“规划”从一次性动作变成了持续过程。实现上，可以让规划Agent订阅执行Agent的结果消息，每次收到结果后重新规划下一步。这样灵活性大大提升，但代价是规划Agent的调用次数变多了，成本会上升。需要在灵活性和成本之间找平衡。

5.2 Agent之间的协商与投票机制

另一个有意思的方向是多Agent投票。当审核Agent拿不准一个结果是否合格时，它可以同时询问多个“专家Agent”，让它们各自给出判断，然后按多数票决定。这种机制在需要主观判断的场景下特别有用，比如“这段文字是否通顺”、“这个分析是否合理”。单个模型的判断可能有偏差，但多个模型投票可以显著提升准确率。

实现上，审核Agent可以往一个“vote”频道发消息，所有订阅了这个频道的专家Agent各自返回判断，审核Agent收集到足够多的票数后做决策。这里的关键是设置投票超时，不能无限等待，一般设5秒，超时就用已收到的票数做决定。

5.3 我踩过的那些坑与最终建议

回顾这大半年的折腾，有几个教训是花钱买来的。第一，不要过早优化。我一开始就想设计一个能支持上百个Agent的通用框架，结果抽象层写得太复杂，调试极其困难。后来推倒重来，从3个Agent的简单系统开始，跑通了再逐步扩展，效率反而高得多。第二，日志要打全。多Agent系统的调试难度是单Agent的十倍，每个消息的发送、接收、处理、结果都要打日志，最好带上message_id和task_id，这样出问题能快速定位。我后来用structlog统一了日志格式，排查效率提升明显。第三，状态一定要可观测。我专门写了一个小脚本，实时打印所有进行中任务的状态，相当于一个简易的监控面板。没有这个，你根本不知道系统内部在发生什么。

最后分享一个小技巧：在开发阶段，给每个Agent加一个“调试模式”，开启后所有消息都会额外抄送一份到一个debug频道，你用redis-cli subscribe debug就能看到完整的消息流转过程。这个技巧帮我省了无数个小时的排查时间。

这套多Agent协作系统目前还在持续迭代中，从最初的3个Agent扩展到了现在的6个，任务完成质量和稳定性都有了质的提升。如果你也在做类似的事情，我的建议是：先从单Agent跑通核心流程，遇到瓶颈再拆多Agent，不要为了多Agent而多Agent。架构是手段，不是目的。