Hermes Agent 源码解析
第 11 讲:ACP 协议与子代理委托——编辑器集成、Agent-to-Agent 通信与任务分发
基于 Hermes Agent v0.16.0 源码 · 2026-06-24
一、ACP:让 Agent 成为编辑器的"一等公民"
前十讲我们覆盖了 Hermes 从核心架构到多模态处理的全链路。这一讲我们进入 Hermes 的ACP(Agent Communication Protocol)适配器与子代理委托系统——两个让 Hermes 从"独立终端工具"进化为"可嵌入编辑器的智能编程伙伴"的关键能力。
ACP 是 Anthropic 提出的 Agent Client Protocol,定义了编辑器与 Agent 之间的标准化通信接口。Hermes 通过 ACP 适配器实现了与 Zed、VS Code、JetBrains 等编辑器的无缝集成。而子代理委托系统则让 Hermes 能够自主拆分复杂任务,将子任务分发给独立的子代理实例并行处理。
📦 本讲核心文件
acp_adapter/server.py(2059 行)— ACP Agent 服务器主模块
acp_adapter/session.py(623 行)— ACP 会话管理器
acp_adapter/events.py(279 行)— AIAgent 事件到 ACP 通知的桥接
acp_adapter/tools.py(1291 行)— 工具调用映射与 ACP 内容构建
acp_adapter/permissions.py(168 行)— 权限审批桥接
acp_adapter/auth.py(79 行)— ACP 认证方法检测
acp_adapter/provenance.py(127 行)— 会话溯源元数据
acp_adapter/entry.py(266 行)— ACP 入口点
agent/copilot_acp_client.py(686 行)— Copilot ACP 客户端
tools/delegate_tool.py(2956 行)— 子代理委托引擎
acp_registry/agent.json — ACP 注册表声明
二、ACP 协议概览:编辑器和 Agent 之间的"USB-C"
ACP(Agent Client Protocol)由 Anthropic 提出,是一个标准化的Agent-客户端通信协议。它的核心思想类似于 MCP(Model Context Protocol),但面向的是编辑器 ↔ Agent这个场景,而非 Agent ↔ 外部工具。
有了 ACP,编辑器不需要为每个 Agent 写定制集成——只要实现 ACP 客户端,就能对接任何 ACP Agent。Hermes 通过 acp_adapter/ 模块实现了完整的 ACP Agent 服务端。
ACP 协议的核心交互流程:
+----------------+ JSON-RPC +------------------+
| Editor Client | <-- stdio/HTTP -->| Hermes ACP |
| (Zed/VSCode) | | Agent Server |
+----------------+ +------------------+
| |
| initialize() | 握手:协议版本、能力协商
| session/new() | 创建会话,返回 sessionId
| session/prompt() | 发送用户请求
| session/update (streaming) | 实时流式推送
| |
| session/reset() | 重置会话
| session/fork() | 分支会话
| session/terminate() | 终止会话
| list_sessions() | 列出活跃会话
所有通信基于 JSON-RPC 2.0 协议,通过 stdio(标准输入输出)或 HTTP 传输。Hermes 的 ACP 适配器同时支持两种模式。
三、Hermes ACP 服务器架构
Hermes 的 ACP 服务器实现位于 HermesACPAgent 类(acp_adapter/server.py:446),继承自 acp.Agent。整个 ACP 适配层由以下组件构成:
核心组件矩阵
| 模块 | 职责 | 关键类/函数 |
|---|---|---|
| server.py | ACP Agent 服务端核心 | HermesACPAgent |
| session.py | 会话生命周期管理 | SessionManager, SessionState |
| events.py | 事件桥接(AIAgent → ACP) | make_tool_progress_cb 等 |
| tools.py | 工具调用映射与内容构建 | TOOL_KIND_MAP, build_tool_* |
| permissions.py | 危险命令审批桥接 | make_approval_callback |
| auth.py | 认证方法检测与注册 | build_auth_methods() |
| provenance.py | 会话溯源元数据 | build_session_provenance() |
| entry.py | CLI 入口与环境初始化 | main() |
3.1 ACP 入口与启动流程
用户通过 hermes acp 或 hermes-acp 命令启动 ACP 服务器。入口点 acp_adapter/entry.py 做了三件事:
1. 加载 ~/.hermes/.env 环境变量
2. 配置日志输出到 stderr(保留 stdout 给 ACP JSON-RPC 传输)
3. 创建 HermesACPAgent 实例并启动 ACP 服务器
# acp_adapter/entry.py — 启动流程
def main():
_load_env() # 加载 .env
_setup_logging() # 日志 → stderr
agent = HermesACPAgent() # 创建 ACP Agent
acp.run_agent(agent, ...) # 启动 ACP 服务器
关键设计:日志必须走 stderr,因为 ACP 协议约定 stdout 用于 JSON-RPC 帧传输。如果 stdout 被日志污染,编辑器客户端会解析失败。
3.2 HermesACPAgent:ACP Agent 服务端核心
HermesACPAgent 是 ACP 协议的核心实现,它包装了 Hermes 的 AIAgent 实例,将 ACP 协议消息翻译为 Hermes 内部调用。
该类实现了 ACP 协议定义的所有核心方法:
class HermesACPAgent(acp.Agent):
# 能力声明
def capabilities(self) -> AgentCapabilities: ...
# 会话管理
def create_session(self, ...) -> NewSessionResponse: ...
def fork_session(self, ...) -> ForkSessionResponse: ...
def reset_session(self, ...) -> None: ...
def terminate_session(self, ...) -> None: ...
def list_sessions(self, ...) -> ListSessionsResponse: ...
def load_session(self, ...) -> LoadSessionResponse: ...
def resume_session(self, ...) -> ResumeSessionResponse: ...
# 核心交互
def prompt(self, ...) -> PromptResponse: ...
# 配置
def set_session_model(self, ...) -> SetSessionModelResponse: ...
def set_session_mode(self, ...) -> SetSessionModeResponse: ...
# 资源
def read_resource(self, ...) -> ReadResourceResponse: ...
# 命令
def available_commands(self, ...) -> AvailableCommandsUpdate: ...
def run_command(self, ...) -> RunCommandResponse: ...
每个方法都通过 SessionManager 管理对应的 Hermes 会话,将 ACP 的请求参数映射为 Hermes 的内部参数。
四、SessionManager:会话管理引擎
SessionManager(acp_adapter/session.py:186)是 ACP 会话的大脑。它管理每个 ACP 会话对应的 Hermes AIAgent 实例,并负责会话的持久化。
4.1 SessionState:会话状态结构
@dataclass
class SessionState:
session_id: str # 唯一会话标识
agent: Any # AIAgent 实例
cwd: str = "." # 工作目录
model: str = "" # 当前模型
history: List[Dict] = field(default_factory=list)
cancel_event: Any = None # 取消事件
is_running: bool = False # 是否正在执行
queued_prompts: List[str] = field(default_factory=list)
runtime_lock: Any = field(default_factory=Lock)
current_prompt_text: str = "" # 当前执行的提示词
interrupted_prompt_text: str = "" # 被中断的提示词
4.2 会话持久化
SessionManager 将每个会话持久化到 ~/.hermes/state.db(SessionDB)。这意味着:
🔹 编辑器和 Hermes ACP 进程重启后,历史会话自动恢复
🔹 session_search 工具可以搜索 ACP 会话历史
🔹 list_sessions() 同时查询内存和数据库
# 持久化路径
def _persist(self, state: SessionState):
db = self._get_db()
db.save_session(
session_id=state.session_id,
source="acp", # 标记来源为 ACP
history=state.history,
model=state.model,
cwd=state.cwd,
)
4.3 会话分支(Fork)
fork_session() 方法支持从现有会话创建分支——深度复制历史到新的 AIAgent 实例。这在编辑器的"多标签"场景中非常有用:
def fork_session(self, session_id: str, cwd: str = ".") -> Optional[SessionState]:
original = self.get_session(session_id)
new_id = str(uuid.uuid4())
state = SessionState(
session_id=new_id,
agent=self._make_agent(session_id=new_id, cwd=cwd),
history=copy.deepcopy(original.history), # 深度复制历史
cancel_event=threading.Event(),
)
self._sessions[new_id] = state
self._persist(state)
五、事件桥接:AIAgent → ACP 通知管道
acp_adapter/events.py 实现了 AIAgent 事件到 ACP 会话更新的桥接。由于 AIAgent 运行在工作线程中而 ACP 事件循环在主线程,这里使用了 asyncio.run_coroutine_threadsafe() 进行跨线程通信。
5.1 事件类型映射
AIAgent 内部事件被翻译为 ACP 的 session_update() 通知:
| AIAgent 事件 | ACP 更新类型 | 编辑器展示 |
|---|---|---|
| tool.started | ToolCallStart | 工具调用开始(进度指示器) |
| tool.completed | ToolCallComplete | 工具结果展示 |
| reasoning.available | AgentThoughtChunk | Agent 思考过程 |
| _thinking | AgentThoughtChunk | 内部思考标记 |
| todo 工具结果 | AgentPlanUpdate | 任务面板(Zed 原生渲染) |
| 消息完成 | AgentMessageChunk | 流式文本输出 |
5.2 Todo → Plan 原生映射
一个精妙的设计是 Hermes 的 todo 工具结果被翻译为 ACP 的 AgentPlanUpdate。Zed 编辑器原生渲染 ACP plan 更新为任务面板:
def _build_plan_update_from_todo_result(result):
# Hermes todo 工具返回 JSON: {"todos": [...]}
status_map = {
"pending": "pending",
"in_progress": "in_progress",
"completed": "completed",
"cancelled": "completed", # ACP plan 无 cancelled 状态
}
entries = [
PlanEntry(content=item["content"], status=status_map[status])
for item in data["todos"]
]
return AgentPlanUpdate(session_update="plan", entries=entries)
这意味着当 Hermes Agent 在 ACP 模式下工作时,编辑器可以看到实时的任务进度面板——就像 Claude Desktop 或 Codex CLI 在编辑器中的体验。
六、工具映射与权限审批
6.1 工具类型映射
acp_adapter/tools.py 定义了 Hermes 工具到 ACP ToolKind 的映射表。ACP 的 ToolKind 包括 read、edit、execute、search、fetch、think、other:
TOOL_KIND_MAP = {
# 文件操作
"read_file": "read",
"write_file": "edit",
"patch": "edit",
"search_files": "search",
# 终端/执行
"terminal": "execute",
"process": "execute",
"execute_code": "execute",
# 会话/元工具
"todo": "other",
"skill_view": "read",
# 网络/浏览器
"web_search": "fetch",
"browser_navigate": "fetch",
"browser_click": "execute",
# Agent 内部
"delegate_task": "execute",
"vision_analyze": "read",
# 思考/元
"_thinking": "think",
}
6.2 危险命令审批桥接
acp_adapter/permissions.py 将 Hermes 的危险命令审批机制桥接到 ACP 的 request_permission 机制。当 Agent 尝试执行危险命令时:
# ACP 权限选项
options = [
PermissionOption("allow_once", "allow_once", "Allow once"),
PermissionOption("allow_session", "allow_always", "Allow for session"),
PermissionOption("allow_always", "allow_always", "Allow always"),
PermissionOption("deny", "reject_once", "Deny"),
]
# 映射回 Hermes 审批字符串
_OPTION_ID_TO_HERMES = {
"allow_once": "once",
"allow_session": "session",
"allow_always": "always",
"deny": "deny",
}
编辑器(如 Zed)会在 UI 中显示审批弹窗,用户选择后结果通过 ACP 协议回传给 Hermes。
6.3 编辑审批模式
Hermes ACP 支持三种编辑审批模式,通过 ACP 的 SessionMode 机制暴露给编辑器:
| 模式 ID | 名称 | 策略 |
|---|---|---|
| default | Default | 每次编辑前询问 |
| accept_edits | Accept Edits | 自动允许工作区和 /tmp 编辑 |
| dont_ask | Don't Ask | 会话内自动允许所有编辑 |
七、Copilot ACP 客户端:反向集成
除了作为 ACP 服务器,Hermes 还实现了 CopilotACPClient(agent/copilot_acp_client.py),可以将 GitHub Copilot 的 ACP 模式作为后端使用。
这个客户端实现了 OpenAI 兼容的接口,将 Hermes 的请求转发到 copilot --acp --stdio 进程:
class CopilotACPClient:
"""Minimal OpenAI-client-compatible facade for Copilot ACP."""
def _run_prompt(self, prompt_text, *, timeout_seconds):
# 1. 启动 copilot --acp --stdio 子进程
proc = subprocess.Popen(
[self._acp_command] + self._acp_args,
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
text=True,
)
# 2. JSON-RPC 握手
_request("initialize", {"protocolVersion": 1, ...})
# 3. 创建会话
session = _request("session/new", {"cwd": self._acp_cwd})
# 4. 发送提示词
_request("session/prompt", {
"sessionId": session_id,
"prompt": [{"type": "text", "text": prompt_text}],
})
# 5. 收集流式响应
return "".join(text_parts), "".join(reasoning_parts)
关键设计点:
🔹 工具调用解析:Copilot ACP 返回的是纯文本,需要正则提取 ```{...}``` 工具调用块
🔹 文件系统安全:ACP 服务器请求的文件读写操作被限制在工作目录内(_ensure_path_within_cwd)
🔹 权限拒绝:权限请求默认被拒绝(_permission_denied),防止 Copilot 执行危险操作
八、子代理委托系统:任务分解与并行执行
tools/delegate_tool.py(2956 行)是 Hermes 最复杂的工具之一。它实现了完整的子代理委托架构——父 Agent 可以将任务分发给独立的子代理实例,每个子代理拥有隔离的上下文、受限的工具集和独立的工作线程。
8.1 子代理隔离架构
每个子代理获得:
🔹 独立会话:全新的对话历史(不继承父会话)
🔹 独立 task_id:自己的终端会话和文件操作缓存
🔹 受限工具集:可配置的工具集,始终排除危险工具
🔹 聚焦系统提示:基于委托目标和上下文构建
父 Agent 的上下文中只看到委托调用和摘要结果,永远不会看到子代理的中间工具调用或推理过程。
8.2 工具黑名单
子代理被永久禁止使用以下工具:
DELEGATE_BLOCKED_TOOLS = frozenset([
"delegate_task", # 禁止递归委托(默认)
"clarify", # 禁止用户交互
"memory", # 禁止写入共享 MEMORY.md
"send_message", # 禁止跨平台副作用
"execute_code", # 子代理应逐步推理而非写脚本
])
8.3 角色系统:Leaf vs Orchestrator
子代理支持两种角色:
| 角色 | 能力 | 适用场景 |
|---|---|---|
| leaf(默认) | 执行具体任务,不可再委托 | 代码修改、文件分析、测试执行 |
| orchestrator | 可以进一步委托子任务 | 大型重构、多模块并行开发 |
深度限制:MAX_DEPTH = 1(默认扁平结构:父 → 子)。可通过 delegation.max_spawn_depth 配置增加深度。
8.4 并发控制
子代理在 ThreadPoolExecutor 中并行运行:
_DEFAULT_MAX_CONCURRENT_CHILDREN = 3 # 默认最多 3 个并行子代理
# 配置优先级: config.yaml > 环境变量 > 默认值
def _get_max_concurrent_children():
cfg = _load_config()
val = cfg.get("max_concurrent_children")
if val is not None:
return max(1, int(val))
# ... 环境变量和默认值
8.5 审批回调隔离
子代理运行在工作线程中,不能继承父线程的交互式审批回调(否则会 input() 死锁)。解决方案是为每个子代理线程安装非交互式回调:
# 默认:自动拒绝(安全)
def _subagent_auto_deny(command, description, **kwargs):
logger.warning("Subagent auto-denied: %s (%s)", command, description)
return "deny"
# 可选:自动批准(YOLO 模式,需显式配置)
def _subagent_auto_approve(command, description, **kwargs):
logger.warning("Subagent auto-approved: %s (%s)", command, description)
return "once"
# 通过 ThreadPoolExecutor initializer 注入
executor = ThreadPoolExecutor(
initializer=_set_subagent_approval_cb,
initargs=(callback,),
)
8.6 可观测性
子代理系统提供了完整的可观测性支持:
🔹 活跃子代理注册表:_active_subagents 跟踪所有运行中的子代理
🔹 进度事件:DelegateEvent 枚举(task_spawned, tool_started, tool_completed 等)
🔹 暂停/恢复:set_spawn_paused() 全局暂停新子代理生成
🔹 中断:interrupt_subagent() 请求子代理在下次迭代边界停止
🔹 输出提取:_extract_output_tail() 提取子代理最后 N 个工具调用结果
九、ACP 注册表与认证
9.1 ACP 注册表声明
acp_registry/agent.json 是 Hermes 在 ACP 生态中的"名片":
{
"id": "hermes-agent",
"name": "Hermes Agent",
"version": "0.16.0",
"description": "Self-improving open-source AI agent by Nous Research
with ACP editor integration, persistent memory, skills,
and rich tool support.",
"repository": "https://github.com/NousResearch/hermes-agent",
"website": "https://hermes-agent.nousresearch.com/docs/user-guide/features/acp",
"distribution": {
"uvx": {
"package": "hermes-agent[acp]==0.16.0",
"args": ["hermes-acp"]
}
}
}
编辑器通过 ACP 注册表发现 Hermes Agent,用户只需在编辑器中搜索 "Hermes Agent" 即可安装。
9.2 认证方法
acp_adapter/auth.py 检测并广告 Hermes 的认证方法:
def build_auth_methods():
methods = []
provider = detect_provider()
if provider:
methods.append(AuthMethodAgent(
id=provider,
name=f"{provider} runtime credentials",
description=f"Authenticate using {provider} credentials.",
))
# 始终广告终端设置方法(首次配置引导)
methods.append(TerminalAuthMethod(
id="hermes-setup",
name="Configure Hermes provider",
type="terminal",
args=["--setup"],
))
return methods
如果用户尚未配置 Hermes,编辑器会引导用户运行 hermes setup 完成首次配置。
十、会话溯源与压缩链
acp_adapter/provenance.py 实现了会话溯源元数据。当上下文压缩触发会话轮换时,ACP 客户端可以看到完整的会话谱系:
def build_session_provenance(db, acp_session_id,
current_hermes_session_id, ...):
# 回溯父会话链,找到谱系根节点
root_id = current_hermes_session_id
compression_depth = 0
for _ in range(_MAX_WALK):
prow = db.get_session(cursor_parent)
root_id = cursor_parent
if prow.get("end_reason") == "compression":
compression_depth += 1
return {
"acpSessionId": acp_session_id,
"currentHermesSessionId": current_hermes_session_id,
"rootSessionId": root_id,
"compressionDepth": compression_depth,
"isContinuation": is_continuation,
}
这些信息通过 ACP 的 _meta.hermes 字段传递给客户端,让编辑器知道当前会话是否经历了压缩轮换。
十一、架构总结:ACP + 子代理 = 完整的 Agent 生态
Hermes 的 ACP 适配器和子代理委托系统共同构成了一个完整的 Agent 生态系统:
+--------------------------------------------------+
| Editor (Zed/VSCode) |
| |
| +----------------+ ACP JSON-RPC +--------+|
| | ACP Client | <----------------->| ACP ||
| | (Zed ACP | stdio/HTTP | Server ||
| | extension) | | (Hermes)||
| +----------------+ +--------+|
| |
| Session Panel: |
| - Task/Plan progress (todo -> AgentPlanUpdate) |
| - Tool call status (read/edit/execute) |
| - Thinking process (reasoning chunks) |
| - Permission prompts (dangerous commands) |
+--------------------------------------------------+
|
v
+--------------------------------------------------+
| Hermes AIAgent Core |
| |
| +--------------------------------------------+ |
| | Conversation Loop | |
| | - System prompt + history | |
| | - Tool selection & execution | |
| | - Context compression | |
| +--------------------------------------------+ |
| |
| +--------------------------------------------+ |
| | Delegate Tool (sub-agent spawning) | |
| | - ThreadPoolExecutor (max 3 concurrent) | |
| | - Isolated context per child | |
| | - Restricted toolset | |
| | - Leaf / Orchestrator roles | |
| +--------------------------------------------+ |
+--------------------------------------------------+
🔑 关键设计原则
🔹 协议标准化:ACP 让 Hermes 成为编辑器的"一等公民",无需定制集成
🔹 会话持久化:ACP 会话与 CLI 会话共享 SessionDB,重启后自动恢复
🔹 事件桥接:AIAgent 内部事件实时映射为 ACP 通知,编辑器获得原生体验
🔹 安全隔离:子代理有独立上下文、受限工具集、自动审批回调
🔹 可观测性:完整的子代理注册表、进度事件、暂停/恢复、中断机制
十二、下一讲预告
下一讲(第 12 讲)我们将深入 Hermes 的CLI 交互架构与 Session 持久化——TUI 界面、会话数据库、命令行解析和交互式对话循环的实现细节。这将串联起 Hermes 作为独立终端应用的完整用户体验。
夜雨聆风