企业官网建设流程全解析

做AI应用这些年，我越来越觉得“单智能体包打天下”这个思路在真实业务约束下并不可靠。最近我搭了一套内部代号叫agency-agents的模拟项目，核心就是让多个智能体像一个小团队一样分工协作。它解决的场景很典型：一次任务里既要做资料搜集，又要写方案，还要做质量审核，与其把这些逻辑全塞进一个超大提示词，不如拆成几个各有专职的小智能体，通过统一的消息通道互相配合。这篇文章会把设计思路、核心代码结构、实际跑通流程以及我踩过的坑都写出来，对正在研究多智能体协作的同学来说，应该能省下不少时间。

1. 整体设计与思路拆解

1.1 为什么要拆成多个智能体

这事情得从一次真实需求说起。我当时接到的需求是做一个“行业趋势调研”，输出一份带数据引用的报告草案。最早我试着让一个智能体一口气完成：先让它搜索行业新闻，再让它整理数据，再让它写报告，最后让它检查引用格式。结果很惨，模型不是丢数据，就是把编造的内容当作搜索结果，最离谱的一次是它为了凑字数，虚构了两家公司合作的消息。

问题不是某个模型能力不行，而是单个智能体在同一轮上下文里要承担的角色太多，推理注意力被稀释了。人的组织里为什么需要分工？因为每次决策都有一个认知边界。智能体也一样，当“检索、提炼、写作、校验”互相穿插时，模型很难在每一跳都保持高质量。

所以我把任务拆成四个角色：检索员、分析师、作者、校对员。每个角色是独立的智能体，有自己的系统提示词、可调用的工具、输入输出格式约定。这样每个智能体只需要做一件事，提示词能写得更具体，模型也更容易遵循。

你可能会问，这不就是几个独立的API调用拼起来吗？是，但不全是。关键是智能体之间有明确的消息语义和状态流转，不是散乱的多次调用。这一步从“函数调用”到“团队协作”的转变，才是多智能体框架真正有优势的地方。

这个设计模式覆盖的应用范围很广，内容生产、营销策划、数据分析、软件研发辅助都能用。只要任务本身可以拆成有先后依赖或者并行处理的子步骤，而且每个子步骤都有清晰输入输出，就值得用多智能体方案。尤其适合那些以前靠堆提示词、反复抽卡才能勉强跑通的复杂任务，换成多智能体之后，可控性会明显提升。

1.2 方案选型：集中调度还是自主协作

多智能体协作有几种常见实现方式。我最早想走自主协作模式：所有智能体都挂在同一个消息总线上，彼此订阅和发布事件，A发布了结果，B觉得跟自己有关，就自动接手。听着灵活，实际调试起来非常痛苦，因为你根本不知道一条消息到底触发了多少个智能体，也不知道某个环节为什么没人处理。

后来我换成集中调度模式，也叫Orchestrator模式。所有智能体自己不决定下一步做什么，而是由一个调度器读任务图，按依赖关系依次或并行调用。这样流程是显式的，失败了能快速定位是哪一步，超时了也能在调度器这一层统一处理。

对比两种方案，有几点体会：

• 自主协作适合探索性强、执行路径不确定的场景，但需要很强的消息契约和容错设计。
• 集中调度适合大部分业务流水线，尤其是我们这种“任务边界清楚、步骤前后依赖”的场景。
• 如果团队以后要做成通用平台，可以设计成组装式：底层支持事件总线，上层支持有向无环图调度。我在项目里只先实现了后者，避免一开始就背上分布式复杂度。

选集中调度的另一个原因是成本可控。自主协作模式里智能体之间会反复“商量”，每次商量都是一次大模型调用，消耗的是真金白银。用调度器控制流程，一次任务跑多少步、调用多少次模型，是可预测的。

影响范围上，集中调度模式更适合需要预算控制、服务等级协议和审计要求的团队。比如给客户交付项目，你总得告诉对方“这次任务总共调用了多少次模型、耗时多少”，集中调度天然能把这份账单直接打出来。自主协作模式在这件事上会非常吃力，因为调用次数和路径都不可控。

1.3 系统架构总览：从任务入口到存储的五层划分

这个项目整体分成五层。

最外层是入口，接收一个任务描述，通常是自然语言，比如“调研某新兴市场智慧零售的落地情况，并输出一份中文简报”。第二层是调度器，会做两件事：把任务拆成小步骤，按依赖关系排成执行计划。第三层是智能体池，里面注册了各个智能体对象，每个对象都持有自己的系统提示词和工具集。第四层是工具层，封装了外部检索API、向量数据库、文件读写、翻译等能力，只对智能体暴露统一接口。第五层是存储，用来保存每一步的原始输入、输出、耗时、token用量，方便回放和审计。

消息在这几层之间流转时，统一走一个消息对象，不搞自定义结构。哪怕两个智能体是同一个开发者写的，也别为了省事直接传字典，标准化的消息结构在后面做日志和重试时会省非常多事情。

从实现角度看，调度器不直接持有调用链，而是保存一个“步骤定义列表”，每个步骤包括 agent_name、input_key、output_key、retry_count、timeout。数据在步骤之间通过一个共享上下文对象读取和写入。这样既清晰，又能在任意一步结束后把整个上下文序列化出来做检查。

我在实际搭建时，先小步跑通一条链路，再慢慢加功能。刚开始只有一个入口脚本，连存储都省了，等链路稳定后才把存储、日志、审计补上。这个顺序很重要，如果你一上来就把架构搭得很重，迭代和调参都变得非常慢。

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

2.1 智能体角色怎么定义

角色定义是整个项目的基石。我总结下来的经验是：角色信息至少要包含五个字段，缺一个后面都会出问题。以检索员为例：

• name：agent-researcher，用于日志和调度。
• role：一句话描述职责，比如“负责抽取用户问题中的检索关键词并调用搜索工具”。
• system_prompt：详细说明它是谁、它要做什么、它不能做什么、输出格式必须是什么。
• tools：允许调用的工具白名单，比如 search_web、read_url，但不能调用 write_doc。
• result_schema：一个JSON Schema或一个Pydantic模型，用于输出校验。

很多半路出家做智能体的开发者容易忽略 result_schema，让模型自己自由发挥。自由发挥的结果就是解析报错，或者下游智能体根本看不懂上游文本。真正稳的做法是给每个智能体定义严格的输出Schema，并且要求输出纯JSON，同时开启强制JSON模式。

这里还有一个容易踩的细节：system_prompt 里不要写“你是团队里的助手”这种模糊话术，要写“当且仅当你需要查找最新资料时调用 search_web，禁止自行编造数据。如果你没有找到有效信息，请明确输出 empty_result 状态”。给模型一个“没有结果时该怎么办”的出口，比单纯叮嘱它别瞎编更有用。

角色定义这件事，直接影响整个链路的下限。如果角色定义模糊，后面加再多的校验和重试都只是补救，因为脏数据源头没堵住。我建议每个角色分配好后，先用一个最简单的测试任务验证输出格式，满意了再交给调度器编排，不要一上来就串全链路。

2.2 消息协议与通信规则

智能体之间通信如果靠自然语言文本直接拼，很快就乱。我在项目里定义了一个统一消息类，字段如下：

• message_id：全局唯一ID，用于追踪和幂等。
• sender：发送方 agent_name。
• receiver：接收方 agent_name，或者是 orchestrator。
• task：本次调用的任务描述，可能是原始任务的一部分，也可能是经过解析后的结构化指令。
• payload：上下文数据，通常是字典，存放上游输出。
• status：pending、running、succeeded、failed。
• error：出错时的异常信息或错误码。
• token_usage：记录本次调用消耗的输入/输出token数。

这个结构看起来简单，但把三个事情解决了：一是消息可追踪，出了问题直接把 message_id 拿去日志过滤器查，几秒钟就能看到整条链路；二是消息可重试，如果下游超时，调度器可以用同一 message_id 重启步骤而不会重复执行副作用；三是消息可审计，每一步谁处理的、花了多少token、结果是什么都一目了然。

通信规则上，我明确规定所有智能体不能相互之间直接发消息，只能通过调度器转发。这是刻意牺牲一点灵活性换来的可控性。如果某天你真需要两个智能体私下同步状态，可以在调度器里加一个“会议室”机制，把它们的对话内容记录成一组消息，本质上还是被调度器包住。

这套通信规则带来的直接好处是：新接一个智能体非常快，你不需要知道其他智能体是怎么写的，只需要知道消息协议长什么样。对一个小团队来说，这大大降低了协作开发的沟通成本。我曾经试过让两个智能体直接对话，结果双方天然形成了一个消息风暴，日志根本看不过来，从那以后我彻底倒向集中转发。

2.3 任务分解与结果聚合

任务分解不是拍脑袋，我是按“输出物反推”来做的。先定义整个任务的最终产物是什么，再往回拆：要产出最终报告，需要章节；要写章节，需要数据和案例；要拿数据和案例，需要检索和分析。每一步都是对下一步输出物的具体化。

比如“某海外新兴市场智慧零售简报”这个任务，我拆成这样：

1. 检索员搜索“新兴市场 智慧零售 落地案例”，整理出十个候选资料，输出带链接和摘要的列表。
2. 分析师读取这十个候选，筛选去重，提取三个关键趋势，输出结构化结论。
3. 作者读取趋势结论，按照预设模板编写简报正文。
4. 校对员核对正文中的事实陈述是否有来源支持，输出修订意见。
5. 调度器把修订意见合并给作者再跑一次，产出终稿。

结果聚合放在调度器里做。每个步骤的输出先写入共享上下文，某个字段依赖多个上游时，调度器只取依赖的key聚合，不把整个上下文传给模型。这能显著降低模型输入空间，减少干扰。

聚合时还有一个细节：要处理“上游输出为空”的情况。我写了一个 result_selector，优先取 status=succeeded 且结果非空的字段；如果所有上游都失败，整个任务直接短路返回失败报告，不往下游传空气。很多人容易漏掉这个边界，导致下游拿到一个空字典之后照样硬跑，最终输出一堆无意义内容。

任务分解的粒度也要控制。粒度太粗，单个智能体还是压力很大；粒度太细，调度开销和token浪费又会抵消收益。我的参考线是：每个智能体完成的任务不超过一个独立业务环节，输出信息不超过两三百个结构化字段。超过这个范围，我宁可多拆一个角色。

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

3.1 运行环境准备与依赖安装：一页纸清单

我先说下我这个项目的运行环境：操作系统是常见的Linux容器，Python版本3.10以上，核心依赖包括 pydantic（用来做输出Schema校验）、requests（调用外部API）、orjson（高性能JSON序列化）。如果你用兼容大模型接口，还需要一个API Key，放在环境变量里，不要写进代码。

安装命令也不复杂：

pip install pydantic requests orjson python-dotenv

然后创建一个.env文件：

LLM_API_BASE=https://your-endpoint.example.com/v1 LLM_API_KEY=sk-xxxxxxxx LLM_MODEL=model-name

注意那个API地址，你们内部服务或者第三方兼容服务都可以，但所有智能体共用同一个模型接入层。想让不同智能体用不同模型，只需要在Agent初始化时传入不同的model字段。

这里有一条实操经验：API Key和模型名称不要写死在代码里，也不要写死在系统提示词里。因为智能体角色定义经常要改，改配置文件总比重发代码要安全。我见过同事把模型名写在提示词里，换模型的时候还得逐字核对提示词，非常容易漏。

3.2 Agent基类实现：一个可继承的骨架

项目里所有智能体都继承同一个基类。基类做了三件事：格式化系统提示词、调用大模型接口、解析并校验输出。没有把业务逻辑塞在基类里，具体工具怎么用交给子类。这样新加一个智能体只要几分钟。

# agent_base.py import json import os import requests from abc import ABC, abstractmethod from typing import Any, Dict class Agent(ABC): def __init__(self, name: str, role: str, system_prompt: str, tools: list = None, result_schema: Any = None, model: str = None): self.name = name self.role = role self.system_prompt = system_prompt self.tools = tools or [] self.result_schema = result_schema self.model = model or os.getenv("LLM_MODEL") def build_user_message(self, task: str, context: Dict[str, Any]) -> str: # 具体子类可以覆盖，默认把任务和相关上下文序列化成JSON entire = {"task": task, "context": context} return json.dumps(entire, ensure_ascii=False) def call_llm(self, messages: list) -> str: endpoint = os.getenv("LLM_API_BASE") api_key = os.getenv("LLM_API_KEY") resp = requests.post( f"{endpoint}/chat/completions", headers={"Authorization": f"Bearer {api_key}"}, json={"model": self.model, "messages": messages, "temperature": 0.3}, timeout=60, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] def validate_result(self, text: str) -> Dict[str, Any]: if not self.result_schema: return {"raw_text": text} data = json.loads(text) return self.result_schema(**data).model_dump() def run(self, task: str, context: Dict[str, Any]) -> Dict[str, Any]: user_content = self.build_user_message(task, context) messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_content}, ] raw_output = self.call_llm(messages) return self.validate_result(raw_output)

基类里我把call_llm做成了标准HTTP调用。为什么要用 pydantic 来做输出Schema校验？因为模型偶尔会输出json格式不够规范，比如尾逗号、单引号，json.loads会直接报错。用pydantic可以先加载再校验字段，类型和必填都能兜住。

这里还有个经验：不要只在最后一步做输出校验，而要在每个智能体执行完都做。宁可让下游拿到一个“校验失败”的错误，也不要拿到一份看起来像那么回事、但字段缺失的脏数据。脏数据会静默地污染整条链路，错误则能被迅速发现和阻断。

3.3 调度器实现：从顺序流到自愈机制

调度器是项目里最需要花心思的部分。我只实现了顺序流和有向无环图两种模式，顺序流已经够用了。调度器持有一个 context 字典，按步骤列表逐个执行。

# orchestrator.py from typing import Dict, Any, List class Orchestrator: def __init__(self, agents: Dict[str, Agent]): self.agents = agents self.context: Dict[str, Any] = {} def add_context(self, key: str, value: Any) -> None: self.context[key] = value def run_step(self, step: Dict[str, str]) -> Dict[str, Any]: agent = self.agents[step["agent"]] task = step["task"] input_context = { key: self.context[key] for key in step.get("input_keys", []) } return agent.run(task, input_context) def run(self, steps: List[Dict[str, str]], initial_context: Dict[str, Any] = None) -> Dict[str, Any]: if initial_context: self.context.update(initial_context) for i, step in enumerate(steps): output_key = step.get("output_key", f"step_{i}") result = self.run_step(step) self.context[output_key] = result print(f"[orchestrator] step {i} done, agent={step['agent']}") return self.context

这个简化版能跑，但真实场景里还要加三样东西。第一是超时控制，给每个步骤设一个最大等待时间，大模型调用超过60秒就按失败处理。第二是重试，失败时根据错误类型判断是重试还是短路，网络抖动可以重试，输出校验失败也建议重试一次，但系统提示词写错导致的原则性问题重试也没用。第三是上下文序列化，每完成一步就把 context 写到本地JSON文件，崩溃之后能恢复现场。

我强烈建议你在调度器里加一个 max_steps 限制，防止有人误配了带环的步骤定义，任务在Agent之间无限循环，账单直接爆炸。我设的是总步数不超过20步。

在自愈机制上，我做过一个小demo：如果校对员返回了问题列表，调度器不会直接把任务标成失败，而是追加一个“修订步骤”，把问题列表合并给作者重新生成一版。这个追加逻辑的触发条件和数据传递，同样写成分离的步骤定义，而不是在代码里硬编码。这样整个流程依然可观察、可审计。

3.4 跑通一个文本生成链路

现在把三个人物实例化出来：检索员、作者、校对员。用最简单的方式串起一条链路。

# main.py researcher = Agent( name="researcher", role="检索汇总最新资料", system_prompt=( "你是检索员。当且仅当需要查找最新资料时，调用search_web工具。" "你必须输出JSON：{\"results\": [{\"title\": str, \"url\": str, \"summary\": str}], \"status\": str}。" "如果没有有效资料，status输出empty，不要编造。" ), tools=["search_web"], result_schema=SearchResultsSchema, ) writer = Agent( name="writer", role="基于资料撰写简报", system_prompt=( "你是作者。根据上游提供的results写一份600字以内的中文简报。" "输出JSON：{\"content\": str, \"sources\": [str]}。" ), result_schema=WriterOutputSchema, ) editor = Agent( name="editor", role="校对简报事实和格式", system_prompt=( "你是校对员。检查content中每个核心论断是否能在sources中找到对应支撑。" "输出JSON：{\"pass\": bool, \"issues\": [str]}。" ), result_schema=EditorOutputSchema, ) orchestrator = Orchestrator(agents={ "researcher": researcher, "writer": writer, "editor": editor, }) steps = [ {"agent": "researcher", "task": "新兴市场智慧零售落地案例", "output_key": "research"}, {"agent": "writer", "task": "基于research撰写简报", "input_keys": ["research"], "output_key": "draft"}, {"agent": "editor", "task": "校对draft", "input_keys": ["draft"], "output_key": "review"}, ] final_context = orchestrator.run(steps, initial_context={"topic": "智慧零售"}) print(final_context["review"])

执行完，你会在日志里看到三步依次完成。上下文对象里最后会有 research、draft、review 三个结果。如果校对员返回 pass=false，调度器可以追加一个步骤，把 issues 合并给作者再改一版，这个逻辑我一般写在run外面作为自愈机制。

这里要提醒一下，实际项目里不要把三个步骤都放在同一个进程里。我后面把它们拆成了三个独立服务，用消息队列传递结果，这样任一环节崩溃，不会拖垮整个流水线。不过在小规模跑通阶段，先用同一个进程把逻辑跑顺，比一上来就搞分布式实用得多。

如果你想把这条链路延伸成生产级服务，需要补的东西也就清晰了：消息队列、任务队列、数据库存储、监控面板、调用链追踪。每一步都不难，难的是在链路变长之后还能快速定位问题，所以我在很早的时候就把结构化日志和message_id设计进去了。

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

4.1 多个智能体来回踢皮球

这是我第一次跑通多智能体时最常遇到的问题。两个智能体互相要求对方补充信息，来回十几轮，任务就是结束不了。后来定位到原因，是系统提示词里给智能体留了“重新表述问题”的选项，它俩就顺着这个口子无休止地确认。

解决方法是两条：一是明确禁止智能体反向提问，每个智能体只接受调度器分发的任务，任务要求模糊就直接输出 insufficient_info 状态，不许把问题抛回去；二是在调度器层面限制每个步骤的执行次数，超过就强制失败。

这个现象在自主协作架构里尤其常见，因为消息可以直接互相触达，没有中间人做裁决。哪怕你用的是集中调度，也要小心步骤定义里意外出现环。我每次修改步骤列表后都会跑一个简单的环检测脚本，确保执行计划是无环的。

4.2 上下文窗口溢出

多智能体协作最容易把整个上下文传给每个步骤。我的分析师初版 prompt 里直接序列化了全部历史输出，导致后几个智能体的输入token量远超标。这个问题的根本原因是对“哪些上下文是当前步骤真正需要的”没有一个清晰的判断。

我的做法是给每个步骤声明 input_keys，只把需要的字段传给智能体。比如校对员只需要 draft 里的 content 和 sources，根本不需要 research。如果你发现输入token还是很大，可以进一步让前一个智能体输出“摘要版”，而不是把原始全文传给下一位。用摘要传递信息，整个链条的token消耗是线性增长的，而不是指数爆炸。

这里还有个容易被低估的点：即使输入token没有超过模型限制，过大的上下文也会明显降低输出质量。模型会优先注意开头结尾的信息，中间的历史数据常常被忽略。所以按需传上下文，不只是为了省钱，更是为了保证下游智能体的推理质量。

4.3 输出格式解析报错

模型虽然被要求输出JSON，但还是经常出错。最常见的是多行字符串没转义，以及模型自作聪明在JSON外面加了markdown代码块标记。我在 validate_result 里做了两件防御：

text = text.strip() if text.startswith("```"): first_line = text.find("\n") text = text[first_line:].strip() text = text.rsplit("```", 1)[0].strip()

然后才是json.loads。另外，如果输出的JSON里字段类型不对，pydantic的 ValidationError 会明确告诉我哪个字段出问题，方便快速调整提示词示例。

这个问题的深层原因，是模型对“输出纯JSON”的理解不稳定。只靠提示词约束不够，还要在上游智能体发出结果后立刻校验一次。凡是校验不通过的，就触发一次重试，并且把上次错误信息写回提示词，让模型知道哪里出问题了。重试两次还不行，再判断是提示词写错还是任务本身有问题。

4.4 日志与链路追踪技巧

在这类项目里，日志不是可选项，是保命项。我的每个步骤都打印一个结构化日志行，包含 message_id、agent、step、status、耗时、token用量、错误摘要。查询的时候用 grep 过滤 message_id，就能还原整条链路。

我把常用的排查问题整理成了一个表，方便大家对照：

还有一点是提示词版本管理。我每次修改某个智能体的 system_prompt，都会把这个prompt连同当时的测试用例一起存到一个prompts目录里。智能体优化本质上是prompt迭代，没有版本管理，你就不知道哪次改动导致效果回退，而且这种回退在系统里很难通过代码测试发现。

如果你用的是Git仓库，建议把提示词也纳入版本管理。每次改动提示词，都单独提一个分支，跑一遍回归用例再合入。这比直接在生产环境里改提示词要稳得多。我后来能快速定位各种问题，很大程度就是靠这层提示词版本库。

最后说点个人体会。跑这个项目给我的最大收获不是“多智能体有多强”，而是“限制会让系统更可靠”。你每允许智能体自由发挥一步，就要准备一个兜底策略；你每着急删掉一条校验规则，后面就会用一次现场事故补回来。如果让我重新设计，我还是会先把调度逻辑、输出Schema和日志体系做扎实，再谈花哨的自主协作。这也算是我踩了不少坑之后的一条实用心得吧。