从零搭建五角色多智能体团队：一人公司架构实践指南-平芜编程栈

在个人开发者和极小型团队里，AI 经常被用来模拟“一人公司”：一个人同时承担产品、研发、运营、客服、财务等多个职能。真正实践过的人会发现，单个 Agent 很难稳定完成这套工作流。Hermes Agent Team 正是围绕这个问题设计的五角色架构，它把任务承接、技能执行、知识记忆、质量检查和运行安全拆成独立角色，形成一人公司模型。v3.1 版本在这个架构上的主要变化，是把角色职责、Skill 技能注册、记忆管理拆成清晰的工程层次，让一个演示项目具备直接演进成生产系统的能力。这篇博客会从零搭建一个最小可复现的 Hermes Agent Team 项目，覆盖角色定义、消息协议、记忆层、Skill 机制、运行验证和常见故障排查。

1. 一人公司为什么需要五角色 Agent 团队，而不是单个 Agent

很多人第一次搭建 Agent 项目时说，我只需要一个“超级提示词”，把所有工作要求写进去，让模型自己完成。但实际跑下来，往往会在任务变复杂后迅速失控。要理解 Hermes Agent Team v3.1 的设计，首先要承认单个 Agent 存在三个明显天花板。

1.1 单个 Agent 的天花板：上下文、记忆和职责不分家

上下文窗口有限是最直接的问题。一个 Agent 如果既要读懂需求、又要调用多个工具、又要检索历史知识、又要输出长文，上下文很快会被占满。更隐蔽的问题是职责耦合：当“拆任务”“查资料”“写文档”“检查质量”“处理报错”全部揉在同一个 System Prompt 里，模型很难在正确时机切换行为模式。结果往往是任务分析做得不完整，执行阶段却急着调用工具，质检阶段又只做格式检查，完全没有事实校验。

记忆混乱是另一个高频现象。同一套会话记忆里如果既有用户偏好、又有项目资料、又有历史报错，下一次任务开始时，Agent 很可能把上一次任务的知识带进来，造成“串味”。在很多 demo 项目里，这个现象表现为第二次生成的文章明显偏离新主题。

职责不分还带来了审计困难。任务出问题时，你不知道是任务拆分错了、工具调用错了、还是输出校验漏了。因为在单 Agent 模式下，所有环节都发生在同一个黑盒上下文里，没有中间产物，没有角色边界，也没有清晰的失败责任点。

1.2 五角色职责模型：从任务承接、拆分、执行、质检到复盘

Hermes Agent Team v3.1 把“一人公司”拆成五个固定角色，每个角色只负责一个稳定职责。

角色	对应一人公司职能	核心职责
Coordinator 主编	CEO / 项目经理	接收用户需求，拆分任务，汇总结果，决定流程是否继续
Executor 执行者	研究员 / 写作者 / 研发人员	调用 Skill 完成检索、计算、内容生成等具体动作
Librarian 知识管理员	知识库 / 文档中心	管理长期记忆，完成知识写入、检索和去重
Quality Gate 质检员	测试 / 审核员	检查输出是否符合格式、长度、事实一致性要求
Ops Guard 运维员	运维 / 安全员	监控任务状态，处理超时、权限、日志和重试

这个设计的第一原则是“职责单一”。Coordinator 不需要自己调用搜索引擎，Executor 不需要维护长期知识库，Quality Gate 不需要参与任务拆分。各角色之间通过消息协作，而不是共享同一个超大 Prompt。

第二原则是“每个角色都可以独立替换”。如果你的任务从“写文章”变成“写代码”，Coordinator、Librarian、Quality Gate 可以基本不变，只要替换 Executor 的 Skill 组合即可。这就是一人公司模型的价值：公司职能不变，具体执行能力随项目切换。

1.3 v3.1 的核心变化：把 Skill、记忆和角色边界拆成三层

v3.1 对比早期版本，工程结构上不再把技能和记忆直接写进角色 Prompt，而是拆成三层。

第一层是角色层，只定义角色的行为目标、消息处理逻辑和允许使用的 Skill 列表。第二层是 Skill 层，负责具体工具能力，比如网页搜索、数据库查询、Markdown 报告生成、代码执行。第三层是记忆层，负责短期会话和长期知识的读写。加上贯穿整个过程的事件日志与安全策略，构成 v3.1 的完整运行模型。

这样拆分后，工具能力可以跨角色复用。同一个“网页检索”Skill，Executor 可以用，Quality Gate 也可以用于事实核对。长期记忆可以按角色隔离，避免 Coordinator 的会话上下文污染 Executor 的知识检索。同时，由于每个 Skill 都有独立输入输出定义，日志里能清楚记录“谁在什么时候调用了什么工具”，这也是 v3.1 在可审计性上最重要的改进。

这里顺便解释一个很多人问过的概念差异：harness 和 agent 的区别。Harness 是 Agent 的“运行外壳”，负责生命周期、工具调用循环、日志、超时和错误重试；Agent 则是“决策大脑”，负责目标规划和下一步动作选择。v3.1 的工程配置同时管理两层：角色 Prompt 属于 agent 层，消息总线、超时机制、权限检查属于 harness 层。很多故障排查到最后，问题不是模型不够聪明，而是 harness 层缺少超时、重试和终止原因记录。

2. 先搭好环境：用一套最小工程骨架避免后面全部返工

Hermes Agent Team v3.1 不需要依赖某一套特定商业产品。它是一套可以用任意 LLM 接口搭出来的多角色 Agent 架构。下面以 Python 作为实现语言，展示一个最小工程骨架。先对齐环境，再开始写角色。

2.1 运行环境与依赖版本

建议使用 Python 3.11 或更高版本，主要考虑是类型注解和tomllib等标准库能力更完整。环境隔离使用虚拟环境。

python -m venv .venv source .venv/bin/activate pip install --upgrade pip

下面这个requirements.txt是最小依赖集。为了先跑通流程，可以暂时用 Mock 模型代替真实 LLM 调用；接入真实模型时再增加对应 SDK。

pyyaml>=6.0 pydantic>=2.0 httpx>=0.27 rich>=13.0

解释一下为什么这样选：

pyyaml：加载config.yaml配置，把角色、超时、模型参数外置。
pydantic：定义 Role、Message、Skill 等数据模型，减少手写校验逻辑。
httpx：用于真实场景中调用模型 HTTP 接口。演示阶段可以实现一个MockLLMClient，不发出真请求。
rich：在命令行里格式化输出日志和事件流，方便观察角色协作过程。

2.2 项目目录结构设计

工程骨架建议按“配置、核心模块、角色、技能、数据、日志”分层。

hermes_agent_team/ ├── config/ │ ├── config.yaml │ └── roles/ │ ├── coordinator.yaml │ ├── executor.yaml │ ├── librarian.yaml │ ├── quality_gate.yaml │ └── ops_guard.yaml ├── hermes/ │ ├── __init__.py │ ├── core/ │ │ ├── role.py │ │ ├── message.py │ │ ├── memory.py │ │ ├── skill.py │ │ └── orchestrator.py │ ├── skills/ │ │ ├── web_search.py │ │ └── report_writer.py │ └── roles/ │ ├── coordinator.py │ ├── executor.py │ ├── librarian.py │ ├── quality_gate.py │ └── ops_guard.py ├── data/ │ ├── memory/ │ └── output/ ├── logs/ └── main.py

这个目录的划分原则是：config放可变配置，hermes/core放不依赖业务的通用能力，hermes/skills放可复用工具，hermes/roles放五个角色的具体实现，data放运行产生的数据，logs放日志。

实际项目里，目录结构可以按团队习惯调整，但建议保留“配置外置、核心通用、技能独立、角色薄层”这个思想。如果一上来就把某个 Skill 的调用逻辑写进 Executor 的类里，后面再用到同一个工具就得多写一份重复代码。

2.3 配置文件：模型、角色、阈值一起外置

config/config.yaml用最小配置覆盖三块内容：模型接入、运行阈值、日志级别。

llm: provider: mock base_url: "" api_key_env: "LLM_API_KEY" model: "hermes-v3.1-demo" orchestrator: max_iterations: 10 global_timeout_seconds: 120 retry_times: 2 retry_backoff_seconds: 2 memory: short_memory_limit: 20 long_term_dir: "data/memory" logging: level: "INFO" log_dir: "logs" log_file: "orchestrator.log"

在 v3.1 里，provider: mock是很重要的学习起点。它让你不依赖外部模型也能验证角色协作、消息流转和目录结构是否正确。等骨架跑通后，再换成provider: openai_compatible或本地模型端点。

注意api_key_env的设计：密钥不要写死在 YAML 里，而是从环境变量读取。这是生产环境的安全底线，演示阶段也不要养成硬编码习惯。

max_iterations和global_timeout_seconds是防止“角色互相等待”的关键参数。后面排错章节会专门讲这两个参数的表现。

3. 五角色模型落地：数据模型、消息协议、记忆层与 Skill 机制

环境准备好之后，开始写核心代码。这一章是 v3.1 架构的技术核心，主要包含四个部分：统一的 Role 数据模型、消息协议、记忆层和 Skill 注册机制。先定义清楚数据模型，才能避免后续角色之间传参错乱。

3.1 Role 数据模型：一个角色 = 提示词 + 工具 + 记忆 + 输出约束

在 v3.1 里，角色不是简单的“一段 Prompt”，而是一个结构化对象。它应该包含角色名称、系统提示词、可用 Skill、记忆范围、是否需要人工审批、最大执行轮数等配置。

from typing import Literal from pydantic import BaseModel, Field MemoryScope = Literal["short", "long", "none"] ApprovalType = Literal["none", "before_skill", "before_output"] class Role(BaseModel): name: str = Field(..., description="角色唯一名称") display_name: str = Field(..., description="角色展示名") system_prompt: str = Field(..., description="角色行为指令") skills: list[str] = Field(default_factory=list, description="允许使用的 Skill 名称列表") memory_scope: MemoryScope = Field(default="short", description="记忆使用范围") approval: ApprovalType = Field(default="none", description="人工审批节点") max_iterations: int = Field(default=3, description="角色内部最大执行轮数") timeout_seconds: int = Field(default=30, description="角色单次处理超时")

这个设计的核心是：Skill 通过skills字段声明，而不是写进 Prompt。Coordinator 不需要读 Executor 的完整工具说明，只要知道“Executor 可以处理子任务执行”。这样 Prompt 更短，模型更不容易跑偏。

角色配置文件config/roles/coordinator.yaml对应如下：

name: coordinator display_name: 主编 system_prompt: | 你是 Hermes Agent Team v3.1 的主编角色。 你可以接收用户需求，将其拆分为多个可并行执行的子任务。 你负责判断子任务结果是否满足原始需求，并输出最终汇总。 不要自己直接调用具体业务技能。 skills: [] memory_scope: short approval: none max_iterations: 3 timeout_seconds: 30

不同角色可以基于这个模型扩展字段。比如同一个人公司模型在写代码场景下，Coordinator 的skills仍为空，但 Executor 的skills会变成["code_generator", "test_runner"]。

3.2 消息协议：让五个角色在统一总线上协作

角色之间不直接调用对方的方法，而是通过消息对象传递任务和结果。这是解耦五个角色的关键。

import uuid from datetime import datetime, timezone from typing import Any, Literal from pydantic import BaseModel, Field MessageType = Literal[ "task_request", "sub_task", "result", "quality_feedback", "memory_write", "memory_read", "event_log", "terminate", ] class Message(BaseModel): message_id: str = Field(default_factory=lambda: uuid.uuid4().hex) task_id: str = Field(..., description="全局任务 ID，用于追踪一条业务链路") source: str = Field(..., description="发送方角色名") target: str = Field(..., description="接收方角色名") msg_type: MessageType = Field(..., description="消息类型") content: dict[str, Any] = Field(default_factory=dict, description="消息内容") parent_message_id: str | None = Field(default=None, description="父消息 ID") created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

消息协议里的task_id是排查故障的关键线索。同一条业务链上，所有角色处理的消息都会带同一个task_id。日志系统只要支持按task_id过滤，就能快速还原“任务从用户请求到最终输出的完整路径”。

parent_message_id用于记录消息之间的因果链。比如 Quality Gate 对 Executor 的初稿提出修改意见，这条反馈消息的parent_message_id应该指向 Executor 提交初稿的那条消息。这个字段在工作流审计时非常有用。

3.3 记忆层：短期对话记录和长期知识库必须分离

v3.1 对记忆的核心要求是隔离。短期记忆只服务当前任务，长期记忆按角色和标签隔离。一个常见的错误是把所有对话历史放在同一个列表里，结果第二个任务开始时，第一个任务的资料仍然残留在上下文中。

一个最小可用的记忆层可以这样实现：

import json from pathlib import Path from typing import Any class Memory: def __init__(self, role_name: str, long_term_dir: str, short_limit: int = 20): self.role_name = role_name self.long_term_dir = Path(long_term_dir) self.short_memory: list[dict[str, Any]] = [] self.short_limit = short_limit self.long_term_dir.mkdir(parents=True, exist_ok=True) def add_short_term(self, role: str, content: Any) -> None: self.short_memory.append({"role": role, "content": content}) if len(self.short_memory) > self.short_limit: self.short_memory.pop(0) def get_short_term(self) -> list[dict[str, Any]]: return list(self.short_memory) def clear_short_term(self) -> None: self.short_memory.clear() def write_long_term(self, key: str, content: Any, tag: str = "general") -> None: path = self.long_term_dir / f"{self.role_name}_{tag}.json" records = [] if path.exists(): records = json.loads(path.read_text(encoding="utf-8")) records.append({"key": key, "content": content}) path.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8") def search_long_term(self, keyword: str, tag: str = "general") -> list[dict[str, Any]]: path = self.long_term_dir / f"{self.role_name}_{tag}.json" if not path.exists(): return [] records = json.loads(path.read_text(encoding="utf-8")) return [r for r in records if keyword in str(r.get("content", ""))]

在真实项目中，长期记忆可以替换成向量数据库，用向量的语义相似度替代这里的字符串包含检索。但从学习角度看，先理解“短期记忆按任务清空、长期记忆按角色隔离”这两个原则，比直接上向量数据库更重要。

clear_short_term()在什么时候调用很关键。任务结束时，Coordinator 应广播一条memory_clear事件，让所有参与角色的短期记忆按task_id清空。否则下一次任务的上下文会带着上一次任务的残留内容。

3.4 Skill 机制：为什么要从 Agent 里把工具能力拆出来

在 v3.1 模型里，Skill 是角色可以调用的外部能力单元。它必须包含三个部分：输入定义、执行逻辑、输出定义。

from abc import ABC, abstractmethod from typing import Any class Skill(ABC): name: str = "base_skill" description: str = "" @abstractmethod def validate_input(self, payload: dict[str, Any]) -> None: """检查输入参数是否合法，不合法时抛出异常。""" @abstractmethod def execute(self, payload: dict[str, Any]) -> dict[str, Any]: """执行具体能力，返回结构化结果。"""

以web_search.py为例，它屏蔽了底层搜索实现细节：

from hermes.core.skill import Skill class WebSearchSkill(Skill): name = "web_search" description = "检索互联网资料，返回标题、链接和摘要列表" def validate_input(self, payload: dict[str, Any]) -> None: if not payload.get("query"): raise ValueError("web_search requires 'query'") def execute(self, payload: dict[str, Any]) -> dict[str, Any]: query = payload["query"] # 演示阶段返回模拟结果，真实场景在这里封装搜索 API return { "query": query, "results": [ {"title": "示例标题", "url": "https://example.com", "snippet": "示例摘要"} ] }

另一个常用 Skill 是report_writer.py，它负责把结构化内容渲染成 Markdown 报告。这样 Executor 不需要关心搜索接口、报告模板和文件写入，它只需要决定“先搜索什么，再总结什么”。Skill 拆分得越细，复用性越高，也越容易做单元测试。

Skill 与 Agent 的区别可以从两个角度看。Agent 是决策者，决定要不要调用工具、调用哪个工具、如何使用工具结果；Skill 是执行者，只负责完成一个确定的功能。把工具调用逻辑从 Agent 的 Prompt 和模型决策里抽出来，是 v3.1 提高稳定性的核心手段。

3.5 安全边界：给执行角色加上工具权限和白名单

多角色 Agent 项目里，安全不只是“接口加个 Token”，还要考虑角色权限边界。在 v3.1 里，安全策略至少包含三个层面。

第一层是 Skill 白名单。Executo 角色配置里声明了skills: ["web_search", "report_writer"]，运行时编排器只允许这些 Skill 被调用。如果某个角色试图调用未注册的 Skill，直接拒绝并写审计日志。

第二层是命令和路径约束。如果 Skill 里涉及读写文件或执行命令，必须在 Skill 层做路径校验，防止通过恶意参数写出到系统目录。

from pathlib import Path def validate_output_path(base_dir: Path, target: str) -> Path: base = base_dir.resolve() target_path = (base / target).resolve() if not target_path.is_relative_to(base): raise PermissionError(f"target path is outside base dir: {target}") return target_path

第三层是人工审批节点。Role.approval字段可以配置为before_skill或before_output。比如在“发布文章”这一步，可以要求 Coordinator 在输出最终报告前暂停，等待人工确认。这个机制用很小的成本规避了“全自动流程不可控”的风险。

4. 实现最小闭环：一篇行业调研文章是如何被五个角色协作完成的

理论部分结束后，用一个完整场景验证架构：用户提交需求，五个角色协作输出一份技术调研报告。这里不依赖真实模型调用，而是用 Mock 模型模拟关键决策，重点观察消息流转和事件日志是否正确。

4.1 场景与输入

用户输入：

调研 2025 年 Agent 开发框架的选型趋势，输出一份 3000 字左右的技术报告。

这条需求进入系统后，会被包装成task_request消息：

{ "task_id": "task-demo-001", "source": "user", "target": "coordinator", "msg_type": "task_request", "content": { "raw_request": "调研 2025 年 Agent 开发框架的选型趋势，输出一份 3000 字左右的技术报告。" } }

从这一步开始，所有后续消息都会携带task_id: task-demo-001。这就为日志追踪提供了统一维度。

4.2 主编角色：把任务拆成可执行子任务

Coordinator 收到任务后，职责是把模糊需求拆成有序子任务。在不调用真实模型的情况下，可以写一个规则版的拆分器来演示。

class Coordinator: name = "coordinator" def handle(self, message: Message, ctx) -> list[Message]: raw_request = message.content["raw_request"] sub_tasks = self.plan(raw_request) return [ Message( task_id=message.task_id, source=self.name, target="executor", msg_type="sub_task", content=sub_task, parent_message_id=message.message_id, ) for sub_task in sub_tasks ] def plan(self, raw_request: str) -> list[dict]: # 最小示例：固定拆成五步 return [ {"step": "collect_materials", "instruction": "检索 Agent 开发框架相关资料"}, {"step": "write_report", "instruction": "基于资料输出技术报告初稿"}, {"step": "quality_check", "instruction": "检查报告结构与长度"}, ]

这个例子里 Coordinator 先向 Executor 发送collect_materials，再发送write_report。真实项目中，Coordinator 会把执行结果传回自己，判断是否满足原始需求，然后再决定下发下一步子任务或输出最终结果。

4.3 执行角色：调用 Skill 完成调研和初稿

Executor 收到sub_task后，根据step字段选择 Skill。它的处理逻辑不是直接写死“新文章”，而是先读config/roles/executor.yaml里的 skills 配置，再做分发。

class Executor: name = "executor" def __init__(self, skills: dict[str, Skill]): self.skills = skills def handle(self, message: Message, ctx) -> list[Message]: step = message.content.get("step") if step == "collect_materials": result = self.skills["web_search"].execute({ "query": message.content.get("instruction") }) return [self.build_result_message(message, result)] if step == "write_report": result = self.skills["report_writer"].execute({ "instruction": message.content.get("instruction") }) return [self.build_result_message(message, result)] # 未知步骤，返回错误信息 return [self.build_error_message(message, f"unknown step: {step}")]

这个实现体现了一层重要边界：Executor 不维护知识库，不判断最终质量，只负责“调度 Skill、收集结果、格式化输出”。职责越小，模型决策越稳定。

4.4 质检与运维角色：自动化检查输出质量

Quality Gate 收到 Executor 提交的result后，执行一组规则校验。最小校验可以包含：长度是否达标、是否包含 Markdown 标题、是否存在空段落、是否包含风险关键词。

class QualityGate: name = "quality_gate" def handle(self, message: Message, ctx) -> list[Message]: content = message.content.get("text", "") issues = [] if len(content) < 2000: issues.append("report length is less than 2000 chars") if "## " not in content: issues.append("report missing H2 headline") if "TODO" in content: issues.append("report contains TODO placeholder") if issues: return [self.build_feedback(message, issues)] return [self.build_pass_message(message)]

当质检不通过时，Quality Gate 会向 Executor 发送quality_feedback。如果系统配置了max_iterations，Executor 可以对反馈进行修订，但不能无限重试。这正好体现 v3.1 中 role 模型的max_iterations字段价值。

Ops Guard 的角色比较特殊。它不参与内容生产，而是监听事件流，把每条消息的耗时、角色、结果状态写入日志。当出现超时或重试超限时，Ops Guard 负责广播terminate消息终止任务。

4.5 全链路事件日志：用事件流验证协作顺序

演示项目的编排器可以很简单：按顺序处理消息，并把每一条消息打印成结构化日志。下面这段日志展示了正常流程的事件流：

2025-06-08T10:00:01Z [event] task_demo_001 coordinator receive task_request 2025-06-08T10:00:02Z [event] task_demo_001 coordinator -> executor sub_task collect_materials 2025-06-08T10:00:03Z [event] task_demo_001 executor call skill web_search 2025-06-08T10:00:03Z [event] task_demo_001 executor -> coordinator result collect_materials 2025-06-08T10:00:04Z [event] task_demo_001 coordinator -> executor sub_task write_report 2025-06-08T10:00:06Z [event] task_demo_001 executor call skill report_writer 2025-06-08T10:00:06Z [event] task_demo_001 executor -> quality_gate result write_report 2025-06-08T10:00:08Z [event] task_demo_001 quality_gate -> executor quality_feedback length_short

当读者看到这类日志时，可以马上确认流程有没有走到 Quality Gate。如果一个项目跑完只看到 coordinator 和 executor 的日志，说明质检角色根本没有被接入编排器。这也是 v3.1 最典型的一个落地问题：五个角色在配置文件中都存在，但实际执行链路只用了两个。

5. 运行验证：单角色、双角色、全链路三种方式依次确认

验证阶段不要直接跑完整流程。推荐按从易到难的顺序分三步验证，这样出问题时能快速定位。

5.1 启动命令与最小验证脚本

先确认核心模块可以导入：

cd hermes_agent_team python -c "from hermes.core.role import Role; print(Role(name='demo', system_prompt='test'))"

再运行主入口：

python main.py --task "调研 2025 年 Agent 开发框架的选型趋势" \ --output data/output/report.md

main.py的启动流程应包含四个动作：加载config.yaml、加载五个角色配置、初始化 Skill 注册表、运行编排器。只有四个动作都完成，任务才正式进入第一轮。

5.2 预期输出：一份可继续人工编辑的 Markdown 报告

演示项目正常结束时，data/output目录下应生成一份 Markdown 报告，包含至少两个 H2 章节，并且正文中不应该有TODO占位符。

报告开头示例：

# Agent 开发框架选型趋势调研分析 ## 1. 背景与方法 ## 2. 主流框架能力对比 ## 3. 选型建议 ## 4. 风险与展望

如果报告只生成了标题而没有正文，说明 Executor 的report_writerSkill 只完成了模板渲染，没有把搜索结果写入正文。此时要检查 Executor 是否把collect_materials的结果传给了write_report阶段。

5.3 三个验证检查点：角色状态、工具调用、最终产物

验证时重点看三个检查点。

第一个检查点是角色状态。在日志中确认五个角色是否全部被加载，且每个角色都至少处理了一条消息。如果某个角色从未出现，优先检查config/config.yaml里的 orchestrator 是否引用了该角色。

第二个检查点是工具调用。日志中应出现executor call skill web_search、executor call skill report_writer等事件。如果出现角色直接调用未注册 Skill 的报错，说明 Role 的skills列表与 Skill 注册表不一致。

第三个检查点是最终产物。不能只看进程退出码为 0，还需要检查输出文件是否存在、内容是否完整、是否通过 Quality Gate 的规则校验。建议在测试脚本中写断言：

from pathlib import Path report = Path("data/output/report.md") assert report.exists(), "report file not found" content = report.read_text(encoding="utf-8") assert len(content) > 2000, "report too short" assert "## " in content, "report missing H2"

这类自动化断言可以沉淀为 CI 测试。以后改动角色配置或 Skill 实现时，跑一遍就能快速发现问题。

5.4 如何确认 v3.1 配置确实生效了

很多人改完配置后，发现程序行为和改之前一样，原因是进程仍在运行旧配置。确认 v3.1 配置生效可以从三处观察。

第一，启动日志中是否打印了配置来源。例如：“load config from config/config.yaml, roles: coordinator, executor, librarian, quality_gate, ops_guard”。

第二，角色配置中的max_iterations和timeout_seconds是否被应用。可以在日志里看到超时前是否触发重试。

第三，记忆文件是否按角色写到data/memory/。如果 Librarian 角色正常工作，应该看到librarian_general.json等文件被创建。

6. 常见问题排查：任务终止、无响应、互相等待

多角色 Agent 项目最常见的故障集中在任务终止、超时无响应、角色互相等待、记忆污染四类。下面分别给出排查思路。

6.1 Agent Execution Terminated：模型报错还是编排器主动终止

现象是日志中出现类似“agent execution terminated due to error.”的提示，任务在某一个角色处停止。

先判断终止来源。如果是模型调用返回错误，日志会出现模型 API 的状态码和错误消息；如果是编排器主动终止，日志会记录termination_reason字段，比如max_iterations_exceeded或quality_gate_failed。

排查顺序：

打开logs/orchestrator.log，按task_id过滤所有事件。
找到第一条terminate消息，查看content.reason。
如果原因是max_iterations_exceeded，把对应角色的max_iterations调大，或检查该角色是否反复陷入同一个失败循环。
如果原因是模型 API 错误，先查看 API 返回的error_code，再决定是否需要降级到 Mock 模型。

一个常见坑是：Executor 每次重试都在生成同样的错误结果，导致max_iterations被快速消耗。这时候调大次数没有意义，应该检查 Skill 输入是否缺少关键字段，或 Quality Gate 的校验规则是否误判。

6.2 Provider Did Not Respond In Time：超时参数怎么查

现象是日志中出现类似“the agent execution provider did not respond in time”的错误。

这种错误说明模型提供方在指定时间内没有返回结果。原因通常有三个：模型推理速度过慢、网络链路超时、timeout_seconds设置过小。

排查方式：

检查config.yaml中的global_timeout_seconds和角色配置中的timeout_seconds。
查看模型 API 平均耗时。如果平均耗时接近超时阈值，应把超时时间上调到平均耗时的 3 倍以上。
查看是否有重试策略。retry_times: 2意味着单次失败会重试两次，这个机制可以缓解偶发网络问题。

推荐做法是把超时参数放到配置中心或环境变量，而不是硬编码在代码里。生产环境中，不同模型的响应速度差异很大，一个大模型可能 10 秒内返回，另一个小模型可能 3 秒返回，给全部角色设置同一个超时时间并不合理。

6.3 角色互相等待导致死锁：拓扑设计问题

现象是流程卡在某一轮，所有角色都不再产生新消息，日志没有任何报错。

这种问题往往不是单点错误，而是消息依赖关系设计错误。比如 Coordinator 等待 Executor 返回结果，而 Executor 又在等待 Coordinator 发布下一步指令；如果两者之间没有超时机制，任务就会卡死。

排查时需要画出消息流向。手工排查时，可以按parent_message_id关联所有消息，检查是否存在循环引用。

解决方向有两个：

给所有角色消息处理加上超时。任何角色处理消息超过timeout_seconds，Ops Guard 都应该收到事件并主动终止或重试。
减少同步等待。如果任务后台执行耗时较长，可以把“提交任务”和“获取结果”拆成两条异步消息，而不是让 Coordinator 一直阻塞。

v3.1 的编排器建议采用事件循环 + 消息队列的方式，而不是简单的函数调用链。函数调用链一旦某个环节阻塞，整个进程都会卡住；消息队列方式则可以保留任务状态，在超时后恢复。

6.4 记忆污染与 Skill 调用失败

现象是第二次执行任务时，输出内容明显包含第一次任务的数据。比如第一次调研“Java 日志框架”，第二次调研“Python Web 框架”，第二次报告里却出现了“Logback”。

原因通常是短期记忆没有按任务隔离。处理方式是在任务开始时给每个角色创建独立的短期记忆实例，任务结束后调用clear_short_term()。

Skill 调用失败的典型现象是角色提示“I cannot access the web”，但日志显示 Skill 已经执行成功。这通常意味着 Skill 的输出没有正确回传给模型上下文。排查时检查 Skillexecute的返回值是否包含在Message.content中，以及模型调用代码是否把该字段拼入对话历史。

6.5 一套从现象到根因的排查顺序表

问题现象	常见原因	检查方式	处理建议
任务在某角色处终止	编排器主动终止或模型报错	按`task_id`查询`terminate`消息	区分`max_iterations_exceeded`与 API 错误
Provider 无响应	超时参数过小或模型过慢	查看单次调用耗时与配置阈值	上调超时时间，增加重试和退避
角色互相等待	同步依赖链没有超时	检查`parent_message_id`是否存在环形等待	引入全局超时与消息队列
第二次任务内容串味	短期记忆未按任务隔离	检查角色记忆文件中是否有旧任务内容	任务开始/结束时刷新短期记忆
Skill 调用失败	Skill 注册名与角色配置不一致	对比`Role.skills`与 Skill 注册表	统一配置名，增加注册校验

7. 生产化最佳实践：从演示项目到一人公司长期运行

项目跑通后，下一步是把它从本地 demo 变成可以长期运行的一人公司工作台。这里最容易犯的错误是：把所有配置、密钥、模型调用直接写死，让后续每一次变更都变得困难。

7.1 发布前检查清单

下面这份清单适用于每次发布前逐项确认：

模型配置已经外置，api_key从环境变量读取，不进入 Git 仓库。
角色配置里的max_iterations和timeout_seconds已经被实际加载，而不是默认值。
每个 Skill 都通过了独立单元测试，包括正常输入、异常输入和边界输入。
所有角色消息都带有task_id，日志可以按任务链路完整还原。
Quality Gate 规则不只检查格式，还包含事实一致性关键词或数字校验。
记忆目录已纳入备份，长期记忆有清理和去重策略。
Ops Guard 的事件日志已接入采集系统，出现terminate时能产生告警。
输出文件有写入权限和冲突处理机制，重复执行不会覆盖重要产物。

这是一份可复用的工程检查清单，不绑定特定业务。无论你用它做内容生成、客服分流还是数据分析，都建议保留这几个维度。

7.2 学习环境和生产环境的差异

学习环境可以用 Mock 模型、单机文件、同步模式跑通。生产环境至少要补上三块能力。

第一是模型接入层。Mock 模型换成兼容 OpenAI 接口的真实模型或本地模型后，要加入重试、退避、Token 计费和日志采集。模型名称、温度、最大输出长度等参数不能散落在代码里。

第二是消息队列与异步执行。演示项目中角色之间直接返回值即可，生产环境建议引入任务队列，让角色可以并行处理不同 task_id 的任务。这样 Coordinator 在等待 Executor 时，整个进程不会被阻塞。

第三是可观测性。生产环境建议在每条消息上记录耗时、Token 数、成本、重试次数。出现质量问题时，能快速定位是哪个环节引入的错误。

7.3 如何扩展成真正的“一人公司”：任务队列、人工审批、模型切换

一人公司模型的价值不在于“无人工介入”，而在于把重复环节自动化、把关键决策保留给人。扩展时建议按这个顺序推进。

先加入任务队列。用户需求进入系统后，不必立即由一个进程同步执行，而是写入队列，由 Worker 异步消费。这样可以处理多个任务，也能在系统崩溃后从队列恢复未完成任务。

再加入人工审批节点。在发布文章、发送对外邮件、执行代码修改等操作前，让 Ops Guard 发送审批消息，由人工确认后才继续。Role.approval字段已经预留了这个开关。

最后实现模型切换。不同子任务使用不同模型是个不错的实践。简单任务用轻量模型，复杂推理用更强模型。这个能力只要在config.yaml的llm配置里扩展一个model_map字段即可实现。

7.4 新手的下一步练习建议

如果你刚接触多角色 Agent 架构，不建议一开始就去研究复杂编排框架。可以先按下面的路径练习：

第一步，把本文的最小项目复制到本地，用 Mock 模型跑通五角色全链路。

第二步，给 Executor 增加一个真实技能，比如访问某个公开 API 获取数据，观察 Skill 层的隔离是否可靠。

第三步，给 Quality Gate 增加一条业务相关的校验规则，比如“必须包含指定格式的结论段落”，观察反馈消息是否能让 Executor 自动修订。

第四步，把 Coordinator 的plan()方法从规则版替换成模型决策版，让 LLM 根据用户需求动态拆分子任务。

第五步，再决定是否引入消息队列、向量数据库和异步任务调度。过早引入外部依赖会掩盖架构本身的问题。

Hermes Agent Team 五角色架构的核心价值，是把“一人公司”的职能模型固化成可执行、可观测、可替换的工程系统。v3.1 版本中，Skill 独立、记忆隔离、角色薄化这三个原则，值得在每一个多 Agent 项目里沿用。后续迭代时，优先改进 Skill 层和记忆层，而不是继续往角色的系统提示词里堆指令。