企业官网建设流程全解析

1. 从"agency-agents"这个标题说起：一个多智能体协作框架的完整拆解

第一次看到"agency-agents"这个命名，我的直觉是：这是一个把"代理"和"智能体"两个概念揉在一起的项目。拆开看，"agency"在软件语境里通常指"代理机构"或"代理能力"，而"agents"则是"智能体"的复数形式。合在一起，它指向的是一个多智能体协作系统——让多个具备独立决策能力的智能体，像一家代理公司里的不同角色一样，各司其职、互相配合，共同完成一个复杂任务。

这个方向最近一年在开发者圈子里讨论度非常高。原因很直接：单个大模型调用已经很难满足真实业务场景的需求了。你让一个模型既做需求分析、又写代码、又做测试、又写文档，它会在某个环节掉链子。但如果把任务拆开，让不同的智能体分别负责不同环节，每个智能体只专注自己那一块，整体输出的质量会明显提升。agency-agents 要解决的，正是这个"分工与协作"的问题。

这篇文章适合谁看？如果你正在做 AI 应用开发，尤其是涉及任务编排、自动化流程、多角色协作的场景，那这篇内容会对你有直接帮助。如果你只是听说过"多智能体"这个词但没动手做过，我也会从最基础的设计思路讲起，把每个环节的取舍逻辑说清楚。全文基于我对这类系统的实际开发经验来写，涉及具体实现的部分会给出可参考的代码结构和参数配置。

2. 整体架构设计：为什么这样拆分智能体

2.1 核心设计思路：角色化分工而非功能化分工

很多人在设计多智能体系统时，第一反应是按"功能"来拆分——一个负责搜索、一个负责计算、一个负责生成。这种拆法在简单场景下能用，但一旦任务变复杂，就会出现问题：功能边界模糊，智能体之间互相推诿，或者某个智能体不知道该在什么时候介入。

agency-agents 的思路更接近真实公司的组织方式——按角色拆分。每个智能体有自己的"职位描述"，包括职责范围、输入输出规范、以及和其他角色的协作关系。比如在一个内容生产流程里，会有"策划者"负责拆解需求、"执行者"负责具体产出、"审核者"负责质量把关。这种设计的好处是，每个智能体的行为逻辑更接近人类的协作模式，调试和扩展都更直观。

我实测下来，角色化分工在任务链路超过三步之后，优势会非常明显。因为每个角色的输入输出都是明确的，你可以单独替换某个角色而不影响整体流程。比如把"审核者"从一个简单规则引擎换成一个更强的模型，其他部分完全不用动。

2.2 通信机制选型：消息队列 vs 直接调用

智能体之间怎么传递信息，这是架构设计里第一个要拍板的事。常见方案有两种：一种是直接函数调用，A 智能体处理完直接把结果传给 B；另一种是引入消息队列，所有智能体通过队列异步通信。

直接调用的优点是简单、延迟低、调试方便。缺点是耦合度高，一旦某个环节卡住，整条链路都会阻塞。消息队列的优点是解耦彻底、支持异步、方便做重试和监控。缺点是引入了额外的基础设施，调试链路变长。

agency-agents 在这块的选择是混合模式：核心链路用直接调用保证响应速度，非关键路径（比如日志记录、结果归档）走消息队列。这个取舍的逻辑是——用户等待的环节不能有额外延迟，但后台处理可以慢慢来。具体实现上，我用的是一个轻量的内存队列加一个持久化队列的组合，内存队列处理实时消息，持久化队列做兜底。

注意：如果你的任务链路里有超过 5 个智能体，强烈建议至少把日志和监控走异步。否则一旦某个智能体出问题，你很难定位是哪一步出的错。

2.3 状态管理：共享内存还是各自维护

多智能体系统里，状态管理是最容易出 bug 的地方。每个智能体都有自己的上下文，但有些信息需要全局共享——比如当前任务的进度、已经产出的中间结果、用户的原始需求。

agency-agents 采用的是分层状态管理：每个智能体维护自己的私有状态（比如当前的思考过程、临时变量），同时有一个全局的共享状态存储（比如任务 ID、已完成步骤、关键产出）。私有状态用对象内部属性存，共享状态用一个独立的 Store 来管理。

这样做的好处是，智能体之间的状态隔离清晰，不会出现 A 改了变量导致 B 行为异常的情况。同时，共享状态可以持久化，方便做断点续跑和故障恢复。我在实际项目里踩过一个坑：早期把所有状态都放在一个全局对象里，结果两个智能体同时读写同一个字段，出现了竞态条件，排查了大半天。后来改成分层管理，这类问题就再也没出现过。

3. 核心智能体的实现细节与实操要点

3.1 策划者智能体：任务拆解的逻辑与边界

策划者（Planner）是整个系统的入口，它的职责是把用户的模糊需求拆解成可执行的具体步骤。这个环节的质量直接决定了后续所有智能体的工作效率。

实现上，策划者的核心是一个任务分解提示词模板。模板里需要明确几件事：任务的最终目标是什么、有哪些约束条件、可用的执行角色有哪些、每个步骤的预期输出格式是什么。我常用的模板结构是这样的：

PLANNER_PROMPT = """ 你是一个任务策划专家。请将以下需求拆解为可执行的步骤序列。 需求：{user_request} 可用角色： - 执行者：负责具体内容产出 - 审核者：负责质量检查 - 归档者：负责结果整理 输出格式要求： 1. 每个步骤必须指定负责角色 2. 每个步骤必须有明确的输入和输出 3. 步骤之间如果有依赖关系，必须标注 请以 JSON 格式输出步骤列表。 """

这里有个关键细节：步骤的粒度控制。拆得太粗，执行者不知道从哪下手；拆得太细，智能体之间的通信开销会急剧增加。我的经验是，每个步骤的工作量控制在"一个智能体一次调用能完成"的范围内。如果某个步骤需要多次调用，就应该继续拆。

另一个容易忽略的点是失败处理。策划者不仅要拆步骤，还要为每个步骤定义"什么算成功、什么算失败、失败了怎么办"。我通常会在步骤定义里加一个fallback字段，指定失败后的备选方案。比如"如果搜索不到相关信息，则基于已有知识生成"。

3.2 执行者智能体：工具调用与结果校验

执行者是真正干活的角色，它的核心能力是工具调用。一个执行者通常需要访问外部工具——比如搜索引擎、数据库、代码执行环境、文件系统。agency-agents 里，每个执行者都绑定了一组工具，工具的定义遵循统一的接口规范。

工具调用的实现要点有三个：

第一，参数校验必须在调用前完成。我见过太多因为参数格式不对导致工具调用失败的案例。解决方案是在工具定义里加上参数 schema，调用前先做一次校验。比如搜索工具的参数必须是字符串且长度不超过 200 字符，不满足就直接返回错误，而不是把错误参数传给工具。

第二，结果解析要容错。外部工具返回的数据格式可能不稳定，今天返回 JSON，明天返回纯文本。执行者需要有一套容错解析逻辑，先尝试按预期格式解析，失败则降级到通用解析。

第三，调用结果要记录。每次工具调用的输入、输出、耗时、是否成功，都要记录下来。这些数据在排查问题时非常有用。我通常会把调用记录写到一个独立的日志文件里，格式是 JSON Lines，方便后续分析。

def execute_with_tool(tool_name, params, context): # 参数校验 if not validate_params(tool_name, params): return {"status": "error", "reason": "invalid_params"} # 调用工具 start_time = time.time() try: result = tools[tool_name].call(params) status = "success" except Exception as e: result = str(e) status = "error" # 记录日志 log_call({ "tool": tool_name, "params": params, "result": result, "status": status, "duration": time.time() - start_time }) return {"status": status, "data": result}

实操心得：工具调用的超时设置非常关键。我一般把默认超时设为 30 秒，对于搜索类工具设为 15 秒，对于代码执行类工具设为 60 秒。超时后不要直接失败，而是返回一个"超时"状态，让上层决定是重试还是降级。

3.3 审核者智能体：质量标准的量化与执行

审核者的职责是检查执行者的产出是否符合要求。这个环节最容易做成"走过场"——审核者随便看一眼就说通过了。要避免这个问题，关键是把质量标准量化。

我通常从三个维度定义质量标准：完整性（是否覆盖了所有要求点）、准确性（是否有事实错误或逻辑矛盾）、格式规范性（是否符合输出格式要求）。每个维度给出一个 0 到 1 的评分，加权求和得到总分。总分低于阈值（比如 0.7）则判定为不通过，需要执行者重新处理。

审核者的提示词里要明确列出检查清单。比如对于一篇文档，检查清单可能包括：是否包含所有章节、是否有错别字、数据引用是否准确、格式是否统一。清单越具体，审核结果越可靠。

REVIEWER_PROMPT = """ 请对以下内容进行质量审核。 内容：{content} 审核维度： 1. 完整性（0-1分）：是否覆盖了所有要求点 2. 准确性（0-1分）：是否有事实错误或逻辑矛盾 3. 格式规范性（0-1分）：是否符合输出格式要求 请逐项打分并给出理由，最后计算加权总分。 权重：完整性 0.4，准确性 0.4，格式规范性 0.2。 输出格式： { "completeness": {"score": 0.9, "reason": "..."}, "accuracy": {"score": 0.8, "reason": "..."}, "format": {"score": 1.0, "reason": "..."}, "total": 0.88, "passed": true } """

这里有个经验：审核者不要和执行者用同一个模型。如果条件允许，用不同厂商或不同版本的模型来做审核，能发现更多问题。因为同一个模型往往有相同的盲区，自己审自己很难发现错误。

4. 完整实操流程：从零搭建一个可运行的多智能体系统

4.1 环境准备与依赖安装

先把基础环境搭起来。我用的技术栈是 Python 3.10 以上，核心依赖包括：一个 HTTP 客户端（用于调用模型 API）、一个异步框架（用于并发处理）、一个配置管理库。具体安装命令如下：

pip install httpx asyncio aiofiles pydantic python-dotenv

配置管理用.env文件来存 API 密钥和模型参数，不要硬编码在代码里。一个典型的配置结构是这样的：

MODEL_API_BASE=https://api.example.com/v1 MODEL_API_KEY=your_key_here PLANNER_MODEL=model-large EXECUTOR_MODEL=model-medium REVIEWER_MODEL=model-large MAX_RETRIES=3 TIMEOUT_SECONDS=30

注意：模型选择上，策划者和审核者建议用能力更强的模型，执行者可以用中等模型。因为策划和审核对逻辑推理要求高，执行对生成速度要求高。这个搭配在成本和效果之间比较平衡。

4.2 智能体基类的设计与实现

所有智能体共享一个基类，把公共逻辑抽出来。基类负责的事情包括：模型调用、重试逻辑、日志记录、状态管理。子类只需要实现自己的核心处理方法。

class BaseAgent: def __init__(self, name, model, system_prompt): self.name = name self.model = model self.system_prompt = system_prompt self.private_state = {} async def call_model(self, messages, retries=3): for attempt in range(retries): try: response = await model_client.chat( model=self.model, messages=messages, timeout=TIMEOUT_SECONDS ) return response except Exception as e: if attempt == retries - 1: raise await asyncio.sleep(2 ** attempt) async def process(self, input_data, shared_state): raise NotImplementedError

这个基类里，重试逻辑用的是指数退避——第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒。这样能有效应对临时的网络抖动或服务限流。

4.3 编排器的实现：任务调度与异常处理

编排器（Orchestrator）是整个系统的中枢，负责按顺序调用各个智能体、传递数据、处理异常。它的核心逻辑是一个循环：从策划者拿到步骤列表，然后逐步执行每个步骤，每步执行完更新共享状态。

class Orchestrator: def __init__(self, planner, executor, reviewer): self.planner = planner self.executor = executor self.reviewer = reviewer self.shared_state = {} async def run(self, user_request): # 第一步：策划 plan = await self.planner.process(user_request, self.shared_state) self.shared_state["plan"] = plan # 第二步：逐步执行 for step in plan["steps"]: result = await self.executor.process(step, self.shared_state) # 第三步：审核 review = await self.reviewer.process(result, self.shared_state) if not review["passed"]: # 审核不通过，重试 result = await self.executor.process( step, self.shared_state, feedback=review["reason"] ) self.shared_state[f"step_{step['id']}_result"] = result return self.shared_state

异常处理上，我设置了三个级别的容错：单步重试（最多 3 次）、步骤降级（用备选方案替代）、整体中断（返回已完成的部分结果）。大部分情况下，单步重试就能解决问题。如果某个步骤连续失败，就触发降级逻辑。

4.4 参数调优与性能优化

系统跑起来之后，性能调优是下一步。我实测下来，影响性能的主要有三个参数：并发数、超时时间、重试次数。

并发数方面，如果步骤之间没有依赖关系，可以并发执行。我一般把并发数设为 3 到 5，再高的话模型 API 容易限流。超时时间前面说过，按工具类型分别设置。重试次数设为 3 次比较合适，再多的话整体延迟会明显增加。

还有一个容易被忽略的优化点：缓存。如果某个步骤的输入和之前某次执行的输入相同，可以直接复用之前的结果。我在执行者里加了一个简单的缓存层，用输入内容的哈希值作为 key，命中率大概在 20% 左右，对整体速度提升很明显。

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

5.1 智能体"卡死"或无限循环的排查

这是最常见的问题。表现是某个智能体反复调用同一个工具，或者两个智能体互相等待。根本原因通常是终止条件不明确。

排查思路：先看日志，确认是哪个智能体在循环。然后检查它的提示词里有没有明确的终止条件。比如执行者的提示词里应该写"如果连续两次调用工具都没有获得新信息，则停止并返回当前结果"。审核者的提示词里应该写"最多审核两轮，两轮后无论结果如何都返回"。

我踩过的一个坑是：策划者拆出的步骤里有一个"等待审核通过"，但审核者又依赖执行者的输出，形成了循环依赖。后来在编排器里加了一个依赖检查，发现循环依赖就直接报错，而不是让它跑下去。

5.2 输出格式不稳定的处理方案

模型输出格式不稳定是另一个高频问题。你要求它输出 JSON，它有时候输出 JSON，有时候输出带 markdown 代码块的 JSON，有时候还加一段解释文字。

解决方案是多层解析：第一层尝试直接解析 JSON；第二层用正则提取代码块里的 JSON；第三层用正则提取花括号内容；第四层如果都失败，就把原始输出传给一个"格式修复"提示词，让模型重新格式化。

def parse_json_safely(text): # 第一层：直接解析 try: return json.loads(text) except: pass # 第二层：提取代码块 match = re.search(r'```(?:json)?\s*(.*?)```', text, re.DOTALL) if match: try: return json.loads(match.group(1)) except: pass # 第三层：提取花括号 match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group(0)) except: pass return None

实操心得：与其在解析上做各种容错，不如在提示词里就把格式要求写死。我通常会在提示词末尾加一句"只输出 JSON，不要有任何其他文字"。这句话能减少 80% 的格式问题。

5.3 常见问题速查表

5.4 几个我踩过的坑和对应的解法

第一个坑：共享状态被意外修改。早期我把共享状态直接传给每个智能体，结果某个智能体在处理时不小心改了共享状态里的字段，导致后续智能体拿到错误数据。解法是传给智能体的是共享状态的深拷贝，智能体只能通过专门的接口来更新共享状态。

第二个坑：日志太多导致磁盘写满。调试阶段我把所有调用都记了详细日志，跑了一天发现磁盘满了。解法是日志分级，INFO 级别只记关键信息，DEBUG 级别才记详细内容，并且设置日志轮转，最多保留 7 天。

第三个坑：模型版本升级导致行为变化。有一次模型服务商升级了模型版本，同样的提示词输出风格变了，导致审核通过率骤降。解法是在配置里锁定模型版本号，不要用"latest"这种浮动标签。

6. 扩展方向与个人经验分享

这套框架跑通之后，可以扩展的方向不少。我目前尝试过的有：接入更多类型的工具（比如数据库查询、文件操作）、增加智能体之间的协商机制（多个执行者投票决定结果）、以及引入长期记忆（把历史执行结果存起来供后续参考）。

协商机制这块值得多说一句。当任务比较复杂时，让多个执行者分别给出方案，然后由一个"决策者"来选择最优方案，效果比单个执行者直接产出要好。代价是成本翻倍，所以只在对质量要求极高的场景下用。

长期记忆的实现思路是：每次执行完成后，把"任务类型 + 输入特征 + 最终结果"存到一个向量数据库里。下次遇到类似任务时，先检索历史结果作为参考。这个做法在重复性任务上效果很好，能显著减少模型调用次数。

最后分享一个小技巧：给每个智能体起一个有意义的名字。不要用 agent1、agent2 这种，而是用 planner、executor、reviewer 这种。名字本身就能提醒你这个智能体的职责边界，在调试和日志分析时也能快速定位问题。这个习惯看起来不起眼，但在系统变复杂之后，能省下大量排查时间。