Hermes Agent — agent 模块超深度专业级代码分析

分析范围: hermes-agent/agent/
模块总量: 34个Python文件，约700K字符
架构风格: 从3600行单体 run_agent.py 提取的模块化微内核
分析日期: 2026-04-19
图表风格: Dark Terminal (fireworks-tech-graph Style 2)

---

📐 架构总览图

---

一、模块定位

1.1 业务职责与功能定位

Hermes Agent 的 agent/ 模块是整个AI智能体系统的内核层（Kernel Layer）。它不包含上层业务编排逻辑（run_agent.py中的AIAgent类），而是提供智能体运行所需的全部基础设施：

功能域	核心模块	业务价值
多LLM提供商接入	anthropic_adapter, bedrock_adapter, gemini_cloudcode_adapter, copilot_acp_client, auxiliary_client	统一OpenAI兼容接口，屏蔽6+后端差异
认证与凭证管理	credential_pool, google_oauth, google_code_assist	多密钥故障转移、OAuth PKCE、自动刷新
上下文与记忆	context_compressor, context_engine, context_references, memory_manager, memory_provider, subdirectory_hints, prompt_builder, prompt_caching	长对话压缩、持久记忆、引用解析、提示装配
模型元数据与定价	model_metadata, models_dev, usage_pricing, smart_model_routing	成本估算、速率追踪、智能路由
速率限制	rate_limit_tracker, nous_rate_guard	跨会话速率保护、防止重试放大
技能系统	skill_utils, skill_commands	技能发现、配置解析、斜杠命令路由
错误分类与重试	error_classifier, retry_utils	结构化API错误分类、去相关指数退避
展示与洞察	display, insights, title_generator	CLI界面、会话分析、自动标题
安全与工具	redact, trajectory, manual_compression_feedback	敏感信息脱敏、轨迹保存、压缩反馈

1.2 在系统中的位置

┌─────────────────────────────────────────────────┐
│                  CLI / Gateway                    │  用户接入层
├─────────────────────────────────────────────────┤
│              AIAgent (run_agent.py)              │  编排层（主循环）
├─────────────────────────────────────────────────┤
│              ★ agent/ 模块（本次分析范围）★         │  内核基础设施层
├─────────────────────────────────────────────────┤
│         tools/  hermes_cli/  gateway/            │  工具/配置/网关层
└─────────────────────────────────────────────────┘

agent/ 被 run_agent.py 的 AIAgent 类单向依赖。它不反向引用上层模块（除了 hermes_constants 和少量工具模块的类型导入），保持清晰的依赖倒置边界。

1.3 核心业务价值

1. 多提供商抽象：一个 chat.completions.create() 接口统一6种后端，实现零成本切换
2. 凭证故障转移：同一提供商支持多密钥轮换，单密钥失效不影响服务连续性
3. 上下文窗口管理：自动压缩长对话，在任意模型上下文限制内保持对话连续
4. 成本可观测：实时定价估算、速率追踪，防止无意识消耗API额度
5. 安全合规：30+种密钥模式的自动脱敏，防止泄露到日志/网关

---

二、模块整体结构

2.1 类结构与接口定义

抽象基类（ABC）

ABC	位置	实现者	设计意图
ContextEngine	context_engine.py	ContextCompressor（默认）	可插拔的上下文压缩引擎，第三方可通过plugin替换
MemoryProvider	memory_provider.py	BuiltinMemoryProvider	可插拔的记忆后端，支持Honcho/Mem0等外部provider

核心数据类

Dataclass	位置	用途
CredentialEntry	credential_pool.py	单个API凭证（key, base_url, label, provider, expires_at, source）
RateLimitBucket / RateLimitState	rate_limit_tracker.py	速率限制窗口状态（limit, remaining, reset_seconds）
CanonicalUsage	usage_pricing.py	归一化token使用量（input, output, cache_read, cache_write, reasoning）
BillingRoute	usage_pricing.py	计费路由（provider, model, billing_mode）
PricingEntry	usage_pricing.py	定价条目（每百万token成本）
CostResult	usage_pricing.py	成本计算结果（amount_usd, status, source）
ModelInfo / ProviderInfo	models_dev.py	模型/提供商元数据
GoogleCredentials	google_oauth.py	Google OAuth凭证（access, refresh, expires, email, project_id）
CodeAssistProjectInfo	google_code_assist.py	Code Assist项目信息（tier, project）
FailoverReason	error_classifier.py	API错误分类枚举（rate_limit, auth, overload, context_limit, unknown）

依赖注入关系

核心依赖链：

AIAgent → prompt_builder → skill_utils → hermes_constants
       → credential_pool → hermes_cli.auth → hermes_constants
       → context_compressor → auxiliary_client → credential_pool
       → anthropic_adapter / bedrock_adapter / gemini_cloudcode_adapter
       → error_classifier → retry_utils
       → usage_pricing → model_metadata → models_dev
       → memory_manager → memory_provider

2.2 核心方法清单

anthropic_adapter.py

方法	作用
translate_to_anthropic()	OpenAI messages[] → Anthropic messages格式
translate_to_openai()	Anthropic响应 → OpenAI兼容响应
apply_anthropic_cache_control()	注入4个cache_control断点（system_and_3策略）
create_client()	创建带认证的Anthropic客户端

bedrock_adapter.py

方法	作用
translate_to_bedrock()	OpenAI → AWS Bedrock Converse API格式
translate_from_bedrock()	Bedrock响应 → OpenAI兼容格式
create_bedrock_client()	使用AWS凭证链创建boto3客户端

context_compressor.py

方法	作用
compress()	执行上下文压缩：保护头尾，摘要中间轮次
_build_summary_prompt()	构建摘要提示（结构化模板：Resolved/Pending/Remaining Work）
_merge_summary()	将摘要合并回消息列表

credential_pool.py

方法	作用
resolve_client()	选择最佳凭证 → 创建OpenAI兼容客户端
rotate_credential()	认证失败时轮换到下一个凭证
refresh_codex_token()	自动刷新Codex OAuth access token
add_credential()	添加新凭证到池

prompt_builder.py

方法	作用
_build_system_prompt()	组装完整系统提示（身份+平台+技能+上下文+记忆）
_scan_context_content()	安全扫描上下文文件内容
_load_memory_files()	加载MEMORY.md/USER.md

2.3 内部调用关系

请求路径：

1. AIAgent 调用 prompt_builder._build_system_prompt() 组装提示
2. context_references.resolve() 处理用户消息中的@引用
3. 发送到 provider_adapter.chat.completions.create()
4. 响应回来后 rate_limit_tracker.parse_rate_limit_headers() 捕获速率信息
5. usage_pricing.normalize_usage() + estimate_usage_cost() 计算成本
6. 错误时 error_classifier.classify() → 根据分类路由到 retry_utils / credential_pool

记忆路径：

1. memory_manager.search() → BuiltinMemoryProvider.search() → 读取MEMORY.md
2. memory_manager.store() → BuiltinMemoryProvider.store() → 写入MEMORY.md
3. 外部provider可选注册（Honcho/Mem0）

2.4 数据流入流出方式

数据流	入口	出口	格式
用户消息	AIAgent	prompt_builder	List[Dict[str, Any]]（OpenAI messages格式）
API请求	provider_adapter	外部API	HTTP/HTTPS
API响应	外部API	provider_adapter	JSON → OpenAI兼容SimpleNamespace
速率限制头	HTTP响应	rate_limit_tracker	x-ratelimit-* headers → RateLimitState
成本数据	usage_pricing	AIAgent	CostResult(amount_usd, status, source)
凭证	~/.hermes/auth.json	credential_pool	JSON → CredentialEntry
记忆	MEMORY.md / 外部API	memory_manager	Markdown / JSON
轨迹	会话历史	trajectory.py	ShareGPT JSONL

---

---

三、核心业务逻辑深度解析

3.1 anthropic_adapter.py — Anthropic Messages API适配器

完整执行流程

AIAgent调用 → create_client(api_key, base_url) → 返回带适配的OpenAI兼容client
           → client.chat.completions.create(messages, tools, ...)
           → translate_to_anthropic(messages) → Anthropic格式
           → 应用prompt_caching.apply_anthropic_cache_control()
           → HTTP请求到api.anthropic.com
           → translate_to_openai(response) → OpenAI兼容响应

关键设计细节

认证方式三级解析：

# 1. 常规API密钥 (sk-ant-api*) → x-api-key header
# 2. OAuth setup-token (sk-ant-oat*) → Bearer auth + beta header  
# 3. Claude Code凭证 (~/.claude.json) → Bearer auth

消息格式翻译：

• OpenAI的 system 角色消息 → Anthropic的 system 参数（不在messages数组中）
• OpenAI的 tool_calls[].function.arguments (JSON string) → Anthropic的 tool_use content block的 input (parsed dict)
• Anthropic的 tool_use block → OpenAI的 tool_calls[]（需生成call_前缀的id）
• Anthropic的 tool_result block → OpenAI的 tool role message

流式响应处理：

• Anthropic的SSE事件类型：message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop
• 每种事件映射到OpenAI的 ChatCompletionChunk delta格式
• thinking 类型的content block映射到 reasoning_content 字段

3.2 bedrock_adapter.py — AWS Bedrock Converse API适配器

核心翻译逻辑

OpenAI → Bedrock Converse：

• messages[] → messages[]（角色映射：assistant→assistant, user→user, system→system参数）
• tools[].function → tools[].toolSpec（functionDeclarations → toolSpec格式）
• tool_choice → toolConfig.toolChoice（auto→AUTO, required→ANY, function→{name:xxx}）
• temperature/max_tokens/top_p → inferenceConfig 对象

Bedrock → OpenAI：

• output.message.content[].text → message.content
• output.message.content[].toolUse → message.tool_calls[]
• stopReason → finish_reason（end_turn→stop, tool_use→tool_calls, max_tokens→length）

AWS凭证链：优先级为 IAM角色 > SSO > 环境变量 > 实例元数据

流式处理：Bedrock使用event stream（ContentBlockDelta, MessageStopEvent等），逐事件翻译为OpenAI chunk

3.3 gemini_cloudcode_adapter.py — Google Code Assist适配器

三层架构

1. google_oauth.py：PKCE OAuth流程

   • 生成verifier+challenge → 浏览器授权 → code exchange → 保存access+refresh token
   • 刷新逻辑：get_valid_access_token() → 检查过期 → 刷新 → 跨进程去重（_refresh_inflight锁）
   • 存储：~/.hermes/auth/google_oauth.json，0o600权限

2. google_code_assist.py：Code Assist控制面

   • load_code_assist()：探测用户tier + 已分配项目
   • onboard_user()：新用户注册（支持LRO长运行操作轮询）
   • retrieve_user_quota()：获取配额桶数组
   • resolve_project_context()：优先级 env > config > 发现 > 自动注册

3. gemini_cloudcode_adapter.py：请求翻译层

   • OpenAI messages[] → Gemini contents[] + systemInstruction
   • OpenAI tools[] → Gemini tools[].functionDeclarations[]
   • OpenAI tool_calls[] → Gemini functionCall parts（附thoughtSignature跳过验证）
   • Gemini candidates[].content.parts[] → OpenAI choices[0].message
   • 流式：SSE ?alt=sse → 逐事件翻译

关键防护

VPC-SC检测：_is_vpc_sc_violation() 检测 SECURITY_POLICY_VIOLATED，自动降级到standard-tier

MODEL_CAPACITY_EXHAUSTED：特殊429错误，提供重试建议和备用提供商提示

3.4 copilot_acp_client.py — GitHub Copilot ACP适配器

设计思路

将 copilot --acp 命令行工具包装为OpenAI兼容的chat.completions接口。每次请求启动一个短命ACP会话。

执行流程：

1. ACPClient.__init__() → 启动 copilot --acp 子进程
2. chat.completions.create() → 格式化消息为单prompt → 通过stdin发送 → 从stdout收集文本块 → 转换为OpenAI响应

线程模型：

• 子进程stdin/stdout通过 queue.Queue 桥接到生成器
• threading.Thread 读取stdout行，放入队列
• 主线程从队列消费，yield chunk

凭证来源：~/.copilot/sessions/ 下的会话文件

3.5 auxiliary_client.py — 辅助LLM客户端路由器

解析优先级（文本任务auto模式）

1. OpenRouter (OPENROUTER_API_KEY)
2. Nous Portal (~/.hermes/auth.json active provider)
3. Custom endpoint (config.yaml model.base_url + OPENAI_API_KEY)
4. Codex OAuth (Responses API via chatgpt.com, gpt-5.3-codex)
5. Native SDK (openai/anthropic SDK直连)
6. Bedrock (boto3 Converse API)
7. Gemini Code Assist (google_oauth)
8. GitHub Copilot ACP (copilot --acp)

call_llm() 函数

统一入口，解析最佳后端后调用。用于：上下文压缩、会话搜索、网页提取、视觉分析、浏览器视觉、标题生成。

关键特性：

• 自动fallback：第一选择失败时尝试下一个
• 超时控制：每个后端有独立超时配置
• 凭证池集成：自动从credential_pool获取API key

3.6 credential_pool.py — 多凭证故障转移池

数据模型

CredentialEntry:
    api_key: str           # API密钥
    base_url: str          # 端点URL  
    label: str             # 人类可读标签
    provider: str          # 提供商标识
    expires_at: float      # 过期时间（Codex OAuth）
    source: str            # 来源（env/config/cli/oauth）
    scope: List[str]       # 权限范围

核心逻辑

resolve_client()：

1. 从pool中筛选匹配provider的所有凭证
2. 过滤已过期凭证
3. 选择优先级最高的（source优先级：cli > config > env > oauth）
4. 创建OpenAI兼容客户端
5. 缓存客户端实例（key = (provider, model, api_key, base_url)签名）

rotate_credential()：

1. 标记当前凭证为failed
2. 选择下一个可用凭证
3. 重建客户端
4. 如果所有凭证都失败，抛出异常

refresh_codex_token()：

1. 检查expires_at是否接近过期
2. 通过refresh_token获取新的access_token
3. 原子更新auth.json文件
4. 跨线程去重刷新（_refresh_inflight锁）

线程安全

• _pool_lock：保护pool字典的读写
• _client_cache_lock：保护客户端缓存
• _refresh_locks：每个凭证独立的刷新锁
• 原子文件写入：tempfile + os.replace

3.7 context_compressor.py — 上下文压缩器

system_and_3策略

┌─────────────────────┐
│ System Prompt (保留) │  ← 始终保留
├─────────────────────┤
│ 前几轮对话 (保留)     │  ← 保留头部
├─────────────────────┤
│                     │
│  中间轮次 (压缩)     │  ← 使用auxiliary LLM摘要
│                     │
├─────────────────────┤
│ 最近几轮 (保留)      │  ← 保留尾部
├─────────────────────┤
│ 当前用户消息 (保留)  │  ← 始终保留
└─────────────────────┘

压缩流程

1. _should_compress()：检查token估算是否超过模型限制的阈值
2. _compress()：
   a. 分离system消息
   b. 分离尾部N条消息
   c. 将中间消息发送给auxiliary LLM
   d. 使用结构化摘要模板（Resolved Questions / Pending Questions / Remaining Work）
   e. 摘要者前缀：“Do not respond to any questions”（防止摘要者"回答"而非"总结"）
   f. 交接框架：“different assistant”（创建分离感，防止摘要被当作指令）
3. _merge_summary()：将摘要作为单条system消息插入

v2改进（注释中标注）

• 结构化摘要模板（Resolved/Pending问题追踪）
• 摘要者前缀防"回答"（来自OpenCode）
• 交接框架"different assistant"（来自Codex）
• “Remaining Work"替代"Next Steps”（避免被误读为指令）
• 清晰分隔符标识摘要合并位置

3.8 context_engine.py — 可插拔上下文引擎ABC

class ContextEngine(ABC):
    @abstractmethod
    def should_compress(self, messages, model, token_limit) -> bool
    
    @abstractmethod  
    def compress(self, messages, model, token_limit, ...) -> List[Dict]

设计意图：默认实现是ContextCompressor，第三方可通过 plugins/context_engine/<name>/ 目录替换。配置选择：context.engine in config.yaml，默认 "compressor"。

3.9 context_references.py — 用户消息@引用解析

支持的引用类型

语法	解析为	来源
@file:path	文件内容	本地文件系统
@folder:path	目录结构	本地文件系统
@git:ref	Git diff	git show/diff 命令
@url:url	网页内容	HTTP GET + 文本提取
@diff	工作区变更	git diff
@staged	暂存区变更	git diff --staged

正则解析

REFERENCE_PATTERN = re.compile(
    r'(?<![\w/])@(?:(?P<simple>diff|staged)\b|'
    r'(?P<kind>file|folder|git|url):(?P<value>`[^`\n]+`|"[^"\n]+"|\'[^\'\n]+\'|\S+))'
)

安全防护：

• _SENSITIVE_HOME_DIRS：拒绝读取 .ssh, .aws, .gnupg 等敏感目录
• _scan_context_content()：检测并拒绝包含潜在恶意内容的文件
• 路径遍历防护：拒绝 .. 路径

3.10 memory_manager.py — 记忆管理器

双Provider架构

MemoryManager
  ├── BuiltinMemoryProvider (always-on, 不可移除)
  │   ├── search() → 搜索MEMORY.md/USER.md
  │   ├── store()  → 写入MEMORY.md
  │   └── tools    → memory_search, memory_store
  └── ExternalProvider (最多1个, 可插拔)
      ├── search() → 调用外部API
      ├── store()  → 写入外部存储
      └── tools    → 额外的工具schema

约束：只允许1个外部provider，防止工具schema膨胀和记忆后端冲突。

3.11 memory_provider.py — 记忆Provider ABC

class MemoryProvider(ABC):
    @abstractmethod
    def search(self, query, **kwargs) -> List[MemoryResult]
    
    @abstractmethod
    def store(self, content, **kwargs) -> bool
    
    @abstractmethod
    def get_tools(self) -> List[Dict]:  # 返回工具schema
    
    @abstractmethod
    def register(self, memory_manager) -> None

BuiltinMemoryProvider：直接操作 MEMORY.md / USER.md 文件，使用简单的文本搜索（非向量搜索）。

3.12 prompt_builder.py — 系统提示装配器

装配流程

1. 加载SOUL.md / IDENTITY.md → 身份段落
2. 检测平台（CLI/Gateway/Telegram等）→ 平台提示
3. 扫描技能目录 → 构建技能索引段落
4. 加载AGENTS.md / CLAUDE.md / .cursorrules → 上下文文件
5. 加载MEMORY.md → 记忆上下文
6. 组装：身份 + 平台 + 技能 + 上下文 + 记忆 + 临时提示

关键约束：

• 每个段落有最大字符限制（防止上下文溢出）
• 技能描述截断为60字符
• 上下文文件安全扫描（_scan_context_content）
• 线程安全的技能缓存（_skills_cache_lock）

3.13 model_metadata.py — 模型元数据与Token估算

Token估算

def estimate_tokens_rough(text: str, model: str = "") -> int:
    # 简单估算：1 token ≈ 4字符（英文）或 2字符（中文）
    # 比调用tiktoken快100倍，误差在可接受范围内

上下文窗口查询

解析顺序：

1. KNOWN_MODEL_CONTEXTS 硬编码表（最可靠）
2. OpenRouter models API 缓存
3. models.dev API 缓存
4. 本地config.yaml中的user_override

模型提供商识别

PROVIDER_PREFIXES = ["anthropic", "openai", "google", "bedrock", "deepseek", ...]
# "anthropic/claude-3-opus" → provider="anthropic", model="claude-3-opus"
# "claude-3-opus" → 无provider前缀，需要从base_url推断

3.14 models_dev.py — Models.dev社区数据库集成

缓存策略

1. 内存缓存（1小时TTL）
2. 磁盘缓存（~/.hermes/models_dev_cache.json）
3. 网络获取（https://models.dev/api.json，15秒超时）
4. 后台每60分钟自动刷新

数据模型

ModelInfo 包含：id, name, family, reasoning, tool_call, attachment, context_window, max_output, cost_input, cost_output, cost_cache_read, cost_cache_write, knowledge_cutoff, status 等20+字段。

Provider映射

PROVIDER_TO_MODELS_DEV 字典将Hermes内部provider名映射到models.dev ID。如 "kilocode" → "kilo", "gemini" → "google"。

过滤逻辑

• _NOISE_PATTERNS：排除TTS、embedding、预览版、纯图像模型
• _GOOGLE_HIDDEN_MODELS：排除低TPM的Gemma模型和已退役的Gemini版本
• list_agentic_models()：只返回 tool_call=True 且不匹配噪音模式的模型

3.15 usage_pricing.py — 使用量归一化与成本估算

使用量归一化（三种API格式）

def normalize_usage(response_usage, *, provider, api_mode) -> CanonicalUsage:
    # Anthropic: input_tokens / output_tokens / cache_read_input_tokens / cache_creation_input_tokens
    # Codex Responses: input_tokens (含cache) / input_tokens_details.cached_tokens
    # OpenAI Chat: prompt_tokens (含cache) / prompt_tokens_details.cached_tokens
    # → 统一为: input_tokens, output_tokens, cache_read_tokens, cache_write_tokens

关键：OpenAI/Codex的 input_tokens 包含缓存token，需要减去cache_read + cache_write才是纯input。

计费路由

def resolve_billing_route(model, provider, base_url) -> BillingRoute:
    # Codex → subscription_included（订阅制，按token不计费）
    # OpenRouter → official_models_api（从API获取定价）
    # Anthropic/OpenAI → official_docs_snapshot（硬编码官方定价表）
    # Custom/localhost → unknown

定价快照

_OFFICIAL_DOCS_PRICING 硬编码了20+个模型的定价（Anthropic 5个, OpenAI 7个, Google 3个, DeepSeek 2个, Bedrock 7个），每个包含 input/output/cache_read/cache_write 每百万token美元价格。

成本计算

def estimate_usage_cost(model, usage, ...) -> CostResult:
    amount = input * input_price / 1M + output * output_price / 1M 
           + cache_read * cache_read_price / 1M + cache_write * cache_write_price / 1M
    return CostResult(amount_usd=amount, status="estimated", source=...)

3.16 smart_model_routing.py — 智能模型路由

保守策略

只对"简单"消息路由到廉价模型。判断条件（全部满足才路由）：

1. enabled=True in config
2. 消息长度 ≤ max_simple_chars（默认160）
3. 词数 ≤ max_simple_words（默认28）
4. 不超过1个换行
5. 不含代码块或反引号
6. 不含URL
7. 不含"复杂"关键词（debug, implement, refactor, test, architecture等30+个）

路由结果

resolve_turn_route(user_message, routing_config, primary) → Dict:
    {
        "model": "cheap-model-or-primary",
        "runtime": {...api_key, base_url, provider...},
        "label": "smart route → cheap-model (provider)" or None,
        "signature": (model, provider, base_url, api_mode, command, args)
    }

签名用于判断是否需要重建客户端。

3.17 rate_limit_tracker.py — 速率限制追踪

12个x-ratelimit头部解析

x-ratelimit-limit-requests          RPM上限
x-ratelimit-limit-requests-1h       RPH上限
x-ratelimit-limit-tokens            TPM上限
x-ratelimit-limit-tokens-1h         TPH上限
x-ratelimit-remaining-requests      分钟窗口剩余请求
x-ratelimit-remaining-requests-1h   小时窗口剩余请求
x-ratelimit-remaining-tokens        分钟窗口剩余token
x-ratelimit-remaining-tokens-1h     小时窗口剩余token
x-ratelimit-reset-requests          分钟窗口重置秒数
x-ratelimit-reset-requests-1h       小时窗口重置秒数
x-ratelimit-reset-tokens            分钟窗口重置秒数
x-ratelimit-reset-tokens-1h         小时窗口重置秒数

格式化显示

format_rate_limit_display() → ASCII进度条 + 百分比 + 使用量/限制 + 重置倒计时 + 80%警告

3.18 nous_rate_guard.py — Nous Portal跨会话速率保护

问题

单个429错误可触发最多9次API调用（3 SDK重试 × 3 Hermes重试），每次都计入RPH。这导致"重试放大"效应。

解决方案

共享文件 ~/.hermes/rate_limits/nous.json：

{"reset_at": 1744848000, "recorded_at": 1744847700, "reset_seconds": 300}

• record_nous_rate_limit()：首次429时记录重置时间
• nous_rate_limit_remaining()：后续请求前检查，如果仍在速率限制期则跳过
• clear_nous_rate_limit()：成功请求后清除

原子写入：tempfile + os.replace，防止并发写入损坏。

3.19 error_classifier.py — API错误分类器

分类枚举

class FailoverReason(enum.Enum):
    rate_limit     = "rate_limit"      # 429, 轮换凭证或退避
    auth           = "auth"            # 401/403, 刷新凭证
    overload       = "overload"        # 503/529, 退避重试
    context_limit  = "context_limit"   # prompt过长, 压缩上下文
    content_filter = "content_filter"  # 内容过滤, 修改提示
    unknown        = "unknown"         # 未知错误, 最终失败

分类管道

1. 提取HTTP状态码（从SDK异常/CodeAssistError/原始响应）
2. 提取Retry-After头部
3. 匹配状态码到FailoverReason
4. 特殊模式检测（如 "context window" → context_limit）
5. 返回 ErrorClassification(reason, status_code, retry_after, provider)

提取状态码的5种路径

def _extract_status_code(error):
    # 1. error.status_code (CodeAssistError)
    # 2. error.response.status_code (OpenAI SDK)
    # 3. error.status_code (httpx.HTTPStatusError)  
    # 4. error.http_status (custom exception)
    # 5. 正则从error message提取数字

3.20 retry_utils.py — 去相关指数退避

jittered_backoff()

def jittered_backoff(attempt, base_delay=5.0, max_delay=120.0, jitter_ratio=0.5) -> float:
    delay = min(base_delay * 2^(attempt-1), max_delay)
    jitter = Random(seed=time_ns ^ counter * golden_ratio).uniform(0, jitter_ratio * delay)
    return delay + jitter

为什么用jitter：多个会话同时被429后，如果都用固定退避，会在同一时刻重试（雷群效应）。jitter打破同步，分散重试时间。

种子设计：time_ns ^ (counter * 0x9E3779B9) — 时间戳异或黄金比例哈希的计数器，保证唯一性和分散性。

3.21 redact.py — 正则密钥脱敏

30+种密钥模式

覆盖：OpenAI (sk-)、GitHub (ghp_/github_pat_/gho_/ghu_/ghs_/ghr_)、Slack (xoxb/xoxp)、Google (AIza)、Perplexity (pplx-)、AWS (AKIA)、Stripe (sk_live/sk_test)、HuggingFace (hf_)、JWT (eyJ) 等。

脱敏规则

• 短token（<18字符）：***
• 长token：前6位...后4位（如 sk-ant...k8x9）

覆盖场景

• ENV赋值：OPENAI_API_KEY=sk-xxx → OPENAI_API_KEY=sk-ant...k8x9
• JSON字段："apiKey": "value" → "apiKey": "sk-ant...k8x9"
• Authorization头：Bearer sk-xxx → Bearer sk-ant...k8x9
• 私钥块：整个替换为 [REDACTED PRIVATE KEY]
• 数据库连接串：postgres://user:password@host → postgres://user:***@host
• Telegram token：bot123:token → bot123:***
• 手机号：+8613800138000 → +8613****8000

安全保护

_REDACT_ENABLED 在import时快照环境变量，防止运行时通过 export HERMES_REDACT_SECRETS=false 禁用脱敏。

3.22 prompt_caching.py — Anthropic提示缓存策略

system_and_3策略

在最多4个位置放置 cache_control: {"type": "ephemeral"} 断点：

1. System prompt：第一个message如果是system角色
2. 最近3条非system消息：滚动窗口

为什么最多4个：Anthropic API限制最多4个cache断点。

TTL支持：cache_ttl="1h" 时marker变为 {"type": "ephemeral", "ttl": "1h"}。

注入位置

def _apply_cache_marker(msg, marker, native_anthropic=False):
    # content=str → [{"type":"text","text":content,"cache_control":marker}]
    # content=list → 在最后一个part加cache_control
    # role=tool + native_anthropic → msg级别cache_control

3.23 skill_utils.py — 轻量技能元数据工具

设计约束

不导入工具注册、CLI配置或任何重量级依赖链。安全在模块级别导入。

Frontmatter解析

def parse_frontmatter(content) -> (frontmatter_dict, remaining_body):
    # YAML frontmatter (---...---) → parse with CSafeLoader
    # Fallback: simple key:value splitting for malformed YAML

平台匹配

PLATFORM_MAP = {"macos":"darwin", "linux":"linux", "windows":"win32"}
# skills可声明platforms: [macos, linux]限制运行平台

条件激活

def extract_skill_conditions(frontmatter) -> Dict:
    # metadata.hermes.fallback_for_toolsets
    # metadata.hermes.requires_toolsets
    # metadata.hermes.fallback_for_tools
    # metadata.hermes.requires_tools

配置变量

技能可在frontmatter声明需要的config.yaml变量：

metadata:
  hermes:
    config:
      - key: wiki.path
        description: Path to knowledge base
        default: "~/wiki"

resolve_skill_config_values() 从 skills.config.<key> 路径读取config.yaml中的值。

3.24 skill_commands.py — 斜杠命令路由

命令扫描

scan_skill_commands() → 遍历 ~/.hermes/skills/ + 外部目录 → 解析每个SKILL.md的frontmatter → 生成 /{name} → skill_info映射

命令规范化

# "Code Search" → "/code-search"
# "claude_code" → "/claude-code"  (下划线→连字符)
# 用户输入 /claude_code → 也匹配 /claude-code

技能加载

def build_skill_invocation_message(cmd_key, user_instruction, ...) -> str:
    # 1. 加载SKILL.md全文
    # 2. 解析并注入resolved config值
    # 3. 列出supporting files（references/, templates/, scripts/, assets/）
    # 4. 添加setup note（如果需要）
    # 5. 拼接为完整用户消息

3.25 display.py — CLI展示层

组件

• Spinner：终端旋转动画，显示当前操作状态
• Kawaii faces：(◕‿◕) 风格的状态表情
• Tool preview：格式化工具调用结果的预览（diff高亮、JSON格式化）
• ANSI颜色：根据终端主题（dark/light）自动调整颜色

Diff显示

使用 difflib.unified_diff 生成差异，ANSI颜色标记增删行。

3.26 insights.py — 会话洞察引擎

指标

• Token消耗（按模型/提供商分组）
• 成本估算（按会话/日/模型汇总）
• 工具使用模式（频率、成功率）
• 活动趋势（按小时/日/周）
• 模型/平台分布
• 会话持续时间与轮次统计

数据源

SQLite状态数据库（~/.hermes/state.db），直接SQL查询。

3.27 title_generator.py — 自动标题生成

流程

1. 取用户消息和助手响应的前500字符
2. 调用 auxiliary_client.call_llm() 用最便宜的模型
3. 提示：“Generate a short, descriptive title (3-7 words)”
4. 清理：去引号、去"Title:"前缀、截断80字符
5. 后台线程写入session_db

触发条件

仅在会话的前2轮用户消息时触发（user_msg_count <= 2）。

3.28 trajectory.py — 轨迹保存

格式

ShareGPT JSONL，每行一个对话：

{
    "conversations": [{"from":"human","value":"..."}, {"from":"gpt","value":"..."}],
    "timestamp": "2026-04-19T09:42:00",
    "model": "claude-3-opus",
    "completed": true
}

辅助函数

• convert_scratchpad_to_think()：<REASONING_SCRATCHPAD> → </think> 标签转换
• has_incomplete_scratchpad()：检测未关闭的思考标签

3.29 subdirectory_hints.py — 子目录上下文发现

工作原理

1. 每次工具调用后，check_tool_call(tool_name, tool_args) 检查参数中的路径
2. 从路径提取目录，向上最多走5层祖先
3. 在每个新目录搜索 AGENTS.md / CLAUDE.md / .cursorrules
4. 找到后加载内容（最大8000字符），追加到工具结果
5. 标记已加载目录，避免重复

路径提取

• 直接路径参数：path, file_path, workdir
• Shell命令：解析 terminal 工具的命令字符串，提取路径式token

3.30 google_oauth.py — Google OAuth PKCE流程

PKCE流程

1. 生成verifier(64字节url-safe) + challenge(SHA256+base64url)
2. 浏览器打开授权URL（含challenge）
3. 回调服务器(localhost:8085)接收code
4. exchange_code(code, verifier) → access_token + refresh_token
5. 保存到~/.hermes/auth/google_oauth.json

无头模式

检测SSH环境变量 → 跳过回调服务器 → 用户手动粘贴重定向URL → 解析code参数

客户端ID解析

1. 环境变量 HERMES_GEMINI_CLIENT_ID
2. 内置默认值（Google gemini-cli的公开OAuth客户端）
3. 从本地安装的gemini-cli二进制文件抓取

RefreshToken打包格式

refresh_token[|project_id[|managed_project_id]]

管道符分隔，支持无项目ID的向后兼容。

3.31 google_code_assist.py — Code Assist控制面

API端点

https://cloudcode-pa.googleapis.com/v1internal:{loadCodeAssist|onboardUser|retrieveUserQuota}

Tier系统

• free-tier：免费层，Google分配managed project
• standard-tier：标准层，需要GCP project_id
• legacy-tier：遗留层

Onboard LRO轮询

for attempt in range(12):  # 最多12次
    time.sleep(5)           # 间隔5秒
    poll_resp = POST /v1internal/{op_name}
    if poll_resp.get("done"): return poll_resp
# 总等待时间：最多60秒

3.32 manual_compression_feedback.py — 手动压缩反馈

单一函数 summarize_manual_compression()，生成用户可读的压缩反馈：

Compressed: 20 → 8 messages
Rough transcript estimate: ~45,000 → ~12,000 tokens

处理边界情况：压缩后消息更少但token更多（更密集的摘要）时给出说明。

---

四、业务规则验证与设计评价

4.1 正确性验证

规则	实现方式	评价
Anthropic最多4个cache断点	apply_anthropic_cache_control 限制breakpoints_used ≤ 4	✅ 正确
OpenAI input_tokens包含cache	normalize_usage() 中 input_tokens = prompt_total - cache_read - cache_write	✅ 正确
同一provider只允许1个外部记忆	MemoryManager.register() 检查并拒绝第二个外部provider	✅ 正确
密钥脱敏不可运行时禁用	_REDACT_ENABLED 在import时快照	✅ 安全设计
凭证刷新跨线程去重	_refresh_inflight dict + Event	✅ 正确
Nous速率限制原子写入	tempfile + os.replace	✅ 正确

4.2 架构亮点

1. Provider适配器模式：每种后端一个适配器，统一到OpenAI兼容接口，新增后端只需实现翻译层
2. 凭证池故障转移：生产级的多密钥管理，自动轮换+自动刷新
3. 可插拔引擎：ContextEngine和MemoryProvider的ABC设计，第三方可扩展
4. 去相关退避：jittered backoff防止雷群效应
5. 跨会话速率保护：文件级共享状态，防止多会话重试放大

4.3 潜在风险

1. credential_pool.py（59KB）：文件过大，CredentialEntry和pool逻辑耦合，建议拆分
2. auxiliary_client.py（117KB）：单一文件包含8种后端解析逻辑，建议按后端拆分
3. 硬编码定价表：_OFFICIAL_DOCS_PRICING 需手动更新，模型价格变化时可能过时
4. 记忆搜索：BuiltinMemoryProvider使用简单文本搜索，大规模记忆时性能不足
5. 子目录提示：_MAX_ANCESTOR_WALK=5 可能错过更深层的上下文文件

---

五、总结

Hermes Agent的 agent/ 模块是一个精心设计的基础设施层，从3600行单体文件提取而来。它遵循单一职责和依赖倒置原则，通过适配器模式统一多LLM后端，通过ABC实现可插拔引擎，通过凭证池实现生产级故障转移。代码质量整体高，有详细的文档字符串和安全防护。主要改进方向是拆分超大文件和引入向量搜索记忆。

模块数量: 34
总代码量: ~700K字符
抽象基类: 2 (ContextEngine, MemoryProvider)
适配器: 5 (Anthropic, Bedrock, Gemini, Copilot, Auxiliary)
数据类: 15+
安全脱敏模式: 30+