ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

五角色Agent团队架构:解决长链路任务失控的协作模式

五角色Agent团队架构:解决长链路任务失控的协作模式 前一阵和几个做 AI 应用的朋友聊天发现大家撞到了同一个墙单个 Agent 在短任务上表现惊艳但一旦让它独立完成“需求分析 → 方案设计 → 代码实现 → 质量检查 → 部署交付”这条完整链路结果就开始失控。要么是做着做着忘了最初的约束要么是中间步骤出了错后面所有环节都跟着错你甚至说不清问题出在哪一步。这就是“一人公司”式 Agent 团队出现的真实背景。Hermes Agent Team 提出的“五角色架构”本质上是把一家微型公司的协作方式压缩进 Agent 系统不再追求一个全能的超级 Agent而是让五个职责边界清晰的角色沿着一条可控的流程协作。v3.1 版本的核心价值也不在于多接入了几个模型而在于这套协作机制的稳定性、可复用性和可观测性有了实质收敛。这篇文章我从架构视角把五角色模型拆开讲透顺便给出一套可以最小化跑通的参考实现。读完你会理解五角色架构到底在解决什么问题每个角色的职责边界画在哪里v3.1 这类版本迭代到底在演进什么以及把这种架构接到真实项目中最容易踩到哪些坑。1. 这篇文章真正要解决的问题先说判断五角色架构不是为了“看起来高级”它解决的是长链路任务里最让人头疼的三个问题——上下文漂移、责任不明、过程不可观测。如果你只是让一个 Agent 写一段独立代码那根本不需要五角色架构一个精心设计的提示词就够。但如果你要做的是以下几类事情单 Agent 模式就会明显吃力第一任务链路长。比如“基于一份产品需求输出技术方案再生成完整代码再做 Code Review最后给出部署建议”。每一步的输出都会影响下一步而单 Agent 在长上下文中经常会“忘记”前期的约束。第二需要质量闸门。一个人写代码也会出错所以需要测试和评审。让同一个 Agent 自己写自己审等于让作者自己当编辑效果十分有限。第三需要过程留痕。单 Agent 模式下你只能看到最终结果。一旦结果不对你没有中间产物可以定位问题。而五角色架构中每个角色的产出都是一个独立节点哪一步出了问题看节点状态就行。那什么样的人最适合读这篇文章如果你正在做 Agent 应用、自动化工作流、企业内部知识助手或者你想用 Agent 体系支撑一个人完成多环节产出这篇文章适合你。如果你只是想让 ChatGPT 帮你写个 SQL那可以收藏本文以后再做复杂任务时再回来参考。还有一点要先说清楚五角色架构是模式不是框架。Hermes Agent Team 是实现这一模式的具体项目之一v3.1 是该项目的版本标识。本文会把重点放在模式本身的可落地方法上具体 API 以你使用的项目文档为准。2. 五角色架构的核心概念与职责边界2.1 为什么是“五”个角色一家真实的小公司哪怕只有一个人在做事时也要在心里扮演多个角色先当产品经理想清楚需求再当开发把功能写出来再当测试把质量关守住最后还要当运营把东西交付出去。一个人做不了这些事同时进行但可以分阶段切换身份。五角色架构就是把这种“分阶段切换身份”固化成系统结构。从公开材料和设计逻辑看五个角色通常对应这样一条链路协调者负责接收任务和整体调度规划者负责把任务拆解成可执行方案开发者负责具体执行产出评审者负责检查产出质量运营者负责最终交付和输出格式化。这五个角色形成了一个完整的闭环协调者发起规划者细化开发者执行评审者把关运营者交付。每个角色只关心自己这一段不需要知道全链条的所有细节。2.2 五个角色的职责边界角色核心职责典型输入典型输出不应做的事协调者 Coordinator接收任务、拆解目标、调度流转用户原始需求任务分解清单、调度指令不直接写业务代码不做技术细节决策规划者 Planner把目标转化为可执行的详细计划任务分解清单技术方案、步骤计划、验收标准不直接产出最终交付物不做执行层面的细节修改开发者 Developer按计划执行产出具体成果技术方案与计划代码、文档、配置、数据结果不擅自变更需求范围不跳过评审直接交付评审者 Reviewer检查产出质量给出修改意见开发者成果评审报告、问题清单、修改建议不直接改代码不替开发者做决定运营者 Operator最终检查、格式整理、交付输出评审通过的成果最终交付内容、使用说明、部署建议不改变业务逻辑不做无依据的额外修改这里真正容易踩坑的地方是“不应做的事”这一列。很多人实现多 Agent 时每个 Agent 的 System Prompt 都写着“你可以做任何事”结果五个 Agent 互相抢活边界形同虚设。五角色架构的灵魂不是角色多而是每个角色有明确的“不做清单”。2.3 交接机制比角色本身更重要如果只看表面很容易误以为五角色架构就是定义五个 System Prompt然后依次调用五个 Agent。但实际项目中角色之间的“交接物”才是系统的关键。所谓交接物就是上一个角色传给下一个角色的结构化信息。比如协调者传给规划者的不能只是“帮我写个工具”而应该是一个包含任务 ID、目标描述、约束条件、已有上下文的任务对象。规划者传给开发者的应该是一份包含步骤、验收标准、优先级的技术方案。为什么交接物要结构化因为 Agent 之间传递的是文本而文本越长信息丢失越严重。把交接物设计成固定的数据结构相当于给每个角色发了一张标准工单谁拿到工单都知道自己该干什么、交付什么。这也是五角色架构和“五个 Prompt 串起来”的本质区别。3. v3.1 版本在演进什么从角色堆叠到流程收敛说到版本号很多人的第一反应是“又加了什么新功能”。但对于 Agent 团队这类系统版本迭代更值得关注的不是模型列表变长了多少而是协作机制变了多少。从架构设计角度看v3.1 这类版本通常在做三件事。第一件事角色协作从“硬编码顺序”走向“规则化流转”。早期实现往往是写死的链条先调 A再调 B再调 C。这种方式的缺点是一旦规划者认为任务不需要开发或者评审者发现问题需要退回流程就卡死了。更成熟的版本会引入“流转规则”允许任务在角色之间来回跳动比如评审者发现问题时任务会退回开发者并携带评审意见。第二件事上下文管理从“全部塞进去”走向“按需传递”。多角色协作最容易出现的问题就是上下文爆炸每个角色都要带全量对话历史成本高且噪音大。v3.1 这类演进方向是让每个角色只拿自己需要的那部分上下文。协调者和规划者看到的是目标和约束开发者看到的是详细方案评审者看到的是产出物和验收标准。第三件事状态可观测。一套五角色系统跑起来之后如果中间某个环节失败你能不能快速定位“卡在哪个角色手里”更成熟的版本会把每个角色的处理状态、交接时间、输入输出摘要记录下来这样整个流程就是可审计的。从材料看关于 v3.1 的具体功能细节目前并不完整。但有一点判断是可靠的这个版本迭代的重点不在模型侧而在流程侧。它试图回答的是“当任务变复杂、角色变多时如何让协作依然稳定”。这个方向比单纯堆角色数量有价值得多。4. 环境准备与前置条件接下来我们动手把五角色架构的最小示例跑通。这里先说明下面这套参考实现不绑定某个特定框架演示的是模式本身。你在接入 Hermes Agent Team 或自研实现时接口以实际项目文档为准。4.1 基础环境建议环境如下版本请以实际项目为准本文重点演示通用思路Python 3.10 及以上用于编写调度逻辑和角色定义。一个可用的 LLM API 接口支持 Chat Completions 形式的调用即可。pip 作为依赖管理工具。示例中我会用openai风格客户端做演示因为这是目前最常见的形式。如果你用的是其他模型服务只需要替换base_url、api_key和模型名。pip install openai pyyaml4.2 配置文件在项目根目录创建配置文件config.yaml用来管理模型接入和角色参数。把模型名、温度、最大轮数这类参数放到配置里而不是写死在代码中是后续维护的基本功。# 文件路径config.yaml llm: base_url: https://your-llm-endpoint.example.com/v1 api_key_env: LLM_API_KEY model: your-model-name temperature: 0.3 max_tokens: 2048 agent_team: version: 3.1 max_rounds: 5 roles: coordinator: enabled: true max_retries: 2 planner: enabled: true max_retries: 2 developer: enabled: true max_retries: 3 reviewer: enabled: true max_retries: 3 operator: enabled: true max_retries: 2配置里有两个关键点第一api_key_env表示 API Key 从环境变量读取而不是直接写在配置文件里。这样即使配置仓库被分享出去也不会泄露密钥。第二max_rounds是整条流程的最大协作轮数。为什么要设上限因为评审环节发现问题时任务会退回开发者修改。如果没有轮数上限理论上会无限循环。设一个合理上限比如 5 轮既允许修正又避免死循环和成本失控。4.3 环境变量把 API Key 设置到当前环境export LLM_API_KEYyour-api-key-here如果你在 Windows 上跑用set LLM_API_KEYyour-api-key-here。这一步只是把密钥准备好真正调用逻辑在后面的代码里统一从os.environ读取。5. 五角色 Agent Team 最小实现下面我们按模块写一个可运行的参考实现。这个实现不会很长但包含了五角色协作的核心骨架角色定义、消息结构、任务状态、提示词模板、调度逻辑。5.1 角色与消息结构文件路径agent_team/roles.pyfrom enum import Enum from dataclasses import dataclass, field from typing import List, Optional class Role(str, Enum): COORDINATOR coordinator PLANNER planner DEVELOPER developer REVIEWER reviewer OPERATOR operator class TaskStatus(str, Enum): CREATED created PLANNED planned DEVELOPED developed REVIEWED reviewed REJECTED rejected DELIVERED delivered FAILED failed dataclass class TaskMessage: task_id: str source_role: str target_role: str content: str status: TaskStatus TaskStatus.CREATED feedback: Optional[str] None trace: List[str] field(default_factorylist) def add_trace(self, node: str) - None: self.trace.append(node)这段代码定义了两类核心对象Role枚举五个角色也就是整条链路的节点。TaskMessage角色之间传递的任务消息。content是当前阶段的产出feedback是评审意见trace是任务经过的所有节点记录。为什么trace这么重要因为它就是过程留痕。任务跑完后你可以打印trace看到任务从协调者到运营者经过了哪些节点、有没有被退回。这是单 Agent 模式没有的东西。5.2 提示词模板文件路径agent_team/prompts.py每个角色需要独立的系统提示词。注意提示词的写法直接决定角色边界是否清晰。ROLE_PROMPTS { coordinator: ( 你是协调者负责接收用户目标并拆解任务。 你不编写业务代码不制定技术细节。 你的输出是一份包含目标、约束、交付物清单的任务分解。 ), planner: ( 你是规划者负责将任务分解转化为可执行的技术方案。 你的输出必须包含具体步骤、验收标准、优先级和风险点。 你不直接编写最终代码。 ), developer: ( 你是开发者负责按规划者的方案实现具体成果。 你的输出是完整的代码、文档或配置。 你不改变需求范围不跳过评审。 ), reviewer: ( 你是评审者负责检查开发者的产出是否满足验收标准。 你的输出是评审结论和问题清单。 如果存在问题必须明确给出修改意见。 你不直接修改代码不替开发者重写实现。 ), operator: ( 你是运营者负责对评审通过的成果做最终检查、格式整理和交付。 你的输出是最终交付物、使用说明和部署建议。 你不改变业务逻辑。 ), }这段代码的要点是每个 Prompt 都有“你不做什么”的表述。在设计多 Agent 系统时给角色划定“不做边界”比堆砌“你可以做什么”更有效。文件路径agent_team/agent.pyimport os from openai import OpenAI class LLMAgent: def __init__(self, role: str, cfg: dict): self.role role self.cfg cfg self.client OpenAI( base_urlcfg[llm][base_url], api_keyos.environ.get(cfg[llm][api_key_env]), ) def run(self, message_content: str, context: str ) - str: system_prompt ROLE_PROMPTS[self.role] messages [ {role: system, content: system_prompt}, ] if context: messages.append({role: system, content: f上下文信息\n{context}}) messages.append({role: user, content: message_content}) response self.client.chat.completions.create( modelself.cfg[llm][model], messagesmessages, temperatureself.cfg[llm][temperature], max_tokensself.cfg[llm][max_tokens], ) return response.choices[0].message.contentLLMAgent是单个角色的执行单元。它做的事很简单把角色提示词和当前输入拼成消息调用模型返回结果。这里把“模型调用”和“角色调度”分开后面扩展会很方便。5.3 调度引擎文件路径agent_team/engine.py调度引擎是五角色架构的核心。它负责决定任务当前该由哪个角色处理以及下一个流转到谁。import uuid from .roles import Role, TaskMessage, TaskStatus class AgentTeamEngine: def __init__(self, agents: dict, max_rounds: int 5): self.agents agents self.max_rounds max_rounds def run(self, user_task: str) - TaskMessage: task TaskMessage( task_iduuid.uuid4().hex, source_roleuser, target_roleRole.COORDINATOR.value, contentuser_task, ) task.add_trace(user) for _ in range(self.max_rounds): current_role task.target_role agent self.agents.get(current_role) if not agent: task.status TaskStatus.FAILED return task task.add_trace(current_role) if current_role Role.COORDINATOR.value: task.content agent.run(task.content) task.target_role Role.PLANNER.value task.status TaskStatus.PLANNED elif current_role Role.PLANNER.value: task.content agent.run(task.content) task.target_role Role.DEVELOPER.value task.status TaskStatus.DEVELOPED elif current_role Role.DEVELOPER.value: task.content agent.run(task.content) task.target_role Role.REVIEWER.value task.status TaskStatus.REVIEWED elif current_role Role.REVIEWER.value: review agent.run( task.content, contextf请评审以下成果并输出 PASS 或修改意见\n{task.content}, ) if PASS in review.upper(): task.target_role Role.OPERATOR.value task.feedback review task.status TaskStatus.REVIEWED else: task.target_role Role.DEVELOPER.value task.feedback review task.status TaskStatus.REJECTED elif current_role Role.OPERATOR.value: task.content agent.run(task.content) task.status TaskStatus.DELIVERED task.target_role None return task task.status TaskStatus.FAILED return task这段代码里最值得关注的是评审环节的分支逻辑。评审者输出的内容里如果包含PASS任务流向运营者否则流回开发者并携带feedback作为修改意见。这就是前文说的“任务可以在角色之间往返”也是 v3.1 这类版本强调的规则化流转。注意这里为了演示用PASS关键字做判定。真实项目中建议让评审者输出结构化 JSON用字段判断而不是字符串匹配后面常见问题部分会展开讲。5.4 主入口文件路径main.pyimport yaml from agent_team.agent import LLMAgent from agent_team.engine import AgentTeamEngine from agent_team.roles import Role def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): cfg load_config(config.yaml) agents { role.value: LLMAgent(role.value, cfg) for role in Role } engine AgentTeamEngine(agents, max_roundscfg[agent_team][max_rounds]) user_task 设计并实现一个命令行待办事项管理工具要求支持增删改查并提供使用说明。 result engine.run(user_task) print( 任务执行结束 ) print(任务状态:, result.status.value) print(流转轨迹:, - .join(result.trace)) print(最终输出:) print(result.content) if result.feedback: print(评审反馈:) print(result.feedback) if __name__ __main__: main()主入口做三件事加载配置、按五个角色创建 Agent、把用户任务交给调度引擎执行。最后打印任务状态、流转轨迹和最终输出。这里用“待办事项管理工具”作为示例任务目的是让大家跑通流程而不是制造一个真实可用的产品。6. 运行流程与效果验证6.1 运行命令项目目录结构如下agent-team-demo/ ├── config.yaml ├── main.py └── agent_team/ ├── __init__.py ├── agent.py ├── engine.py ├── prompts.py └── roles.py执行python main.py6.2 预期输出如果一切正常你会看到类似下面的输出结构具体内容取决于模型返回 任务执行结束 任务状态: delivered 流转轨迹: user - coordinator - planner - developer - reviewer - operator 最终输出: 【待办事项管理工具使用说明】 1. 添加python todo.py add 任务内容 2. 查看python todo.py list 3. 修改python todo.py update id 新内容 4. 删除python todo.py delete id ...6.3 如何判断成功判断一次运行是否成功不能只看“有没有输出”要看三个信号第一任务状态为delivered。说明流程完整走完而不是中途失败。第二流转轨迹包含全部五个角色且顺序正确。如果轨迹里出现了reviewer - reviewer这种重复或者developer - planner这种回退说明调度逻辑或提示词可能有异常。第三评审环节有PASS或修改反馈。如果评审者一直不通过任务会在循环里被反复退回直到触发max_rounds上限变为FAILED。这不是 Bug而是评审闸门在起作用你反而应该检查开发者产出的质量或者评审者的判定标准是否过严。6.4 如果运行失败第一步看哪里按照下面的顺序排查看日志里有哪个角色没执行到。如果卡在coordinator就没往下走大概率是 API 调用失败先检查base_url和api_key。看异常信息。网络超时、鉴权失败、模型名错误都会有明确报错。看任务状态。如果变成FAILED十有八九是超过了max_rounds此时把评审者的判定标准放宽或者提高轮数上限。7. 常见问题与排查方法问题现象可能原因排查方式解决方案任务卡在第一个角色不动API Key 缺失或错误检查环境变量LLM_API_KEY是否存在正确设置环境变量确认 Key 有效调用报 401 或 403鉴权失败或没有调用权限查看完整报错信息核对base_url与 Key 的匹配关系调用报 404模型名错误或接口路径不对打印请求的 URL 和模型名改成服务商提供的正确模型名称角色之间互相抢活每个角色的 Prompt 边界不清晰检查各角色输出是否越界强化每个 Prompt 中的“不应做”约束评审一直不通过循环到上限评审者标准过严或判定逻辑有误查看feedback内容是否合理调整评审 Prompt必要时改用结构化 JSON 判定上下文过长导致超时或超限每个角色都传了全量历史检查传入context的内容量按需传递上下文只传当前角色需要的信息最终输出格式混乱运营者 Prompt 缺少格式要求查看运营者输出是否符合预期在运营者 Prompt 中明确输出格式模板单个角色任务状态异常调度引擎分支条件写错检查engine.py中状态流转分支补齐异常分支设置兜底的FAILED状态这里我要重点提醒一个真实项目中很容易忽略的问题不要让评审者只输出“通过”或“不通过”这种短文本。更稳妥的做法是要求评审者输出结构化 JSON例如{ verdict: PASS, score: 85, issues: [ {severity: medium, description: 缺少参数校验, suggestion: 增加对输入参数的合法性检查} ] }用 JSON 而不是关键字匹配有三个好处判定更稳定、反馈可追踪、问题清单可以直接作为开发者下一轮的修改上下文。上面的最小示例为了控制篇幅用了PASS关键字生产环境建议按 JSON 方案改造。8. 最佳实践与工程建议8.1 上下文按需传递不要全量堆叠多角色协作最大的成本陷阱就是盲目堆上下文。协调者规划完就不需要再关注对话的全部开发者拿到方案后只需要方案文本不需要用户最初的十轮闲聊。在实际项目中我建议为每个角色定义一个“上下文窗口”协调者用户原始目标、约束条件、协作历史摘要。规划者任务分解清单、相关背景资料。开发者完整技术方案、评审反馈如果有。评审者产出物、验收标准。运营者评审通过的产出物、交付格式要求。让每个角色只拿自己需要的信息既控制成本也降低噪音干扰。8.2 交接物结构化为 JSON角色之间传递的内容最好用固定结构。前面TaskMessage已经演示了这种思路但在生产环境里content字段内部也应该结构化。比如规划者的输出不要是一大段散文而应该是{ objective: 实现命令行待办工具, steps: [ {id: 1, action: 定义数据存储格式, output: storage.py}, {id: 2, action: 实现增删改查命令, output: todo.py} ], acceptance_criteria: [支持增删改查, 提供使用说明], risks: [文件并发写入冲突] }结构化输出意味着后续角色可以直接从 JSON 里取字段而不是让模型从一大段文字里“猜”关键信息。8.3 日志与可观测性五角色架构的竞争力很大程度来自可观测性。建议每个角色执行时记录开始时间、结束时间。输入内容的长度或摘要。输出内容的长度或摘要。状态变更。日志不用记录全量对话那样会非常大。记录摘要就够了目的是定位“哪一步慢、哪一步错、哪一步状态异常”。等任务失败时你能直接回答“问题出在哪个角色”而不是从头到尾重新跑一遍。8.4 成本控制每次调用模型都要花钱五角色架构跑一个任务至少调用五次模型加上评审退回可能更多。控制成本的方法有三个第一给每个角色配不同的模型。简单的角色比如运营者可以用速度更快的模型复杂的角色比如规划者和评审者用能力更强的模型。第二个方法是为角色设置max_tokens防止输出失控长文。第三设置总轮数上限防止无限循环烧钱。8.5 安全与权限边界如果 Agent 团队要操作真实系统比如写文件、执行命令、调用数据库务必遵循最小权限原则。不要让开发者角色拥有生产环境的写权限如果必须操作先在测试环境验证执行前备份并确保所有变更可回滚。评审者角色在设计上可以承担一部分“安全检查”的职责但规则要明确涉及删除、覆盖、批量修改的操作必须走人工确认。8.6 什么时候不应该用五角色架构这是最容易被忽略的一点。五角色架构是有成本的每次任务都调用五个模型延迟更高、费用更多、失败概率更大。如果你的任务只有一个简单步骤比如“把这段文本改成英文”用单 Agent 就够了。五角色架构适合的是“值得拆成多阶段并需要质量闸门”的任务而不是所有任务。9. 总结与后续学习方向回到开头那个判断单 Agent 做长链路任务会失控本质原因不是模型不够聪明而是缺少角色分工、交接机制和过程留痕。五角色架构解决的就是这三个问题。Hermes Agent Team 的 v3.1从架构演进角度看也在沿着“规则化流转、按需传上下文、过程可观测”这三个方向收敛。这篇文章的参考实现很简但它把这个模式的核心部分跑通了角色定义、消息结构、提示词边界、调度引擎、评审回退机制。你可以把它作为骨架往里面加真实业务逻辑。下一步建议这样实践先跑通最小示例然后把评审者的输出改成结构化 JSON再给每个角色配上真实业务场景的提示词。当你把交接物、上下文窗口、日志和轮数上限都调稳之后再考虑接入更多外部工具比如文件系统、数据库或代码执行环境。最后提醒一句Agent 团队的价值不在于角色多而在于每个角色都能被约束住。把职责边界、交接结构、质量闸门和可观测性这四件事做好五个角色就够用了。建议收藏备用。等你要做 Agent 自动化流程的时候按这篇文章的步骤搭一套最小实现再逐步完善。
返回列表