企业官网建设流程全解析

1. 从单兵作战到团队协作：为什么要把 Agent 组织起来

很多人第一次接触 Agent 这个概念，都是从一个简单的场景开始的：装好一个框架，配好 API Key，写一段提示词，然后看着它在终端里回答问题、调用工具、完成一个小任务。这个阶段很兴奋，但也很短暂。因为很快你就会发现，单个 Agent 的能力边界非常明显——它记不住太长的上下文，它没法同时处理多个方向的任务，它在遇到复杂问题时容易陷入死循环，它甚至会在同一个错误上反复撞墙。

我自己最开始也是这样，装了三四个不同的 Agent 框架，每个都跑通了 demo，每个都能完成一些基础任务，但真正想把它们用起来解决实际问题的时候，就发现单打独斗的效率极低。比如我让一个 Agent 帮我整理一份技术调研报告，它会先去搜索，然后整理，然后写摘要，整个过程串行执行，中间任何一步卡住，后面全部停摆。更麻烦的是，当任务涉及多个领域时，一个 Agent 既要懂代码又要懂文案还要懂数据分析，提示词写得再长也覆盖不全。

这就是“把 Agent 组织起来干活”这个想法诞生的背景。不是让一个 Agent 变得更强，而是让多个各有所长的 Agent 像一个小团队一样协作。有的负责规划，有的负责执行，有的负责检查，有的负责汇总。每个 Agent 的职责单一，提示词可以写得很精炼，工具配置也可以很有针对性。整体上，这个系统的鲁棒性和吞吐量都比单 Agent 方案高出一个量级。

这篇文章主要面向已经装好至少一个 Agent 框架、跑通过基础对话和工具调用的读者。如果你还没到这一步，建议先把单个 Agent 的安装、配置、工具调用跑通，再来看多 Agent 协作的内容。因为多 Agent 系统的调试复杂度是单 Agent 的好几倍，基础不牢会非常痛苦。

我会从整体设计思路开始讲，然后拆解核心细节，接着给出完整的实操过程，最后分享我在排查问题时积累的经验。整个过程中涉及的工具选型、参数配置、代码结构，都会给出具体的理由和可复现的方案。

2. 多 Agent 协作的整体设计与思路拆解

2.1 为什么选择“角色分工”而不是“模型堆叠”

很多人想到多 Agent，第一反应是搞一个超级模型，或者把多个模型的输出简单拼接。这两种思路我都试过，效果都不理想。超级模型的问题在于成本和延迟，而且单一模型再强也有知识盲区。简单拼接的问题在于没有真正的协作，只是把多个回答放在一起，信息冗余严重，还容易互相矛盾。

真正有效的多 Agent 系统，核心在于角色分工。每个 Agent 有明确的职责边界、独立的系统提示词、专属的工具集，以及清晰的输入输出规范。这就像一个小型创业团队：有人负责产品规划，有人负责技术实现，有人负责质量检查，有人负责对外沟通。每个人不需要是全才，但组合起来能覆盖整个业务流程。

我目前使用的架构是四层结构：规划层、执行层、检查层、汇总层。规划层负责把用户需求拆解成可执行的任务列表；执行层根据任务类型调用不同的专业 Agent；检查层对执行结果做质量把关；汇总层把多个执行结果整合成最终输出。这个结构不是固定的，你可以根据任务复杂度增减层级，但核心思想是“分工明确、接口清晰”。

2.2 通信机制的选择：消息队列还是共享状态

多 Agent 系统最关键的工程决策之一，就是 Agent 之间怎么通信。我试过三种方案：直接函数调用、消息队列、共享状态。

直接函数调用最简单，Agent A 直接调用 Agent B 的函数，同步等待结果。这种方式适合简单的串行流程，但一旦某个 Agent 执行时间过长，整个系统就会阻塞。而且错误处理很麻烦，一个环节抛异常，整个调用链就断了。

消息队列方案更灵活，每个 Agent 监听自己的队列，收到消息就处理，处理完把结果发到下一个队列。这种方式解耦彻底，可以异步执行，也方便做重试和监控。但引入消息队列意味着额外的运维成本，对于单机环境来说有点重。

共享状态方案是我目前最推荐的。所有 Agent 共享一个状态存储（可以用内存字典、Redis、或者简单的 JSON 文件），每个 Agent 从状态中读取自己需要的输入，处理完后把输出写回状态。这种方式既解耦又轻量，调试的时候可以直接查看状态快照，非常直观。

我现在的做法是用一个 Python 字典作为共享状态，配合文件持久化。每个 Agent 是一个独立的函数或类，接收状态对象，返回更新后的状态。调度器负责按顺序或按条件调用各个 Agent。这种设计在单机环境下足够用，迁移到分布式环境时也容易改造成消息队列模式。

2.3 工具集的设计原则：最小权限与最大复用

每个 Agent 的工具集需要仔细设计。我踩过的坑是给每个 Agent 配了太多工具，结果它不知道该用哪个，经常调用错误的工具，或者在一个简单任务上反复尝试不同的工具。

后来我总结出一个原则：每个 Agent 的工具集应该刚好覆盖它的职责范围，不多不少。比如规划 Agent 只需要一个“任务拆解”工具和一个“优先级排序”工具，不需要搜索、不需要代码执行、不需要文件读写。执行 Agent 根据具体类型配置工具，代码执行 Agent 只配代码运行和文件读写，搜索 Agent 只配搜索和网页抓取。

另一个原则是工具复用。不要把同一个功能在不同 Agent 里重复实现，而是抽成公共工具库，每个 Agent 按需引入。这样维护成本低，行为也一致。我现在的做法是建一个 tools 目录，里面按功能分类存放工具函数，每个 Agent 的配置文件里声明它需要哪些工具。

2.4 调度策略：串行、并行还是混合

调度策略直接影响系统的吞吐量和响应时间。我试过全串行、全并行、以及混合模式。

全串行最简单，一个 Agent 执行完再执行下一个。适合有严格依赖关系的任务链，但效率低。全并行适合独立任务，比如同时搜索多个关键词，但任务之间有依赖时就会出问题。

混合模式是我目前的主力方案。规划层和执行层之间是串行的，因为必须先有规划才能执行。执行层内部，如果多个任务之间没有依赖关系，就并行执行。检查层和汇总层是串行的，因为需要等所有执行结果都出来才能检查。

具体实现上，我用 Python 的 concurrent.futures 线程池来管理并行任务。每个执行 Agent 是一个独立的线程，主调度器等待所有线程完成后继续。这里要注意线程安全，共享状态的读写需要加锁，否则会出现数据竞争。

3. 核心细节解析与实操要点

3.1 Agent 角色定义与提示词工程

每个 Agent 的角色定义决定了它的行为边界。我通常从三个维度来定义：职责描述、输入输出规范、行为约束。

职责描述要具体，不能太宽泛。比如“你是一个规划助手”这种定义太模糊，Agent 不知道自己该做什么。更好的定义是“你负责把用户需求拆解成 3 到 7 个可独立执行的子任务，每个子任务必须包含明确的输入、输出和验收标准”。

输入输出规范要明确格式。我一般要求 Agent 输出 JSON 格式，包含固定的字段。比如规划 Agent 的输出格式是：

{ "tasks": [ { "id": "task_1", "type": "search", "description": "搜索 XXX 相关信息", "input": {"query": "XXX"}, "dependencies": [] } ] }

行为约束是防止 Agent 越界的关键。比如执行 Agent 不能修改任务列表，检查 Agent 不能直接修改执行结果，只能标记问题。这些约束要写在系统提示词里，并且在代码层面做校验。

提示词工程方面，我建议每个 Agent 的提示词控制在 500 到 1000 字之间。太短了约束不够，太长了模型容易忽略重点。关键约束要放在提示词的开头和结尾，中间放具体说明。我习惯用“必须”、“禁止”、“只能”这类强约束词，效果比“建议”、“尽量”好很多。

3.2 共享状态的结构设计

共享状态是整个系统的中枢，设计好坏直接影响调试难度和扩展性。我目前用的结构包含四个部分：任务元数据、执行上下文、中间结果、最终输出。

任务元数据记录任务的原始需求、创建时间、当前状态。执行上下文记录每个 Agent 的执行状态、开始时间、结束时间、错误信息。中间结果按任务 ID 存储每个子任务的输出。最终输出是汇总后的结果。

shared_state = { "meta": { "task_id": "xxx", "user_request": "原始需求", "status": "running", "created_at": "2024-01-01T00:00:00" }, "context": { "planner": {"status": "done", "duration": 1.2}, "executor_1": {"status": "running", "duration": 3.5} }, "intermediate": { "task_1": {"result": "...", "status": "done"}, "task_2": {"result": "...", "status": "failed", "error": "..."} }, "final": { "output": "...", "quality_score": 0.85 } }

这个结构的好处是信息分层清晰，调试的时候可以快速定位问题。比如某个任务失败了，直接看 intermediate 里对应任务的状态和错误信息。某个 Agent 执行时间过长，看 context 里的 duration。

状态持久化我建议每个阶段结束后写一次文件，格式用 JSON。这样即使程序崩溃，也能从上次的状态恢复。文件命名带上时间戳和任务 ID，方便追溯。

3.3 错误处理与重试机制

多 Agent 系统的错误处理比单 Agent 复杂得多，因为错误可能发生在任何一层，而且可能级联传播。我的策略是分层处理：Agent 内部错误自己处理，Agent 之间错误由调度器处理，系统级错误由监控模块处理。

Agent 内部错误包括 API 调用失败、工具执行异常、输出格式错误。这些错误在 Agent 内部捕获，根据错误类型决定是重试还是返回失败。API 调用失败通常重试两到三次，每次间隔递增。工具执行异常要看具体类型，如果是参数错误就直接返回失败，如果是临时故障就重试。输出格式错误最麻烦，我一般让 Agent 重新生成一次，如果还是格式错误就标记为失败。

Agent 之间的错误由调度器处理。调度器检查每个 Agent 的返回状态，如果失败，根据任务依赖关系决定是否继续。如果失败的任务是其他任务的前置依赖，就跳过依赖它的任务，并在最终输出里说明。如果失败的任务是独立的，就继续执行其他任务。

重试机制我建议设置最大重试次数和超时时间。最大重试次数一般设 3 次，超时时间根据任务类型定，搜索类任务 30 秒，代码执行类任务 60 秒，汇总类任务 120 秒。超过限制就标记为失败，不再重试。

3.4 日志与可观测性设计

多 Agent 系统的调试难度很高，没有良好的日志几乎没法排查问题。我的日志设计包含三个层次：Agent 级日志、调度级日志、系统级日志。

Agent 级日志记录每个 Agent 的输入、输出、执行时间、错误信息。格式用结构化日志，方便后续分析。我一般用 Python 的 logging 模块，配置 JSON 格式的输出。

调度级日志记录任务的分发、依赖关系的处理、并行任务的启动和完成。这些日志帮助理解系统的整体执行流程。

系统级日志记录资源使用情况、API 调用次数、错误率等指标。这些日志用于监控系统健康度，发现性能瓶颈。

日志存储我建议按天分文件，保留最近 7 天。关键错误日志单独存一份，方便快速检索。如果条件允许，可以接入简单的日志分析工具，做错误聚合和趋势分析。

4. 实操过程与核心环节实现

4.1 环境准备与依赖安装

我假设你已经有一个可用的 Python 环境，版本 3.9 以上。先创建一个虚拟环境，避免依赖冲突。

python -m venv agent_env source agent_env/bin/activate # Linux/Mac # 或者 agent_env\Scripts\activate # Windows

安装核心依赖。我用的框架是轻量级的，主要依赖 requests 做 API 调用，concurrent.futures 做并行调度，pydantic 做数据校验。

pip install requests pydantic python-dotenv

如果你用的 Agent 框架有特定依赖，按框架文档安装。我建议把依赖写进 requirements.txt，方便复现。

pip freeze > requirements.txt

环境变量配置用 .env 文件，存放 API Key、模型名称、超时时间等。不要硬编码在代码里，方便切换环境。

API_KEY=your_key_here MODEL_NAME=your_model MAX_RETRIES=3 TIMEOUT=60

4.2 基础 Agent 类的实现

先定义一个基础 Agent 类，封装通用的 API 调用、重试、日志逻辑。所有具体 Agent 继承这个类。

import os import json import time import logging from abc import ABC, abstractmethod from typing import Dict, Any class BaseAgent(ABC): def __init__(self, name: str, config: Dict[str, Any]): self.name = name self.config = config self.logger = logging.getLogger(f"agent.{name}") self.max_retries = int(os.getenv("MAX_RETRIES", 3)) self.timeout = int(os.getenv("TIMEOUT", 60)) @abstractmethod def build_prompt(self, state: Dict[str, Any]) -> str: pass @abstractmethod def parse_response(self, response: str) -> Dict[str, Any]: pass def call_model(self, prompt: str) -> str: for attempt in range(self.max_retries): try: # 这里替换成你实际使用的模型调用方式 response = self._invoke_api(prompt) return response except Exception as e: self.logger.warning(f"Attempt {attempt+1} failed: {e}") if attempt == self.max_retries - 1: raise time.sleep(2 ** attempt) return "" def _invoke_api(self, prompt: str) -> str: # 具体 API 调用逻辑，根据你的框架实现 pass def run(self, state: Dict[str, Any]) -> Dict[str, Any]: start_time = time.time() self.logger.info(f"{self.name} started") try: prompt = self.build_prompt(state) response = self.call_model(prompt) result = self.parse_response(response) state["context"][self.name] = { "status": "done", "duration": time.time() - start_time } return result except Exception as e: self.logger.error(f"{self.name} failed: {e}") state["context"][self.name] = { "status": "failed", "duration": time.time() - start_time, "error": str(e) } raise

这个基类封装了重试、日志、状态更新。具体 Agent 只需要实现 build_prompt 和 parse_response 两个方法。

4.3 规划 Agent 的实现

规划 Agent 负责把用户需求拆解成任务列表。提示词要强调任务的可执行性和依赖关系。

class PlannerAgent(BaseAgent): def build_prompt(self, state: Dict[str, Any]) -> str: user_request = state["meta"]["user_request"] return f"""你是一个任务规划专家。请把以下需求拆解成 3 到 7 个可独立执行的子任务。 需求：{user_request} 输出格式必须是 JSON，包含 tasks 数组，每个任务包含： - id: 任务唯一标识 - type: 任务类型（search/code/analysis/summary） - description: 任务描述 - input: 任务输入参数 - dependencies: 依赖的任务 id 列表 只输出 JSON，不要有其他内容。""" def parse_response(self, response: str) -> Dict[str, Any]: # 清理可能的 markdown 代码块标记 cleaned = response.strip().strip("```json").strip("```").strip() data = json.loads(cleaned) if "tasks" not in data: raise ValueError("Missing tasks field") return data

这里要注意，模型输出经常带有 markdown 代码块标记，解析前要清理。另外要校验 tasks 字段是否存在，避免后续流程出错。

4.4 执行 Agent 的实现

执行 Agent 根据任务类型有不同的实现。我以搜索 Agent 为例。

class SearchAgent(BaseAgent): def build_prompt(self, state: Dict[str, Any]) -> str: task = state["current_task"] return f"""你是一个信息搜索专家。请根据以下任务描述，整理出关键信息。 任务：{task['description']} 输入参数：{json.dumps(task['input'], ensure_ascii=False)} 输出格式为 JSON，包含 summary 和 sources 两个字段。 summary 是信息摘要，sources 是来源列表。""" def parse_response(self, response: str) -> Dict[str, Any]: cleaned = response.strip().strip("```json").strip("```").strip() return json.loads(cleaned)

代码执行 Agent 类似，但需要配置文件读写和代码运行工具。这里要注意安全，代码执行要在沙箱环境里，限制文件访问范围。

4.5 调度器的实现

调度器负责按依赖关系执行任务，管理并行和串行。

import concurrent.futures class Orchestrator: def __init__(self, agents: Dict[str, BaseAgent]): self.agents = agents self.logger = logging.getLogger("orchestrator") def execute(self, state: Dict[str, Any]) -> Dict[str, Any]: tasks = state["intermediate"]["tasks"] completed = set() failed = set() while len(completed) + len(failed) < len(tasks): ready = [ t for t in tasks if t["id"] not in completed and t["id"] not in failed and all(d in completed for d in t["dependencies"]) ] if not ready: # 没有可执行的任务，说明有循环依赖或前置失败 break with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: futures = {} for task in ready: agent = self.agents[task["type"]] state["current_task"] = task future = executor.submit(agent.run, state) futures[future] = task for future in concurrent.futures.as_completed(futures): task = futures[future] try: result = future.result() state["intermediate"][task["id"]] = { "result": result, "status": "done" } completed.add(task["id"]) except Exception as e: state["intermediate"][task["id"]] = { "status": "failed", "error": str(e) } failed.add(task["id"]) return state

这个调度器每次找出所有依赖已满足的任务，并行执行。执行完后更新状态，继续下一轮。注意这里用了线程池，共享状态的读写需要加锁，实际使用时要注意线程安全。

4.6 汇总 Agent 的实现

汇总 Agent 把所有执行结果整合成最终输出。

class SummaryAgent(BaseAgent): def build_prompt(self, state: Dict[str, Any]) -> str: results = state["intermediate"] results_text = json.dumps(results, ensure_ascii=False, indent=2) return f"""你是一个报告汇总专家。请把以下多个子任务的结果整合成一份完整的报告。 子任务结果： {results_text} 要求： 1. 按逻辑顺序组织内容 2. 标注每个结论的来源任务 3. 如果有任务失败，在报告中说明 4. 输出格式为 Markdown""" def parse_response(self, response: str) -> Dict[str, Any]: return {"output": response}

汇总 Agent 的提示词要强调逻辑组织和来源标注，这样最终报告的可信度更高。

4.7 完整流程串联

把所有组件串起来，主流程如下：

def main(user_request: str): state = { "meta": { "task_id": f"task_{int(time.time())}", "user_request": user_request, "status": "running" }, "context": {}, "intermediate": {}, "final": {} } # 初始化 Agent agents = { "planner": PlannerAgent("planner", {}), "search": SearchAgent("search", {}), "code": CodeAgent("code", {}), "summary": SummaryAgent("summary", {}) } # 规划 planner = agents["planner"] plan_result = planner.run(state) state["intermediate"]["tasks"] = plan_result["tasks"] # 执行 orchestrator = Orchestrator(agents) state = orchestrator.execute(state) # 汇总 summary = agents["summary"] final_result = summary.run(state) state["final"] = final_result state["meta"]["status"] = "done" # 持久化 with open(f"state_{state['meta']['task_id']}.json", "w") as f: json.dump(state, f, ensure_ascii=False, indent=2) return state["final"]["output"]

这个流程跑通后，你就有了一个基础的多 Agent 协作系统。后续可以根据需要增加更多类型的 Agent，比如检查 Agent、优化 Agent、翻译 Agent 等。

5. 常见问题与排查技巧实录

5.1 Agent 输出格式错误怎么办

这是最常见的问题。模型有时候会忘记输出 JSON，或者输出带 markdown 标记的 JSON，或者字段名拼错。我的处理策略是三层防护。

第一层是提示词约束，明确要求只输出 JSON，不要有其他内容。第二层是解析前清理，去掉可能的代码块标记和多余空白。第三层是解析失败后重试，让模型重新生成一次。

如果重试后还是格式错误，就标记该任务失败，继续执行其他任务。不要让一个格式错误阻塞整个流程。

我踩过的坑是提示词里写了“输出 JSON”，但模型还是输出了带解释文字的 JSON。后来我在提示词里加了“只输出 JSON，不要有任何解释文字”，情况好了很多。另外用 pydantic 做字段校验，能提前发现字段缺失或类型错误。

5.2 任务依赖关系处理错误

依赖关系处理错误通常表现为死循环或者任务被跳过。死循环的原因是依赖关系有环，比如 A 依赖 B，B 依赖 A。任务被跳过是因为前置任务失败，但调度器没有正确处理。

我的解决方案是在规划阶段就检测依赖关系是否有环。用拓扑排序算法检查，如果有环就让规划 Agent 重新生成。调度器里也要加超时机制，如果一轮调度没有产生任何完成的任务，就跳出循环，标记剩余任务为失败。

另一个技巧是在任务 ID 里加入层级信息，比如 task_1_1 表示第一个任务的第一个子任务。这样依赖关系更清晰，也方便调试。

5.3 并行执行时的资源竞争

并行执行时多个 Agent 同时读写共享状态，容易出现数据竞争。表现是状态数据不一致，或者某些任务的输出丢失。

我的解决方案是给共享状态的读写加锁。Python 的 threading.Lock 可以解决这个问题。但要注意锁的粒度，太粗会影响并行效率，太细容易死锁。

我现在的做法是每个任务的结果单独存储，不共享可变对象。调度器收集所有结果后再统一更新状态。这样避免了大部分竞争问题。

另外要注意 API 调用的速率限制。并行调用太多会导致 API 返回 429 错误。我一般限制最大并行数为 4，并且在 API 调用失败时做指数退避重试。

5.4 常见问题速查表

5.5 独家避坑技巧

第一个技巧是给每个 Agent 的输出加一个“置信度”字段。模型自己评估对结果的把握程度，汇总的时候可以根据置信度加权。这个技巧在搜索结果质量参差不齐的时候特别有用。

第二个技巧是在规划阶段就让 Agent 预估每个任务的执行时间，调度器根据预估时间动态调整并行数。预估时间长的任务优先启动，避免最后卡在某个慢任务上。

第三个技巧是保留每次执行的完整状态快照，包括输入、输出、中间状态。这样出问题的时候可以精确复现，而不是靠猜。我一般保留最近 10 次执行的快照，超过就自动清理。

第四个技巧是给 Agent 设置“最大思考轮数”。有些 Agent 会在一个问题上反复思考，消耗大量 token。设置一个上限，超过就强制输出当前最佳结果。

第五个技巧是用小模型做规划，大模型做执行。规划任务相对简单，小模型足够用，而且速度快、成本低。执行任务复杂，需要大模型的能力。这样整体成本和延迟都能优化。

5.6 性能优化的几个方向

如果系统跑起来感觉慢，可以从几个方向优化。首先是减少不必要的 Agent 调用，有些简单任务不需要经过规划层，直接执行就行。其次是缓存重复的搜索结果，同样的查询不要重复调用 API。第三是并行化独立任务，这个前面已经讲了。第四是压缩提示词，去掉冗余的描述，减少 token 消耗。第五是选择更快的模型做简单任务，复杂任务再用大模型。

我实测下来，经过这些优化，整体执行时间能减少 40% 左右。特别是缓存和并行化，效果最明显。

5.7 扩展方向

这个基础框架跑通后，可以往几个方向扩展。一是增加更多类型的 Agent，比如翻译 Agent、格式化 Agent、审核 Agent。二是引入更复杂的调度策略，比如基于优先级的调度、基于资源可用性的调度。三是接入外部工具，比如数据库查询、API 调用、文件处理。四是做可视化界面，实时展示每个 Agent 的状态和输出。

我个人觉得最有价值的扩展方向是增加一个“学习”机制。每次执行完后，把成功的任务拆解和提示词保存下来，下次遇到类似任务时直接复用。这样系统会越用越聪明，而不是每次从零开始。

这个内容后续还可以这样扩展：把单机多 Agent 系统改造成分布式系统，每个 Agent 跑在不同的机器上，通过消息队列通信。这样能处理更大规模的任务，也能提高系统的容错性。不过分布式系统的复杂度会高很多，建议先把单机版本跑稳再说。