企业官网建设流程全解析

1. 从单兵作战到团队协同：为什么我们需要重新思考智能体架构

做过智能体项目的人大概都有这种体会：一开始写个单Agent的Demo，感觉无所不能，接上大模型、挂几个工具、写个提示词模板，就能跑通一个像模像样的任务。但一旦把场景放大到真实业务里——比如需要同时处理需求分析、代码生成、测试验证、文档撰写这几条线——单Agent立刻捉襟见肘。上下文窗口不够用、角色切换混乱、任务依赖理不清，最后要么是提示词越写越长变成一锅粥，要么是拆成多个独立脚本但彼此之间完全没法协作。

Agency Agents这套多智能体协作系统，就是冲着这个痛点来的。它的核心思路很直接：既然一个Agent搞不定复杂任务，那就组一个“开发团队”，让不同角色的Agent各司其职，通过一套明确的协作协议把任务串起来。你可以把它理解成一个虚拟的软件工作室——里面有负责拆解需求的项目经理、有写代码的工程师、有做代码审查的质检员、还有负责整理交付物的文档专员。每个角色都是一个独立的智能体实例，拥有自己的系统提示词、工具集和记忆空间，它们之间通过消息传递和任务队列来协同工作。

这套系统适合谁看？如果你正在做以下任何一件事，这篇内容应该能给你一些可以直接抄作业的思路：一是手头有复杂任务需要多个大模型调用串联，但用单Agent的链式调用已经维护不动了；二是想了解多智能体系统的架构设计，特别是任务分配、状态管理和冲突解决这几个硬骨头怎么啃；三是已经在用某些智能体框架，但觉得角色划分太粗或者协作机制不够灵活，想自己搭一套更贴合业务的东西。

我接下来会从整体架构设计讲到每个核心模块的实现细节，再到实际落地时踩过的坑和排查技巧。内容会尽量贴着实操走，参数怎么定、消息格式怎么设计、状态怎么同步，这些都会给到具体的方案和代码片段。你不需要有很深的分布式系统背景，但最好对Python异步编程和大模型API调用有基本了解，这样读起来会更顺畅。

2. 整体架构设计：角色分工、通信机制与状态管理

2.1 为什么是“Agency”而不是“Pipeline”

很多人第一反应会把这个系统理解成一条流水线：需求进来，经过分析Agent、编码Agent、测试Agent，最后输出结果。Pipeline模式确实简单，但它在真实场景里有个致命问题——任务不是线性的。编码过程中发现需求有歧义，需要回头找分析Agent确认；测试不通过，要把缺陷单打回给编码Agent；文档Agent可能需要等所有代码冻结之后才能动笔。这些反馈回路和依赖关系，用Pipeline根本表达不了。

Agency这个词的核心在于“代理关系”和“自主决策”。每个Agent不是被动等待上游输入，而是主动从共享的任务池里领取自己能处理的任务，处理完之后把结果和后续任务写回池子。这种设计借鉴了分布式任务队列的思路，但加上了角色权限和领域知识的约束。打个比方，Pipeline像工厂流水线，Agency像一群自由职业者在一个共享工作台上协作——谁有空谁接活，但每个人只接自己擅长的活。

具体到架构上，整个系统分成四层：

• 角色定义层：每个Agent的角色描述、能力边界、可用工具、系统提示词模板。这一层决定了“你是谁、你能干什么”。
• 通信层：基于消息队列的异步通信机制，支持点对点消息和广播消息。消息格式统一为结构化JSON，包含发送者、接收者、消息类型、负载和上下文引用。
• 任务调度层：负责任务的创建、分配、状态流转和依赖管理。核心是一个带优先级和依赖检查的任务队列。
• 状态存储层：保存每个Agent的私有记忆、共享的全局状态、以及任务执行的历史记录。这一层保证了Agent重启后能恢复上下文。

2.2 角色划分的粒度怎么定

角色划分太粗，等于没分；太细，通信开销爆炸。我试过几种粒度，最后稳定下来的方案是“按交付物划分”。什么意思？每个Agent负责一种明确的交付物类型。比如：

这个划分的好处是每个角色的输出都是可验证的——需求文档可以评审，代码可以跑测试，审查报告可以逐条核对。而且角色之间的依赖关系天然清晰：编码依赖架构设计，测试依赖编码，文档依赖所有前置产物。

注意：不要按“技能”划分角色，比如“Python专家”“前端专家”。这种划分在任务分配时会产生大量歧义——一个任务到底该给谁？按交付物划分就没这个问题，因为交付物类型是唯一的。

2.3 通信机制：消息队列还是直接调用

早期版本我用的是Agent之间直接函数调用，A调用B的方法，B返回结果。简单直接，但很快遇到三个问题：一是同步阻塞，A等B的时候什么都干不了；二是错误传播，B抛异常直接把A也带崩；三是无法追踪，调用链一长根本不知道哪一步出了问题。

后来换成基于消息队列的异步通信，核心设计如下：

# 消息结构定义 class AgentMessage: def __init__(self, sender, receiver, msg_type, payload, context_ref=None): self.msg_id = generate_uuid() self.sender = sender # 发送者角色ID self.receiver = receiver # 接收者角色ID，None表示广播 self.msg_type = msg_type # task_assign / task_result / query / response self.payload = payload # 具体内容，结构化字典 self.context_ref = context_ref # 关联的任务ID或会话ID self.timestamp = now()

消息队列用Redis的Stream或者RabbitMQ都行，我选Redis Stream的原因是部署简单、支持消费者组、消息持久化够用。每个Agent启动时注册自己的消费者组，只消费发给自己的消息和广播消息。

这里有个关键设计：消息不直接触发Agent执行，而是先进入Agent的待处理队列，由Agent的主循环决定什么时候处理。这样做的好处是Agent可以根据自己的当前负载和任务优先级来调度，而不是被消息推着走。比如编码Agent正在处理一个复杂函数生成，这时候收到一条低优先级的文档更新通知，它可以先处理完手头的再响应。

2.4 状态管理：私有记忆与共享黑板

多智能体系统最头疼的就是状态一致性。A改了需求文档，B还在用旧版本编码，最后交付物对不上。我的方案是“私有记忆+共享黑板”的混合模式。

私有记忆是每个Agent自己的上下文窗口和长期存储，保存它处理过的任务、生成的中间结果、以及从共享黑板读取的快照。私有记忆用向量数据库加结构化存储，向量库存语义相关的历史片段，结构化存储存任务ID和状态。

共享黑板是一个全局的键值存储，所有Agent都可以读写。黑板上放的是“事实性”信息：当前需求版本号、接口定义、已完成的模块列表、已知缺陷列表。任何Agent修改黑板上的内容，都要发一条广播消息通知其他Agent。其他Agent收到通知后，可以选择刷新自己的本地快照。

# 共享黑板操作示例 class SharedBlackboard: def __init__(self, redis_client): self.redis = redis_client def update(self, key, value, version, agent_id): # 乐观锁：检查版本号 current = self.redis.hget("blackboard", key) if current and json.loads(current)["version"] >= version: raise VersionConflict("黑板内容已被其他Agent更新") new_value = {"value": value, "version": version + 1, "updated_by": agent_id} self.redis.hset("blackboard", key, json.dumps(new_value)) # 广播变更通知 self.publish_change_notification(key, version + 1, agent_id)

版本冲突的处理策略是“后写者重试”。如果编码Agent想更新接口定义但发现版本号落后了，它需要先读取最新版本，合并自己的修改，再尝试写入。如果合并有冲突（比如两个Agent改了同一个接口的不同部分），就升级为人工介入或者由架构设计师Agent来仲裁。

3. 核心模块实现：任务调度、角色引擎与协作协议

3.1 任务调度器：带依赖检查的优先级队列

任务调度器是整个系统的心脏。它要解决三个问题：任务怎么创建、任务怎么分配、任务之间的依赖怎么处理。

任务创建有两种方式：一是外部触发，比如用户提交一个需求，系统自动生成一个“需求分析”任务；二是Agent在处理任务过程中动态生成子任务，比如编码Agent发现需要先定义接口，就创建一个“接口设计”任务并设置依赖关系。

任务的数据结构如下：

class Task: def __init__(self, task_id, task_type, payload, priority=5, depends_on=None): self.task_id = task_id self.task_type = task_type # 对应角色类型 self.payload = payload # 任务输入 self.priority = priority # 1-10，数字越小优先级越高 self.depends_on = depends_on or [] # 依赖的任务ID列表 self.status = "pending" # pending / assigned / running / done / failed self.assigned_to = None self.created_at = now() self.updated_at = now()

调度循环的逻辑是：每隔一个心跳周期（我设的是2秒），扫描所有pending状态的任务，检查依赖是否全部完成。如果依赖满足，就根据任务类型找到对应的空闲Agent，分配任务并更新状态为assigned。Agent处理完后把状态改为done或failed，并触发下一轮调度。

优先级的设计有个细节：不要用绝对优先级，用相对优先级加老化机制。什么意思？如果高优先级任务一直插队，低优先级任务可能永远得不到执行。我的做法是每个任务有一个基础优先级，然后每等待一个调度周期优先级提升一级。这样即使是最低优先级的任务，等待足够长时间后也能被处理。

实操心得：依赖检查不要用递归查询，性能会很差。我的做法是维护一个反向依赖表——每个任务记录“我完成后会解锁哪些任务”。任务完成时直接查这个表，把解锁的任务标记为可调度。这样依赖检查是O(1)的。

3.2 角色引擎：提示词模板与工具绑定

每个Agent的角色引擎由三部分组成：系统提示词、工具集、执行循环。

系统提示词不是随便写一段“你是一个程序员”就完事了。我总结了一个模板结构，包含五个部分：

1. 角色身份：你是谁，你的专业背景是什么。
2. 职责边界：你负责什么，不负责什么。这一条特别重要，防止Agent越界干别人的活。
3. 输入输出规范：你接收什么格式的输入，必须输出什么格式的结果。
4. 协作规则：什么时候需要向其他Agent发消息，什么时候需要更新共享黑板。
5. 质量准则：你的输出需要满足哪些质量标准。

举个例子，编码工程师的系统提示词片段：

你是开发团队中的编码工程师，专注于将技术方案转化为可运行的代码。 你的职责是编写功能代码和单元测试，不负责架构决策和需求变更。 输入：架构设计师提供的接口定义和模块划分方案。 输出：符合PEP8规范的Python代码文件，包含docstring和类型注解。 协作规则：完成编码后，向代码审查员发送审查请求；如果发现接口定义有歧义，向架构设计师发起查询。 质量准则：代码必须通过静态检查，单元测试覆盖率不低于80%。

工具绑定方面，每个角色只挂载自己需要的工具。编码工程师挂载代码生成、文件读写、静态检查工具；测试工程师挂载测试框架、覆盖率分析工具；需求分析师挂载文本解析和知识库检索工具。这样做的好处是减少工具描述对上下文窗口的占用，同时降低Agent误用工具的概率。

执行循环是一个标准的“感知-决策-行动”循环：

async def run_loop(self): while self.running: # 感知：检查消息队列和任务队列 messages = await self.fetch_messages() tasks = await self.fetch_assigned_tasks() # 决策：根据当前状态决定下一步 if tasks: task = self.select_task(tasks) plan = await self.plan(task) elif messages: msg = self.select_message(messages) plan = await self.plan_for_message(msg) else: await asyncio.sleep(self.idle_interval) continue # 行动：执行计划 result = await self.execute(plan) # 反馈：更新状态和发送消息 await self.update_task_status(task, result) await self.notify_dependents(task)

3.3 协作协议：请求-响应与发布-订阅的混合模式

Agent之间的协作分两种场景：一种是需要对方回复的，比如编码Agent问架构师“这个接口的参数类型是什么”；另一种是不需要回复的，比如测试Agent广播“发现一个严重缺陷”。

请求-响应模式用点对点消息实现。发送方创建一个请求消息，附带一个回调地址（通常是发送方的消息队列），接收方处理完后把结果发到回调地址。这里有个超时机制：如果发送方在指定时间内没收到响应，就重试或者降级处理。超时时间我设的是30秒，重试次数2次。

发布-订阅模式用广播消息实现。任何Agent都可以往一个主题（topic）发消息，订阅了该主题的Agent都会收到。主题按领域划分，比如“需求变更”“接口更新”“缺陷通知”。订阅关系在Agent启动时注册，运行时可动态调整。

# 协作协议示例：编码Agent请求架构师澄清接口 async def request_clarification(self, architect_id, question): request_msg = AgentMessage( sender=self.agent_id, receiver=architect_id, msg_type="query", payload={"question": question, "context": self.current_task_id}, context_ref=self.current_task_id ) await self.message_bus.send(request_msg) # 等待响应，带超时 response = await self.message_bus.wait_for_response( request_msg.msg_id, timeout=30 ) if response is None: # 超时降级：基于现有信息做合理假设，并记录假设 self.log_assumption(question, "超时未响应，基于常见实践假设") return self.make_reasonable_assumption(question) return response.payload["answer"]

注意：请求-响应模式要避免循环等待。A等B，B等C，C又在等A，整个系统就死锁了。我的做法是给每个请求消息设置一个“跳数”计数器，每经过一个Agent跳数加一，超过阈值（我设的是5跳）就强制返回当前最佳结果，不再继续等待。

4. 落地实操：从零搭建一个多智能体开发团队

4.1 环境准备与依赖安装

先说一下基础环境。我用的Python 3.10以上，主要是需要asyncio的完整支持。Redis 7.0以上，用于消息队列和共享黑板。向量数据库我选的Chroma，轻量够用，如果数据量大可以换Milvus或者Qdrant。

依赖清单如下：

pip install redis>=4.5.0 pip install chromadb>=0.4.0 pip install openai>=1.0.0 # 或其他大模型SDK pip install pydantic>=2.0 # 用于消息和任务的数据校验 pip install structlog # 结构化日志，排查问题必备

Redis的配置需要注意两点：一是开启Stream的持久化，防止重启丢消息；二是设置合理的内存淘汰策略，我用的allkeys-lru，因为消息和黑板数据都是可重建的。

# redis.conf 关键配置 appendonly yes appendfsync everysec maxmemory 2gb maxmemory-policy allkeys-lru

4.2 定义你的第一个Agent角色

假设我们要搭建一个最小可用的开发团队，先从需求分析师和编码工程师两个角色开始。

需求分析师的配置文件：

# roles/analyst.py ANALYST_ROLE = { "role_id": "analyst", "name": "需求分析师", "system_prompt": """你是开发团队中的需求分析师。 你的职责是将用户需求拆解为结构化的用户故事和验收标准。 你不负责技术方案设计和代码实现。 输入：自然语言描述的需求。 输出：JSON格式的用户故事列表，每个故事包含id、描述、验收标准、优先级。 协作规则：完成分析后，创建架构设计任务并分配给架构设计师。 质量准则：每个用户故事必须可独立测试，验收标准必须具体可验证。""", "tools": ["text_parser", "knowledge_base_search"], "max_context_tokens": 8000, "temperature": 0.3 }

编码工程师的配置文件：

# roles/coder.py CODER_ROLE = { "role_id": "coder", "name": "编码工程师", "system_prompt": """你是开发团队中的编码工程师。 你的职责是根据技术方案编写功能代码和单元测试。 你不负责需求变更和架构决策。 输入：架构设计师提供的接口定义和模块划分。 输出：Python代码文件，包含docstring、类型注解和单元测试。 协作规则：完成后向代码审查员发送审查请求；发现接口歧义时向架构设计师查询。 质量准则：代码通过pylint检查，单元测试覆盖率不低于80%。""", "tools": ["code_generator", "file_io", "static_checker"], "max_context_tokens": 16000, "temperature": 0.2 }

温度参数的选择有讲究。需求分析和编码这种需要精确输出的角色，温度设低一点（0.2-0.3），减少随机性。如果是创意类角色比如文档专员，可以设到0.5-0.7，让输出更自然。

4.3 启动调度器与Agent实例

调度器和Agent是分开的进程，通过Redis通信。这样做的好处是Agent可以独立重启和扩容，调度器不会成为单点。

调度器启动代码：

# scheduler.py import asyncio from task_queue import TaskQueue from blackboard import SharedBlackboard async def main(): redis_client = await create_redis_client() task_queue = TaskQueue(redis_client) blackboard = SharedBlackboard(redis_client) # 注册角色与任务类型的映射 role_task_mapping = { "analyst": ["requirement_analysis"], "architect": ["architecture_design"], "coder": ["code_implementation"], "reviewer": ["code_review"], "tester": ["test_design", "test_execution"], "writer": ["doc_writing"] } scheduler = Scheduler(task_queue, blackboard, role_task_mapping) await scheduler.run() if __name__ == "__main__": asyncio.run(main())

Agent启动代码：

# agent_runner.py import asyncio from agent import Agent from roles.coder import CODER_ROLE async def main(): agent = Agent( role_config=CODER_ROLE, redis_url="redis://localhost:6379", llm_config={"model": "gpt-4", "api_key": "your-key"} ) await agent.start() await agent.run_loop() if __name__ == "__main__": asyncio.run(main())

启动顺序是：先起Redis，再起调度器，最后起各个Agent。Agent启动后会向调度器注册自己的角色ID和消费者组，调度器收到注册后开始给它分配任务。

4.4 一个完整任务的流转实录

假设用户提交了一个需求：“做一个待办事项管理工具，支持增删改查和优先级排序。”

第一步：需求分析。调度器创建requirement_analysis任务，分配给analyst。analyst调用大模型拆解需求，输出如下：

{ "user_stories": [ {"id": "US-001", "desc": "用户可以创建待办事项", "criteria": "输入标题和优先级，保存后出现在列表中", "priority": 1}, {"id": "US-002", "desc": "用户可以删除待办事项", "criteria": "点击删除按钮，事项从列表移除", "priority": 2}, {"id": "US-003", "desc": "用户可以修改待办事项", "criteria": "点击编辑，修改后保存生效", "priority": 2}, {"id": "US-004", "desc": "用户可以按优先级排序", "criteria": "列表按优先级从高到低排列", "priority": 3} ] }

analyst把结果写入共享黑板，并创建architecture_design任务，依赖requirement_analysis完成。

第二步：架构设计。architect读取黑板上的用户故事，设计数据模型和接口：

{ "data_model": { "TodoItem": { "id": "string", "title": "string", "priority": "int", "created_at": "datetime" } }, "interfaces": [ {"name": "create_todo", "input": "title, priority", "output": "TodoItem"}, {"name": "delete_todo", "input": "id", "output": "bool"}, {"name": "update_todo", "input": "id, title, priority", "output": "TodoItem"}, {"name": "list_todos", "input": "sort_by", "output": "List[TodoItem]"} ] }

architect创建code_implementation任务，分配给coder。

第三步：编码实现。coder根据接口定义生成代码：

# todo_service.py from dataclasses import dataclass from datetime import datetime from typing import List, Optional @dataclass class TodoItem: id: str title: str priority: int created_at: datetime class TodoService: def __init__(self): self._items: dict[str, TodoItem] = {} def create_todo(self, title: str, priority: int) -> TodoItem: item_id = generate_id() item = TodoItem(id=item_id, title=title, priority=priority, created_at=datetime.now()) self._items[item_id] = item return item def delete_todo(self, item_id: str) -> bool: return self._items.pop(item_id, None) is not None def update_todo(self, item_id: str, title: str = None, priority: int = None) -> Optional[TodoItem]: item = self._items.get(item_id) if not item: return None if title is not None: item.title = title if priority is not None: item.priority = priority return item def list_todos(self, sort_by: str = "priority") -> List[TodoItem]: items = list(self._items.values()) if sort_by == "priority": items.sort(key=lambda x: x.priority) return items

coder同时生成单元测试，然后创建code_review任务。

第四步：代码审查。reviewer检查代码规范、逻辑漏洞、边界条件。发现update_todo方法没有处理空字符串标题的情况，生成审查意见并打回给coder。coder修改后重新提交，reviewer通过。

第五步：测试执行。tester设计测试用例并执行，覆盖增删改查和排序场景。全部通过后，创建doc_writing任务。

第六步：文档撰写。writer根据代码和测试报告生成用户文档和API文档。

整个流程走下来，从需求到交付物，大概需要15-20次大模型调用，耗时取决于模型响应速度。我实测下来，用中等规模的模型（比如GPT-3.5级别），整个流程在3-5分钟内完成。

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

5.1 Agent卡死或无限循环

这是最常见的问题。表现是某个Agent一直在处理同一个任务，或者两个Agent互相发消息停不下来。

排查思路分三步：先看日志，确认是哪个Agent在循环；再看消息队列，确认是不是有消息被重复消费；最后看任务状态，确认是不是依赖检查出了问题。

我遇到过的几个具体原因：

• 消息重复消费：Redis Stream的消费者组在Agent重启后没有正确确认消息，导致消息被重新投递。解决方法是Agent启动时先检查未确认的消息，处理完再确认。
• 依赖循环：A任务依赖B，B任务依赖A。调度器检测到循环依赖后应该直接标记为失败，而不是无限等待。我在调度器里加了一个依赖图检测，发现环就报警。
• 提示词导致的行为循环：Agent的系统提示词里写了“如果不确定就询问架构师”，结果架构师也不确定，又反问回来。解决方法是在提示词里加一条“如果对方也无法确定，基于常见实践做合理假设并记录”。

实操心得：给每个Agent加一个“最大连续执行次数”限制，比如连续处理20个任务后强制进入冷却期，等待人工检查。这个阈值根据任务复杂度调整，我一般设15-25。

5.2 共享黑板数据不一致

表现是不同Agent读到的黑板数据版本不同，导致基于旧数据做决策。

根本原因是读写没有原子性。我的解决方案是引入版本号和乐观锁，前面已经提过。但还有一个隐蔽的问题：Agent的本地缓存没有及时失效。Agent为了提高性能，会把黑板数据缓存在本地内存里，但广播通知可能因为网络延迟晚到。

解决方法是给缓存设置一个较短的TTL（我设的是5秒），同时收到广播通知时立即失效对应缓存。另外，关键决策前强制从黑板重新读取最新版本，不依赖缓存。

5.3 大模型输出格式不符合预期

Agent之间的通信依赖结构化数据，但大模型有时候会输出多余的解释文字或者格式错误的JSON。

我的处理策略是三层防护：第一层，在提示词里明确要求“只输出JSON，不要任何额外文字”；第二层，用Pydantic做输出校验，格式不对直接抛异常；第三层，异常时触发重试，重试时在提示词里加上“上次输出格式错误，请严格按照JSON格式输出”。

如果重试三次还是失败，就降级为人工处理，把原始输出和错误信息记录到日志里。

from pydantic import BaseModel, ValidationError class UserStory(BaseModel): id: str desc: str criteria: str priority: int def parse_llm_output(raw_output: str) -> list[UserStory]: # 尝试提取JSON部分 json_str = extract_json(raw_output) try: data = json.loads(json_str) return [UserStory(**item) for item in data["user_stories"]] except (json.JSONDecodeError, ValidationError) as e: raise OutputFormatError(f"输出格式错误: {e}, 原始输出: {raw_output[:200]}")

5.4 任务分配不均

有的Agent忙死，有的Agent闲死。原因是任务类型分布不均匀，比如编码任务特别多，但审查任务很少。

解决方案有两个：一是动态调整Agent数量，给繁忙的角色启动多个实例，共享同一个消费者组；二是任务窃取，空闲的Agent可以“借调”去处理其他类型的任务，但需要临时加载对应的角色提示词和工具集。

我一般先用方案一，因为实现简单。方案二适合任务类型差异大但总量不大的场景。

5.5 性能优化：减少不必要的模型调用

多智能体系统跑起来之后，大模型调用费用是主要成本。我做过统计，一个中等复杂度的开发任务，如果不做优化，大概需要30-40次模型调用。优化后可以降到15-20次。

优化手段包括：

• 缓存常见查询：比如“Python中如何读取JSON文件”这种问题，第一次调用后把结果缓存起来，后续直接返回。
• 批量处理：把多个小任务合并成一个提示词，一次调用完成。比如代码审查时，把规范检查、逻辑检查、边界检查合并成一个提示词。
• 规则引擎前置：能用规则判断的不要用模型。比如代码格式检查，直接用pylint跑一遍，只把pylint无法判断的问题交给模型。
• 模型分级：简单任务用便宜的小模型，复杂任务用大模型。比如需求拆解用大模型，格式转换用小模型。

注意：缓存要设置合理的过期时间，并且要区分上下文。同一个问题在不同项目背景下答案可能不同，缓存键要包含项目ID和上下文摘要。

6. 扩展方向与个人实践体会

这套系统跑通之后，我陆续做了一些扩展，有些效果不错，有些踩了坑。

第一个扩展是引入人类审批节点。在关键决策点（比如架构设计完成、代码审查通过）插入人工确认步骤。实现方式是在任务状态里加一个awaiting_human状态，调度器遇到这个状态就暂停后续任务，等人工在管理界面点击确认后再继续。这个功能在真实项目里非常必要，因为大模型有时候会做出看似合理但实际有问题的决策。

第二个扩展是跨项目知识复用。把历史项目的需求文档、架构方案、代码片段存入向量数据库，新项目启动时先检索相似项目作为参考。这个功能对提升输出质量帮助很大，但要注意隐私隔离——不同项目的知识不能混在一起。

第三个扩展是Agent自我评估。每个Agent完成任务后，对自己的输出做一个置信度评分。低置信度的任务自动触发人工复核。置信度评分用一个小模型来做，成本很低但效果不错。

踩过的坑也有几个。一个是过度设计，一开始想给每个Agent加复杂的记忆机制和反思循环，结果系统变得极其难调试。后来砍掉了大部分花哨功能，只保留最核心的任务队列和消息通信，反而稳定了很多。另一个是忽视日志，早期没有结构化日志，出问题只能靠猜。后来用structlog把每个Agent的每个决策都记录下来，排查效率提升了一个数量级。

我个人在实际操作中的体会是，多智能体系统的核心难点不在模型能力，而在工程实现。任务怎么拆、消息怎么传、状态怎么同步、错误怎么恢复，这些才是决定系统能不能跑起来的关键。模型选型反而没那么重要，中等能力的模型配合好的工程架构，效果远超大模型配烂架构。

最后分享一个小技巧：先用单Agent跑通全流程，再拆成多Agent。很多人一上来就设计复杂的多Agent协作，结果连基本流程都没跑通。我的做法是先写一个“全能Agent”，把所有角色的提示词和工具都塞进去，跑通一个完整任务。然后观察哪些环节是瓶颈，再把瓶颈环节拆成独立Agent。这样拆出来的多Agent系统，每个角色的边界都是经过验证的，不会出现“为了拆而拆”的情况。