架构深入解析¶
本页提供 Hermes Agent 内部工作原理的详细技术说明。
系统概览¶
Hermes Agent 采用模块化架构，实现关注点分离：
┌─────────────────────────────────────────────────────────────┐
│ 系统组件 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 入口点 │ │
│ │ cli.py │ gateway/ │ acp_server.py │ cron/ │ │
│ └─────────────────────┬───────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 核心 Agent │ │
│ │ run_agent.py │ agent/ │ model_tools.py │ │
│ └─────────────────────┬───────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 工具和技能 │ │
│ │ tools/ │ skills/ │ toolsets.py │ │
│ └─────────────────────┬───────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 存储层 │ │
│ │ hermes_state.py │ memories/ │ sessions/ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
核心组件¶
1. AIAgent (run_agent.py)¶
AIAgent 类是 Hermes 的核心：
class AIAgent:
def __init__(self, config, tools, memory):
self.config = config
self.tools = tools
self.memory = memory
self.messages = []
def run_conversation(self, user_input):
"""主对话循环"""
# 1. 构建系统提示词
system_prompt = self.build_system_prompt()
# 2. 添加用户消息
self.messages.append({"role": "user", "content": user_input})
# 3. 循环直到完成
while self.iterations < self.max_turns:
# 调用 LLM
response = self.call_llm(system_prompt, self.messages)
# 处理响应
if response.tool_calls:
# 执行工具
for tool_call in response.tool_calls:
result = self.execute_tool(tool_call)
self.messages.append({
"role": "tool",
"content": result
})
else:
# 返回文本响应
return response.content
2. 提示词构建器 (agent/prompt.py)¶
提示词构建器负责构建系统提示词：
def build_system_prompt(self):
"""构建完整的系统提示词"""
sections = []
# 1. 基础人设
sections.append(self.load_persona())
# 2. 注入记忆
sections.append(self.inject_memory())
# 3. 技能上下文
sections.append(self.load_skills())
# 4. 工具 Schema
sections.append(self.get_tool_schemas())
# 5. 上下文文件 (AGENTS.md 等)
sections.append(self.load_context_files())
return "\n\n".join(sections)
3. 工具调度 (model_tools.py)¶
工具通过中央注册表进行调度：
# tools/registry.py
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, name, schema, handler, check_fn=None):
"""注册工具"""
self.tools[name] = {
"schema": schema,
"handler": handler,
"check_fn": check_fn
}
def dispatch(self, name, arguments, **kwargs):
"""执行工具"""
tool = self.tools[name]
# 检查要求
if tool["check_fn"] and not tool["check_fn"]():
return {"error": "Requirements not met"}
# 执行处理器
return tool["handler"](arguments, **kwargs)
4. 上下文压缩 (agent/compression.py)¶
当对话变长时，Hermes 会压缩上下文：
def compress_context(self, messages, target_ratio=0.2):
"""压缩对话上下文"""
# 1. 计算当前大小
current_size = self.calculate_token_count(messages)
target_size = int(current_size * target_ratio)
# 2. 保护最近的消息
protected = messages[-self.protect_last_n:]
compressible = messages[:-self.protect_last_n]
# 3. 总结较早的消息
summary = self.summarize_messages(compressible)
# 4. 重建消息
compressed = [
{"role": "system", "content": f"之前的上下文摘要: {summary}"},
*protected
]
return compressed
工具系统架构¶
工具注册¶
每个工具定义在 tools/ 目录中：
# tools/terminal.py
import json
from tools.registry import registry
def terminal_handler(arguments, **kwargs):
"""执行 Shell 命令"""
command = arguments.get("command", "")
# 执行命令
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True
)
return json.dumps({
"output": result.stdout,
"error": result.stderr,
"exit_code": result.returncode
})
registry.register(
name="terminal",
toolset="terminal",
schema={
"name": "terminal",
"description": "执行 Shell 命令",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "要执行的命令"
}
},
"required": ["command"]
}
},
handler=terminal_handler,
check_fn=lambda: True # 始终可用
)
工具集组织¶
工具按工具集分组：
# toolsets.py
_HERMES_CORE_TOOLS = [
"terminal",
"file",
"web",
"browser",
"vision",
"memory",
"skills",
"delegation",
"cronjob",
# ...
]
def get_toolsets(platform="cli"):
"""获取平台可用的工具集"""
return _PLATFORM_TOOLSETS.get(platform, _HERMES_CORE_TOOLS)
记忆系统¶
记忆存储¶
记忆存储在 ~/.hermes/memories/ 中：
记忆注入¶
记忆会注入到每次对话中：
def inject_memory(self):
"""将记忆注入系统提示词"""
sections = []
# 用户档案
user = self.load_user_profile()
if user:
sections.append(f"## 用户档案\n{user}")
# 记忆事实
memory = self.load_memory()
if memory:
sections.append(f"## 记忆\n{memory}")
return "\n\n".join(sections)
技能系统¶
技能加载¶
技能从 ~/.hermes/skills/ 加载：
def load_skills(self, skill_names=None):
"""将技能加载到上下文中"""
skills = []
for skill_dir in self.skills_dir.iterdir():
skill_file = skill_dir / "SKILL.md"
if skill_file.exists():
# 解析 frontmatter
content = skill_file.read_text()
frontmatter, body = parse_frontmatter(content)
skills.append({
"name": frontmatter["name"],
"description": frontmatter["description"],
"content": body
})
return skills
技能创建¶
当 Hermes 学到新东西时会创建技能：
def create_skill(self, name, description, content):
"""创建新技能"""
skill_dir = self.skills_dir / name
skill_dir.mkdir(parents=True, exist_ok=True)
skill_file = skill_dir / "SKILL.md"
skill_file.write_text(f"""---
name: {name}
description: {description}
---
{content}
""")
网关架构¶
平台适配器¶
每个平台都有一个适配器：
# gateway/platforms/telegram.py
class TelegramAdapter:
def __init__(self, config):
self.bot = telegram.Bot(token=config["token"])
async def handle_message(self, update, context):
"""处理传入的消息"""
user_id = update.effective_user.id
message = update.message.text
# 通过 agent 处理
response = await self.agent.process(message)
# 发送响应
await self.bot.send_message(
chat_id=update.effective_chat.id,
text=response
)
消息路由¶
消息通过网关路由：
class Gateway:
def __init__(self):
self.adapters = {}
self.agent = AIAgent()
def register_adapter(self, name, adapter):
"""注册平台适配器"""
self.adapters[name] = adapter
async def route_message(self, platform, message):
"""将消息路由到 agent"""
response = await self.agent.process(message)
# 通过平台发回
await self.adapters[platform].send_response(response)
凭证池¶
密钥轮换¶
多个 API 密钥自动轮换：
class CredentialPool:
def __init__(self, provider, keys):
self.provider = provider
self.keys = keys
self.current_index = 0
def get_next_key(self):
"""获取下一个可用密钥"""
key = self.keys[self.current_index]
self.current_index = (self.current_index + 1) % len(self.keys)
return key
def mark_exhausted(self, key):
"""标记密钥已达速率限制"""
# 切换到下一个密钥
pass
配置系统¶
配置结构¶
配置存储在 ~/.hermes/config.yaml 中：
model:
default: claude-sonnet-4
provider: anthropic
api_key: ${ANTHROPIC_API_KEY}
agent:
max_turns: 90
tool_use_enforcement: auto
terminal:
backend: local
timeout: 180
compression:
enabled: true
threshold: 0.5
target_ratio: 0.2
memory:
memory_enabled: true
user_profile_enabled: true
环境变量¶
密钥存储在 ~/.hermes/.env 中：
数据流¶
请求处理¶
用户输入
│
▼
┌─────────────────┐
│ 平台适配器 │ (Telegram, Discord 等)
└────────┬────────┘
│
▼
┌─────────────────┐
│ 网关 │ (消息路由)
└────────┬────────┘
│
▼
┌─────────────────┐
│ AIAgent │ (核心 agent 循环)
└────────┬────────┘
│
▼
┌─────────────────┐
│ 提示词构建器 │ (系统提示词构建)
└────────┬────────┘
│
▼
┌─────────────────┐
│ LLM API │ (模型推理)
└────────┬────────┘
│
▼
┌─────────────────┐
│ 工具调度 │ (执行工具)
└────────┬────────┘
│
▼
┌─────────────────┐
│ 响应生成 │ (生成响应)
└────────┬────────┘
│
▼
┌─────────────────┐
│ 平台适配器 │ (发送响应)
└─────────────────┘
性能考虑¶
Token 优化¶
- **上下文压缩**减少 80% 的 token 使用
- **技能加载**是懒加载（仅在需要时）
- **工具 Schema**在首次加载后缓存
内存管理¶
- **会话记录**存储在 SQLite 中
- **记忆事实**是紧凑的 JSON
- **技能**按需加载
并发¶
- **网关**并发处理多个平台
- **委派**生成子代理进行并行工作
- **定时任务**在独立进程中运行
安全模型¶
命令审批¶
def should_approve(command):
"""检查命令是否需要审批"""
dangerous_patterns = [
r'rm\s+-rf',
r'git\s+reset\s+--hard',
r'dd\s+if=',
r'mkfs',
]
for pattern in dangerous_patterns:
if re.search(pattern, command):
return True
return False
密钥脱敏¶
def redact_secrets(text):
"""从文本中移除密钥"""
patterns = [
r'sk-[a-zA-Z0-9]{48}', # API 密钥
r'ghp_[a-zA-Z0-9]{36}', # GitHub token
r'AKIA[0-9A-Z]{16}', # AWS 密钥
]
for pattern in patterns:
text = re.sub(pattern, '[REDACTED]', text)
return text
