Hermes Agent Gateway统一控制面架构深度解析

Gateway统一控制面架构深度解析

文档定位：本文档是 Hermes Agent 自进化智能体高阶实战知识体系中「产品定位与核心架构」章节的第二个拆解点，聚焦于 Gateway 统一控制面的架构设计、核心机制与工程实现。

适用读者：架构设计人员、AI Agent 平台开发者、后端高级工程师、技术决策者

前置依赖：建议先阅读《01_产品定位与系统全景》以建立整体认知框架

👉写博客的朋友可以看看这个，助力内容创作，懂技术更懂增长，开发者与企业的创作增长引擎 ！👈👉点击进入

目录

• 第一章 Gateway 架构总览
  • 1.1 设计理念
  • 1.2 核心职责矩阵
  • 1.3 在系统中的中枢地位
  • 1.4 架构分层视图
  • 1.5 技术选型概览
• 第二章 会话管理器深度解析
  • 2.1 多轮对话上下文维护
  • 2.2 会话 ID 路由机制
  • 2.3 上下文窗口压缩策略
  • 2.4 并行会话隔离方案
  • 2.5 会话生命周期管理
• 第三章 渠道适配器架构
  • 3.1 适配器模式设计
  • 3.2 消息格式归一化流程
  • 3.3 多平台协议差异处理
  • 3.4 回调机制与异步消息处理
  • 3.5 渠道健康检查与故障转移
• 第四章 工具注册表设计
  • 4.1 工具注册与发现机制
  • 4.2 MCP 协议集成
  • 4.3 内置工具与自定义工具统一管理
  • 4.4 权限校验与调用审计
  • 4.5 工具版本管理与热更新
• 第五章 核心路由引擎
  • 5.1 意图识别算法
  • 5.2 任务编排策略
  • 5.3 记忆调度机制
  • 5.4 安全沙箱集成
• 第六章 Gateway 与其他模块的交互
  • 6.1 与记忆系统的协作
  • 6.2 与模型路由器的协作
  • 6.3 与自进化引擎的协作
• 第七章 Gateway 性能与扩展性
  • 7.1 并发处理能力
  • 7.2 水平扩展方案
  • 7.3 故障恢复机制
• 第八章 与其他 Agent 框架的 Gateway 对比
  • 8.1 架构哲学对比
  • 8.2 核心能力对比矩阵
  • 8.3 场景适用性分析

第一章 Gateway 架构总览

1.1 设计理念

Gateway 统一控制面是 Hermes Agent 架构体系中的中枢神经系统，其设计理念根植于以下五个核心原则：

1.1.1 单一入口原则（Single Entry Point）

所有外部渠道——无论是 Web 客户端、移动应用、即时通讯平台还是 API 调用——都必须经过 Gateway 这唯一的入口点。这一设计带来了三个关键收益：

• 统一安全边界：所有请求经过同一套认证、鉴权、限流、审计管线，消除了多入口带来的安全盲区
• 协议归一化：不同渠道的异构协议在 Gateway 层面被统一翻译为内部标准消息格式，下游模块无需感知渠道差异
• 可观测性收敛：全链路追踪、指标采集、日志聚合在同一层完成，为系统提供端到端的事务可见性

1.1.2 控制面与数据面分离（Control-Data Plane Separation）

Gateway 严格遵循云原生架构中控制面与数据面分离的设计范式：

┌─────────────────────────────────────────────────────────────────────┐
│ Hermes Agent 系统全景 │
│ │
│ ┌─────────────────┐ 控制信号 ┌─────────────────────────────┐ │
│ │ │ ─────────> │ │ │
│ │ 控制面(Gateway) │ │ 数据面 (执行层) │ │
│ │ │ <───────── │ │ │
│ │ · 会话管理 │ 状态回传 │ · 模型推理 │ │
│ │ · 渠道适配 │ │ · 工具执行 │ │
│ │ · 工具调度 │ │ · 记忆读写 │ │
│ │ · 意图路由 │ │ · 外部API调用 │ │
│ │ · 策略决策 │ │ · 知识检索 │ │
│ └────────┬────────┘ └─────────────────────────────┘ │
│ │ │
│ │ 控制指令 │
│ v │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 基础设施层 │ │
│ │ 消息队列 · 分布式缓存 · 对象存储 · 日志中心 · 监控 │ │
│ └─────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘

控制面负责「做什么」和「怎么调度」——决定请求的路由方向、编排策略、安全策略；数据面负责「实际执行」——模型推理、工具调用、记忆持久化。这种分离使得控制逻辑可以独立演进、独立部署、独立扩展，同时为数据面的多副本水平扩展提供了前提条件。

1.1.3 声明式配置优于硬编码（Declarative over Imperative）

Gateway 的所有可变行为——渠道映射、工具路由规则、限流策略、安全沙箱策略——均通过声明式配置定义，而非硬编码在业务逻辑中。这使得运维人员可以在不修改代码的情况下调整系统行为：

# gateway-config.yaml — Gateway 声明式配置示例
gateway:
server:
port: 8080
worker_threads: 16
max_connections: 10000
request_timeout: 120s
grpc:
enabled: true
port: 9090
max_recv_msg_size: 16MB

session:
default_ttl: 3600s
max_concurrent_per_user: 5
context_window:
max_tokens: 32768
compression_threshold: 0.75
compression_strategy: sliding_window_summary
storage:
type: redis_cluster
endpoints:
– redis–0.internal:6379
– redis–1.internal:6379
– redis–2.internal:6379
key_prefix: "hermes:session:"

channels:
– name: web
adapter: WebSocketAdapter
enabled: true
rate_limit:
requests_per_minute: 120
burst: 30
– name: feishu
adapter: FeishuAdapter
enabled: true
rate_limit:
requests_per_minute: 60
callback:
url: https://gateway.internal/feishu/callback
verify_token: ${FEISHU_VERIFY_TOKEN}
– name: dingtalk
adapter: DingTalkAdapter
enabled: true
rate_limit:
requests_per_minute: 60
– name: wechat_work
adapter: WeComAdapter
enabled: true
– name: slack
adapter: SlackAdapter
enabled: false
– name: telegram
adapter: TelegramAdapter
enabled: false

routing:
intent_model: hermes–intent–v2
fallback_strategy: llm_based
max_orchestration_depth: 8
parallel_task_limit: 4
timeout_per_task: 30s

tools:
registry:
type: dynamic
refresh_interval: 30s
sandbox:
enabled: true
engine: firecracker
memory_limit: 512MB
cpu_limit: 1.0
network: restricted
allowed_domains:
– api.internal
– "*.hermes.ai"
audit:
enabled: true
log_retention: 90d
sensitive_params_mask: true

observability:
tracing:
enabled: true
exporter: otlp
endpoint: jaeger–collector.observability:4317
sampling_rate: 0.1
metrics:
enabled: true
exporter: prometheus
port: 9091
logging:
level: info
format: json
output: file
path: /var/log/hermes/gateway.log
rotation:
max_size: 100MB
max_files: 10
max_age: 30d

1.1.4 渐进式退化原则（Graceful Degradation）

Gateway 被设计为在任何单点故障下都能保持核心功能的可用性。当记忆系统不可用时，Gateway 退化为无记忆模式继续服务；当某个渠道适配器故障时，其他渠道不受影响；当模型路由器超时时，Gateway 使用本地缓存的最优模型配置执行降级推理。这一原则确保了系统在恶劣条件下的韧性。

1.1.5 可插拔架构原则（Pluggable Architecture）

Gateway 的每一个核心组件——渠道适配器、工具执行器、意图识别器、记忆后端——都通过标准接口定义，实现可插拔替换。这意味着用户可以用自定义实现替换任何一个组件，而无需修改 Gateway 核心代码：

┌──────────────────────────────────────────────────────────────────────────┐
│ Gateway 可插拔架构 │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────┐ │
│ │ Channel │ │ Tool │ │ Intent │ │ Memory │ │
│ │ Adapter SPI │ │ Executor SPI │ │ Detector SPI │ │ Store SPI│ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └────┬─────┘ │
│ │ │ │ │ │
│ ┌────┴────┐ ┌────┴────┐ ┌────┴────┐ ┌────┴────┐ │
│ │ 内置实现 │ │ 内置实现 │ │ 内置实现 │ │ 内置实现 │ │
│ │ Web │ │ Code │ │ Rule │ │ Redis │ │
│ │ Feishu │ │ Search │ │ ML Model │ │ Postgres│ │
│ │ Slack │ │ HTTP │ │ LLM Based│ │ Vector │ │
│ │ … │ │ MCP │ │ Hybrid │ │ Hybrid │ │
│ └─────────┘ └─────────┘ └──────────┘ └─────────┘ │
│ ┌─────────┐ ┌─────────┐ ┌──────────┐ ┌─────────┐ │
│ │ 用户扩展 │ │ 用户扩展 │ │ 用户扩展 │ │ 用户扩展 │ │
│ │ Custom │ │ Custom │ │ Custom │ │ Custom │ │
│ │ Adapter │ │ Tool │ │ Detector │ │ Store │ │
│ └─────────┘ └─────────┘ └──────────┘ └─────────┘ │
└──────────────────────────────────────────────────────────────────────────┘

1.2 核心职责矩阵

Gateway 承担着四大核心职责，每一项职责都对应一组具体的子功能和工程要求：

1.3 在系统中的中枢地位

Gateway 在 Hermes Agent 系统中的位置可以用「六边形中枢」来形容——它向上对接所有外部渠道，向下调度所有内部能力模块，横向贯穿记忆系统与模型路由器，同时接受自进化引擎的策略指导：

┌─────────────────────┐
│ 自进化引擎 │
│ (Self-Evolution) │
│ · 策略优化 │
│ · 参数调优 │
│ · 能力评估 │
└─────────┬───────────┘
│ 策略下发
v
┌───────────┐ 请求 ┌─────────────────────────────────┐ 调用 ┌───────────────┐
│ │ ═══════> │ │ ════════> │ │
│ Web UI │ │ │ │ 模型路由器 │
│ │ <═══════ │ G A T E W A Y │ <════════ │ (Model │
├───────────┤ 响应 │ │ 结果 │ Router) │
│ │ │ 统 一 控 制 面 │ │ │
│ 飞书/钉钉 │ ═══════> │ ┌───────────┐ ┌──────────────┐ │ ├───────────────┤
│ │ │ │ 会话管理器 │ │ 渠道适配器集群 │ │ │ │
├───────────┤ │ │ Session │ │ Adapters │ │ │ 记忆系统 │
│ │ │ │ Manager │ │ │ │ │ (Memory │
│ Slack/ │ ═══════> │ └───────────┘ └──────────────┘ │ │ System) │
│ Telegram │ │ ┌───────────┐ ┌──────────────┐ │ │ │
│ │ │ │ 工具注册表 │ │ 核心路由引擎 │ │ ════════> ├───────────────┤
├───────────┤ │ │ Tool │ │ Routing │ │ │ │
│ │ │ │ Registry │ │ Engine │ │ │ 工具执行层 │
│ API/SDK │ ═══════> │ └───────────┘ └──────────────┘ │ │ (Tool │
│ │ │ │ │ Executors) │
│ │ └─────────────────────────────────┘ └───────────────┘
│ │ │
│ │ 基础设施层
│ │ ┌─────────────┴──────────────┐
│ │ │ Redis · Kafka · S3 · ES │
│ │ │ Jaeger · Prometheus · ELK │
│ │ └────────────────────────────┘
└───────────┘

从数据流的角度看，Gateway 是所有请求的必经之路。一个典型的请求生命周期如下：

用户消息 → 渠道适配器(协议归一化) → 会话管理器(上下文加载)
→ 核心路由引擎(意图识别+任务编排)
→ [并行] 模型路由器(LLM推理) + 记忆系统(记忆检索)
→ 工具注册表(工具调用) → 结果聚合
→ 会话管理器(上下文更新) → 渠道适配器(格式转换) → 用户响应

1.4 架构分层视图

Gateway 内部采用清晰的六层架构，每层职责单一、边界明确：

┌─────────────────────────────────────────────────────────────────────┐
│ Layer 6: 接入层 (Access Layer) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ HTTP/ │ │ gRPC │ │ WebSocket│ │ Webhook │ │ SDK │ │
│ │ REST │ │ Server │ │ Server │ │ Handler │ │ Gateway │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
├─────────────────────────────────────────────────────────────────────┤
│ Layer 5: 协议适配层 (Protocol Adaptation Layer) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌────────────┐ │
│ │ Web Adapter │ │ Feishu │ │ DingTalk │ │ Slack │ │
│ │ │ │ Adapter │ │ Adapter │ │ Adapter │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ └────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌────────────┐ │
│ │ WeCom Adapter│ │ Telegram │ │ Discord │ │ Email │ │
│ │ │ │ Adapter │ │ Adapter │ │ Adapter │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ └────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ WhatsApp │ │ Teams │ │ Custom… │ │
│ │ Adapter │ │ Adapter │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
├─────────────────────────────────────────────────────────────────────┤
│ Layer 4: 会话管理层 (Session Management Layer) │
│ ┌─────────────┐ ┌───────────────┐ ┌───────────────┐ ┌────────────┐ │
│ │ Session │ │ Context │ │ Session │ │ TTL │ │
│ │ Store │ │ Compressor │ │ Router │ │ Manager │ │
│ └─────────────┘ └───────────────┘ └───────────────┘ └────────────┘ │
├─────────────────────────────────────────────────────────────────────┤
│ Layer 3: 路由编排层 (Routing & Orchestration Layer) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌────────────┐ │
│ │ Intent │ │ Task │ │ Memory │ │ Safety │ │
│ │ Detector │ │ Orchestrator │ │ Scheduler │ │ Sandbox │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ └────────────┘ │
├─────────────────────────────────────────────────────────────────────┤
│ Layer 2: 工具调度层 (Tool Orchestration Layer) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌────────────┐ │
│ │ Tool │ │ Permission │ │ Tool │ │ Audit │ │
│ │ Registry │ │ Checker │ │ Dispatcher │ │ Logger │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ └────────────┘ │
├─────────────────────────────────────────────────────────────────────┤
│ Layer 1: 基础设施层 (Infrastructure Layer) │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Cache │ │ Queue │ │ Storage │ │ Config │ │ Monitor │ │
│ │ (Redis) │ │ (Kafka) │ │ (S3/DB) │ │ Center │ │ Agent │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────────────┘

各层职责详解

Layer 6 — 接入层：负责网络协议的接入与连接管理。支持 HTTP/REST、gRPC、WebSocket 三种主要协议，同时提供 Webhook 回调端点和 SDK 直连通道。接入层实施 TLS 终止、连接池管理、请求大小限制等基础网络策略。

Layer 5 — 协议适配层：将各渠道的私有协议适配为 Gateway 内部统一消息格式。每个适配器实现标准的 ChannelAdapter SPI 接口，负责入站消息解析和出站消息格式化。适配器之间完全隔离，互不影响。

Layer 4 — 会话管理层：维护多轮对话的上下文状态。Session Store 负责会话数据的持久化与缓存；Context Compressor 在上下文接近 Token 预算时执行压缩；Session Router 根据会话 ID 将请求路由到正确的处理节点；TTL Manager 管理会话的生命周期过期。

Layer 3 — 路由编排层：Gateway 的大脑。Intent Detector 识别用户意图；Task Orchestrator 根据意图编排任务执行计划；Memory Scheduler 决定何时读写记忆；Safety Sandbox 为高风险操作提供隔离执行环境。

Layer 2 — 工具调度层：管理所有可用工具的注册、发现、调度与审计。Tool Registry 维护工具元数据；Permission Checker 校验调用权限；Tool Dispatcher 执行实际工具调用；Audit Logger 记录全量调用日志。

Layer 1 — 基础设施层：为上层提供分布式缓存、消息队列、对象存储、配置管理和监控代理等基础能力。

1.5 技术选型概览

第二章 会话管理器深度解析

会话管理器是 Gateway 中最核心的组件之一，它承担着维护多轮对话上下文完整性、保障会话隔离性、优化上下文窗口利用率的关键职责。在一个典型的 Agent 系统中，会话管理的质量直接决定了用户交互的连贯性和推理质量的上限。

2.1 多轮对话上下文维护

2.1.1 上下文数据模型

Hermes Agent 的会话上下文采用分层结构进行组织，每一层承载不同粒度的信息：

/// 会话上下文的核心数据结构
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SessionContext {
/// 会话唯一标识
pub session_id: SessionId,
/// 用户标识
pub user_id: UserId,
/// 渠道来源
pub channel: ChannelType,
/// 会话创建时间
pub created_at: DateTime<Utc>,
/// 最后活跃时间
pub last_active_at: DateTime<Utc>,
/// 会话状态
pub status: SessionStatus,
/// 对话消息序列（按时间排序）
pub messages: Vec<Message>,
/// 会话级元数据
pub metadata: SessionMetadata,
/// 压缩后的摘要（当上下文被压缩时填充）
pub compressed_summary: Option<Summary>,
/// 活跃工具调用栈
pub active_tool_calls: Vec<ToolCallContext>,
/// 会话级记忆引用
pub memory_refs: Vec<MemoryRef>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Message {
/// 消息唯一ID
pub message_id: MessageId,
/// 消息角色：user / assistant / system / tool
pub role: MessageRole,
/// 消息内容（支持多模态）
pub content: MessageContent,
/// 消息时间戳
pub timestamp: DateTime<Utc>,
/// 消息元数据（模型ID、Token消耗、延迟等）
pub metadata: MessageMetadata,
/// 关联的工具调用（如果有）
pub tool_calls: Vec<ToolCall>,
/// 消息状态
pub status: MessageStatus,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum MessageContent {
/// 纯文本
Text(String),
/// 多模态内容（文本+图片+文件等）
Multimodal {
text: Option<String>,
images: Vec<ImageContent>,
files: Vec<FileContent>,
attachments: Vec<Attachment>,
},
/// 工具调用结果
ToolResult {
tool_call_id: String,
result: serde_json::Value,
is_error: bool,
},
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SessionMetadata {
/// 当前使用的模型
pub current_model: Option<String>,
/// 累计 Token 消耗
pub total_tokens_consumed: u64,
/// 对话轮次
pub turn_count: u32,
/// 会话主题（由意图识别自动提取）
pub topic: Option<String>,
/// 用户偏好（从记忆系统加载）
pub user_preferences: HashMap<String, serde_json::Value>,
/// 自定义标签
pub tags: Vec<String>,
}

2.1.2 上下文增量更新机制

每次消息交互后，会话上下文需要进行增量更新。Hermes Agent 采用 Write-Ahead Log (WAL) 机制保证上下文更新的原子性和持久性：

/// 会话上下文增量更新器
pub struct ContextUpdater {
session_store: Arc<SessionStore>,
wal: Arc<WriteAheadLog>,
compressor: Arc<ContextCompressor>,
}

impl ContextUpdater {
/// 追加新消息到会话上下文
pub async fn append_message(
&self,
session_id: &SessionId,
message: Message,
) -> Result<UpdateResult, SessionError> {
// 1. 写入 WAL，保证原子性
let wal_entry = WalEntry::AppendMessage {
session_id: session_id.clone(),
message: message.clone(),
timestamp: Utc::now(),
};
self.wal.append(&wal_entry).await?;

// 2. 更新内存中的会话上下文
let mut session = self.session_store.get(session_id).await?;
session.messages.push(message.clone());
session.metadata.turn_count += 1;
session.last_active_at = Utc::now();

// 3. 计算 Token 使用量
let token_count = self.estimate_tokens(&session);
let token_budget = self.get_token_budget(session_id).await?;

// 4. 判断是否需要压缩
let needs_compression = token_count as f64 / token_budget as f64
> self.compressor.compression_threshold;

if needs_compression {
let compressed = self.compressor.compress(&session).await?;
session.compressed_summary = Some(compressed.summary);
// 保留最近 N 条原始消息 + 压缩摘要
let retain_count = self.compressor.recent_message_window;
session.messages = session.messages.split_off(
session.messages.len().saturating_sub(retain_count)
);
}

// 5. 持久化到 Session Store
self.session_store.put(session_id, &session).await?;

// 6. 标记 WAL 条目为已完成
self.wal.mark_complete(&wal_entry).await?;

Ok(UpdateResult {
session_id: session_id.clone(),
token_count,
token_budget,
compressed: needs_compression,
turn_count: session.metadata.turn_count,
})
}
}

2.1.3 上下文版本控制

为了支持会话回滚和调试，会话管理器为每次上下文变更维护版本号：

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SessionVersion {
/// 版本号（单调递增）
pub version: u64,
/// 变更类型
pub change_type: ChangeType,
/// 变更时间
pub timestamp: DateTime<Utc>,
/// 变更前快照的引用（仅保留最近 N 个版本）
pub parent_version: Option<u64>,
}

pub enum ChangeType {
MessageAppended,
ContextCompressed,
ToolCallStarted,
ToolCallCompleted,
MetadataUpdated,
SessionResumed,
SessionForked,
}

/// 版本控制操作接口
pub trait SessionVersioning {
/// 获取当前版本
async fn current_version(&self, session_id: &SessionId) -> u64;

/// 回滚到指定版本
async fn rollback_to(
&self,
session_id: &SessionId,
target_version: u64,
) -> Result<SessionContext, SessionError>;

/// 创建会话分支（用于并行探索）
async fn fork(
&self,
session_id: &SessionId,
at_version: u64,
) -> Result<SessionId, SessionError>;

/// 获取变更历史
async fn changelog(
&self,
session_id: &SessionId,
from: Option<u64>,
to: Option<u64>,
) -> Vec<SessionVersion>;
}

上下文版本控制的核心应用场景包括：

2.2 会话 ID 路由机制

2.2.1 会话 ID 生成策略

会话 ID 是 Gateway 路由的核心标识。Hermes Agent 采用 Snowflake 变体 生成全局唯一、时间有序的会话 ID：

/// 会话 ID 结构（128位）
/// | 1位符号 | 41位时间戳 | 10位机器ID | 10位渠道ID | 66位序列号 |
#[derive(Clone, Debug, Hash, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionId(u128);

impl SessionId {
pub fn new(machine_id: u16, channel_id: u16) -> Self {
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as u64;

let sequence = SEQUENCE_COUNTER.fetch_add(1, Ordering::SeqCst);

let id = ((timestamp as u128) << 87)
| ((machine_id as u128) << 77)
| ((channel_id as u128) << 67)
| (sequence & 0x7FFFFFFFFFFFFF);

SessionId(id)
}

/// 从会话 ID 提取时间戳
pub fn timestamp(&self) -> u64 {
(self.0 >> 87) as u64
}

/// 从会话 ID 提取机器 ID（用于路由）
pub fn machine_id(&self) -> u16 {
((self.0 >> 77) & 0x3FF) as u16
}

/// 从会话 ID 提取渠道 ID
pub fn channel_id(&self) -> u16 {
((self.0 >> 67) & 0x3FF) as u16
}
}

会话 ID 的设计要点：

• 时间有序：41位时间戳使得会话 ID 天然按创建时间排序，有利于范围查询和时间窗口过滤
• 机器可路由：10位机器 ID 直接编码在会话 ID 中，无需额外查表即可确定会话归属节点
• 渠道可追溯：10位渠道 ID 使得从会话 ID 即可识别来源渠道，便于渠道维度的统计和故障排查
• 序列号充足：66位序列号理论上支持每毫秒 2^66 个会话创建，永远不会溢出

2.2.2 一致性哈希路由

Gateway 集群中，会话数据分布在多个节点上。采用一致性哈希环确保会话亲和性——同一个会话的请求始终路由到同一节点：

┌──────────────────────────────────────────────────────────────────────┐
│ 一致性哈希路由环 │
│ │
│ 0 / 2^128 │
│ │ │
│ Node A ●──────┼──────● Node D │
│ (hash:0x3F..) │ (hash:0xC0..) │
│ ╱ │ ╲ │
│ ╱ │ ╲ │
│ ╱ │ ╲ │
│ Node B ●─────╱ │ ╲─────● Node C │
│ (hash:0x10..) │ (hash:0x80..) │
│ │ │
│ Session ID hash ──> 沿顺时针找到的第一个节点 │
│ │
│ 虚拟节点 (Virtual Nodes): │
│ Node A: v1, v2, v3, v4, v5 (5个虚拟节点) │
│ Node B: v1, v2, v3, v4, v5 (5个虚拟节点) │
│ Node C: v1, v2, v3, v4, v5 (5个虚拟节点) │
│ Node D: v1, v2, v3, v4, v5 (5个虚拟节点) │
│ │
│ 故障转移: Node B 宕机 → 其虚拟节点顺时针迁移到 Node C 和 Node D │
└──────────────────────────────────────────────────────────────────────┘

/// 一致性哈希路由器
pub struct ConsistentHashRouter {
/// 哈希环：SortedMap<Hash, NodeId>
ring: Arc<RwLock<BTreeMap<u128, NodeId>>>,
/// 虚拟节点数
virtual_nodes: usize,
/// 节点列表
nodes: Arc<RwLock<HashMap<NodeId, NodeInfo>>>,
}

impl ConsistentHashRouter {
/// 根据会话 ID 路由到目标节点
pub async fn route(&self, session_id: &SessionId) -> NodeId {
let ring = self.ring.read().await;
let hash = self.hash(session_id);

// 沿顺时针找到第一个节点
match ring.range(hash..).next() {
Some((_, node_id)) => *node_id,
None => {
// 环回到第一个节点
ring.iter().next().map(|(_, id)| *id).unwrap()
}
}
}

/// 获取会话的备用节点（用于故障转移）
pub async fn get_backup_nodes(
&self,
session_id: &SessionId,
count: usize,
) -> Vec<NodeId> {
let ring = self.ring.read().await;
let hash = self.hash(session_id);

let mut backups = Vec::new();
let mut seen = HashSet::new();

for (_, node_id) in ring.range(hash..).chain(ring.iter()) {
if !seen.contains(node_id) {
seen.insert(*node_id);
backups.push(*node_id);
if backups.len() >= count {
break;
}
}
}
backups
}

/// 节点加入集群
pub async fn add_node(&self, node_id: NodeId, node_info: NodeInfo) {
let mut ring = self.ring.write().await;
let mut nodes = self.nodes.write().await;

nodes.insert(node_id, node_info);

// 添加虚拟节点
for i in 0..self.virtual_nodes {
let vhash = self.hash_with_suffix(&node_id, i);
ring.insert(vhash, node_id);
}
}

/// 节点离开集群
pub async fn remove_node(&self, node_id: NodeId) {
let mut ring = self.ring.write().await;
let mut nodes = self.nodes.write().await;

nodes.remove(&node_id);

let to_remove: Vec<u128> = ring
.iter()
.filter(|(_, id)| **id == node_id)
.map(|(h, _)| *h)
.collect();

for h in to_remove {
ring.remove(&h);
}
}
}

一致性哈希路由的核心优势在于最小化节点变更时的数据迁移。当集群中增加或减少一个节点时，平均只有 1/N（N为节点数）的会话需要迁移，而非全部重新分布。每个物理节点配置 5-10 个虚拟节点，可以进一步均衡数据分布，避免热点问题。

2.2.3 会话亲和性与跨节点迁移

当持有某会话的节点发生故障时，Gateway 需要将该会话迁移到备用节点。迁移过程需要保证三个关键属性：

/// 会话迁移管理器
pub struct SessionMigrationManager {
router: Arc<ConsistentHashRouter>,
session_store: Arc<SessionStore>,
migration_queue: Arc<MigrationQueue>,
}

impl SessionMigrationManager {
/// 执行会话迁移
pub async fn migrate_session(
&self,
session_id: &SessionId,
from_node: NodeId,
to_node: NodeId,
) -> Result<MigrationResult, MigrationError> {
tracing::info!(
session_id = %session_id,
from = %from_node,
to = %to_node,
"Starting session migration"
);

// 1. 在源节点标记会话为「迁移中」状态
self.session_store
.set_migration_state(session_id, MigrationState::InProgress)
.await?;

// 2. 从源节点读取完整会话数据
let session_data = self.session_store
.get_full_session(session_id)
.await?;

// 3. 将会话数据写入目标节点
self.session_store
.put_to_node(&to_node, session_id, &session_data)
.await?;

// 4. 验证数据完整性
let verify_data = self.session_store
.get_from_node(&to_node, session_id)
.await?;

if verify_data.version != session_data.version {
return Err(MigrationError::VerificationFailed);
}

// 5. 更新路由表
self.router
.update_session_route(session_id, to_node)
.await?;

// 6. 从源节点删除旧数据
self.session_store
.delete_from_node(&from_node, session_id)
.await?;

// 7. 标记迁移完成
self.session_store
.set_migration_state(session_id, MigrationState::Completed)
.await?;

tracing::info!(
session_id = %session_id,
"Session migration completed successfully"
);

Ok(MigrationResult {
session_id: session_id.clone(),
from_node,
to_node,
duration: session_data.messages.len() as u64,
})
}
}

会话迁移的触发场景包括：

2.3 上下文窗口压缩策略

随着对话轮次增加，上下文 Token 数会逐渐逼近甚至超出模型的最大上下文窗口。Hermes Agent 实现了多级压缩策略，在保留关键信息的同时控制 Token 消耗。

2.3.1 压缩策略总览

┌─────────────────────────────────────────────────────────────────────┐
│ 上下文压缩策略决策树 │
│ │
│ 当前 Token 数 │
│ │ │
│ ┌───────────┴───────────┐ │
│ │ │ │
│ < 50% 预算 > 50% 预算 │
│ │ │ │
│ 不压缩 ┌─────────┴─────────┐ │
│ │ │ │
│ < 75% 预算 > 75% 预算 │
│ │ │ │
│ 轻度压缩 ┌────┴────┐ │
│ (消息裁剪) < 90% > 90% │
│ │ │ │ │
│ │ 中度压缩 重度压缩 │
│ │ (摘要+裁剪) (全量摘要+ │
│ │ │ 仅保留最近3轮) │
│ v v v │
│ ┌──────────────────────────────────┐ │
│ │ 压缩后 Token < 50% 预算 │ │
│ └──────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘

2.3.2 滑动窗口摘要压缩算法

这是 Hermes Agent 的核心压缩算法，采用滑动窗口 + 渐进式摘要的方式：

/// 上下文压缩器
pub struct ContextCompressor {
/// 压缩阈值（占预算的比例）
pub compression_threshold: f64,
/// 保留最近消息的窗口大小
pub recent_message_window: usize,
/// 摘要模型
pub summary_model: String,
/// Token 估算器
pub token_estimator: TokenEstimator,
}

#[derive(Clone, Debug, PartialEq)]
pub enum CompressionStrategy {
Light,
Medium,
Heavy,
None,
}

impl ContextCompressor {
/// 执行上下文压缩
pub async fn compress(
&self,
session: &SessionContext,
) -> Result<CompressionResult, CompressionError> {
let total_tokens = self.token_estimator.estimate_session(session);
let budget = self.token_estimator.get_budget(&session.metadata.current_model);

if (total_tokens as f64 / budget as f64) < self.compression_threshold {
return Ok(CompressionResult::NoCompressionNeeded);
}

// 根据压缩级别选择策略
let ratio = total_tokens as f64 / budget as f64;
let strategy = if ratio < 0.75 {
CompressionStrategy::Light
} else if ratio < 0.90 {
CompressionStrategy::Medium
} else {
CompressionStrategy::Heavy
};

match strategy {
CompressionStrategy::Light => self.light_compress(session).await,
CompressionStrategy::Medium => self.medium_compress(session).await,
CompressionStrategy::Heavy => self.heavy_compress(session).await,
CompressionStrategy::None => Ok(CompressionResult::NoCompressionNeeded),
}
}

/// 轻度压缩：裁剪工具调用的详细输出
async fn light_compress(
&self,
session: &SessionContext,
) -> Result<CompressionResult, CompressionError> {
let mut compressed_messages = Vec::new();
let mut tokens_saved = 0u64;

for msg in &session.messages {
if msg.role == MessageRole::Tool {
// 将工具调用的详细输出替换为摘要
let summary = self.summarize_tool_output(&msg.content).await?;
tokens_saved += self.token_estimator.estimate_message(msg)
– self.token_estimator.estimate_text(&summary);

compressed_messages.push(Message {
content: MessageContent::Text(summary),
metadata: msg.metadata.clone(),
..msg.clone()
});
} else {
compressed_messages.push(msg.clone());
}
}

Ok(CompressionResult::Compressed {
strategy: CompressionStrategy::Light,
messages: compressed_messages,
tokens_saved,
})
}

/// 中度压缩：历史消息摘要 + 保留最近N轮
async fn medium_compress(
&self,
session: &SessionContext,
) -> Result<CompressionResult, CompressionError> {
let split_point = session.messages.len()
.saturating_sub(self.recent_message_window);

// 将早期消息生成摘要
let old_messages = &session.messages[..split_point];
let recent_messages = &session.messages[split_point..];

let summary = self.generate_summary(old_messages).await?;

let mut compressed = vec![Message::system(summary)];
compressed.extend_from_slice(recent_messages);

let tokens_saved = self.token_estimator.estimate_messages(old_messages)
– self.token_estimator.estimate(&compressed[0]);

Ok(CompressionResult::Compressed {
strategy: CompressionStrategy::Medium,
messages: compressed,
tokens_saved,
})
}

/// 重度压缩：全量摘要 + 仅保留最近3轮
async fn heavy_compress(
&self,
session: &SessionContext,
) -> Result<CompressionResult, CompressionError> {
// 保留最近3轮对话（6条消息：3 user + 3 assistant）
let retain = 6.min(session.messages.len());
let split_point = session.messages.len() – retain;

let old_messages = &session.messages[..split_point];
let recent_messages = &session.messages[split_point..];

// 如果已有之前的摘要，将旧摘要 + 新消息一起压缩
let summary_input = if let Some(prev_summary) = &session.compressed_summary {
self.merge_summary(prev_summary, old_messages).await?
} else {
self.generate_detailed_summary(old_messages).await?
};

let mut compressed = vec![Message::system(summary_input)];
compressed.extend_from_slice(recent_messages);

Ok(CompressionResult::Compressed {
strategy: CompressionStrategy::Heavy,
messages: compressed,
tokens_saved: self.token_estimator.estimate_session(session)
– self.token_estimator.estimate_messages(&compressed),
})
}

/// 使用 LLM 生成对话摘要
async fn generate_summary(
&self,
messages: &[Message],
) -> Result<String, CompressionError> {
let prompt = format!(
r#"请将以下对话历史压缩为简洁的摘要，保留：
1. 用户的核心需求和目标
2. 已经做出的关键决策
3. 重要的上下文信息（人名、技术栈、约束条件等）
4. 未解决的问题和待办事项

对话历史：
{}

摘要："#,
messages.iter()
.map(|m| format!("[{}]: {}", m.role,
m.content.as_text().unwrap_or_default()))
.collect::<Vec<_>>()
.join("\\n")
);

let summary = self.llm_call(&self.summary_model, &prompt).await?;
Ok(format!("[对话历史摘要]\\n{}", summary))
}

/// 合并已有摘要与新消息（增量摘要）
async fn merge_summary(
&self,
prev_summary: &Summary,
new_messages: &[Message],
) -> Result<String, CompressionError> {
let prompt = format!(
r#"以下是之前的对话摘要和新增的对话内容，请合并为一个更新的摘要：

【已有摘要】
{}

【新增对话】
{}

【更新后的摘要】"#,
prev_summary.text,
new_messages.iter()
.map(|m| format!("[{}]: {}", m.role,
m.content.as_text().unwrap_or_default()))
.collect::<Vec<_>>()
.join("\\n")
);

let merged = self.llm_call(&self.summary_model, &prompt).await?;
Ok(format!("[对话历史摘要（更新）]\\n{}", merged))
}
}

2.3.3 压缩效果评估

以下是不同压缩策略在实测中的效果对比：

信息保留率的衡量方式：压缩前后，使用同一组测试问题评估 Agent 回答的准确率，以未压缩版本的准确率为基准计算保留率。

2.3.4 压缩策略的选择考量

压缩策略的选择并非简单的阈值触发，还需要综合考虑以下因素：

/// 压缩决策上下文
pub struct CompressionDecisionContext {
/// 当前 Token 使用量
pub current_tokens: u64,
/// 模型上下文窗口大小
pub context_window: u64,
/// 对话轮次
pub turn_count: u32,
/// 工具调用密度（最近10轮中工具调用的比例）
pub tool_call_density: f64,
/// 用户情绪指标（基于最近消息的语义分析）
pub user_sentiment: Sentiment,
/// 会话重要性评分
pub session_importance: f64,
/// 可用压缩预算（时间）
pub time_budget: Duration,
}

/// 智能压缩策略选择器
pub fn select_strategy(ctx: &CompressionDecisionContext) -> CompressionStrategy {
let ratio = ctx.current_tokens as f64 / ctx.context_window as f64;

// 如果时间预算紧张，选择最快的策略
if ctx.time_budget < Duration::from_millis(200) {
return if ratio > 0.9 {
CompressionStrategy::Medium
} else {
CompressionStrategy::Light
};
}

// 如果用户情绪消极（不耐烦），减少压缩延迟
if ctx.user_sentiment == Sentiment::Negative && ratio < 0.9 {
return CompressionStrategy::Light;
}

// 高重要性会话，使用更保守的压缩
if ctx.session_importance > 0.8 && ratio < 0.85 {
return CompressionStrategy::Light;
}

// 标准决策路径
if ratio < 0.5 {
CompressionStrategy::None
} else if ratio < 0.75 {
CompressionStrategy::Light
} else if ratio < 0.90 {
CompressionStrategy::Medium
} else {
CompressionStrategy::Heavy
}
}

2.4 并行会话隔离方案

2.4.1 隔离架构

Hermes Agent 支持同一用户同时持有多个活跃会话，每个会话拥有独立的上下文空间，互不干扰：

┌──────────────────────────────────────────────────────────────────────┐
│ 并行会话隔离架构 │
│ │
│ ┌──────────────┐ │
│ │ User A │ │
│ └──────┬───────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ │ │ │ │
│ Session #1 Session #2 Session #3 │
│ (飞书渠道) (Web渠道) (API渠道) │
│ │ │ │ │
│ ┌────────┴───┐ ┌────┴─────┐ ┌───┴────────┐ │
│ │ Context │ │ Context │ │ Context │ │
│ │ · messages │ │ · messages│ │ · messages │ │
│ │ · tools │ │ · tools │ │ · tools │ │
│ │ · metadata │ │ · metadata│ │ · metadata │ │
│ └────────────┘ └──────────┘ └────────────┘ │
│ │ │ │ │
│ └────────────┼────────────┘ │
│ │ │
│ ┌──────┴───────┐ │
│ │ 共享记忆层 │ │
│ │ (User Level) │ │
│ │ · 用户画像 │ │
│ │ · 长期记忆 │ │
│ │ · 偏好设置 │ │
│ └──────────────┘ │
└──────────────────────────────────────────────────────────────────────┘

会话隔离的核心设计原则：

• 上下文隔离：每个会话拥有独立的消息序列、工具调用栈和元数据，一个会话的上下文变更不会影响其他会话
• 记忆共享：用户级长期记忆（用户画像、偏好设置、历史决策）在所有会话间共享，但通过只读快照方式访问，避免并发写入冲突
• 资源隔离：每个会话有独立的 Token 配额、工具调用配额和速率限制，防止单个会话耗尽用户资源
• 显式共享：会话间的信息共享是显式操作而非自动行为，用户或系统必须明确指定要共享的消息和上下文

2.4.2 隔离实现

/// 会话隔离管理器
pub struct SessionIsolationManager {
/// 用户会话映射：UserId -> Set<SessionId>
user_sessions: Arc<RwLock<HashMap<UserId, HashSet<SessionId>>>>,
/// 会话上下文存储（按 SessionId 隔离）
session_contexts: Arc<SessionStore>,
/// 用户级共享记忆
user_memory: Arc<UserMemoryStore>,
/// 每用户最大并发会话数
max_sessions_per_user: usize,
}

impl SessionIsolationManager {
/// 创建新会话（检查并发限制）
pub async fn create_session(
&self,
user_id: &UserId,
channel: ChannelType,
) -> Result<SessionId, IsolationError> {
let mut user_sessions = self.user_sessions.write().await;

let active_count = user_sessions
.get(user_id)
.map(|s| s.len())
.unwrap_or(0);

if active_count >= self.max_sessions_per_user {
return Err(IsolationError::MaxSessionsExceeded {
user_id: user_id.clone(),
current: active_count,
max: self.max_sessions_per_user,
});
}

let session_id = SessionId::new(
self.get_machine_id(),
channel.as_u16(),
);

// 创建隔离的上下文空间
let context = SessionContext::new(
session_id.clone(),
user_id.clone(),
channel,
);

self.session_contexts.put(&session_id, &context).await?;

user_sessions
.entry(user_id.clone())
.or_insert_with(HashSet::new)
.insert(session_id.clone());

Ok(session_id)
}

/// 获取会话上下文（确保隔离性）
pub async fn get_context(
&self,
session_id: &SessionId,
) -> Result<SessionContext, IsolationError> {
// 直接从隔离存储获取，不经过用户级路由
self.session_contexts
.get(session_id)
.await?
.ok_or(IsolationError::SessionNotFound(session_id.clone()))
}

/// 获取用户级共享记忆（所有会话可读）
pub async fn get_user_memory(
&self,
user_id: &UserId,
) -> Result<UserMemory, IsolationError> {
self.user_memory.get(user_id).await
}

/// 会话间信息共享（显式操作，非自动）
pub async fn share_context(
&self,
from_session: &SessionId,
to_session: &SessionId,
message_ids: &[MessageId],
) -> Result<(), IsolationError> {
// 验证两个会话属于同一用户
let from_ctx = self.session_contexts.get(from_session).await?.unwrap();
let to_ctx = self.session_contexts.get(to_session).await?.unwrap();

if from_ctx.user_id != to_ctx.user_id {
return Err(IsolationError::CrossUserSharing {
from_user: from_ctx.user_id,
to_user: to_ctx.user_id,
});
}

// 仅复制指定消息到目标会话
let messages_to_share: Vec<Message> = from_ctx.messages
.iter()
.filter(|m| message_ids.contains(&m.message_id))
.cloned()
.collect();

let mut to_ctx = to_ctx;
for msg in messages_to_share {
// 添加来源标记
let shared_msg = Message {
metadata: MessageMetadata {
shared_from: Some(from_session.clone()),
..msg.metadata
},
..msg
};
to_ctx.messages.push(shared_msg);
}

self.session_contexts.put(to_session, &to_ctx).await?;
Ok(())
}
}

2.4.3 资源配额管理

为防止单个用户或会话消耗过多资源，Gateway 实现了多维度资源配额管理：

# 资源配额配置示例
quota:
per_user:
max_active_sessions: 5
max_requests_per_minute: 120
max_tokens_per_hour: 500000
max_tool_calls_per_day: 1000

per_session:
max_messages: 500
max_tokens: 100000
max_tool_calls: 50
max_duration: 24h

per_channel:
web:
max_concurrent_sessions_per_user: 3
feishu:
max_concurrent_sessions_per_user: 2
api:
max_concurrent_sessions_per_user: 10
max_rps: 50

enforcement:
strategy: reject # reject | queue | degrade
queue_timeout: 30s
degrade_message: "当前请求量较大，已切换到简化模式"

配额管理的执行策略有三种模式：

• reject（拒绝）：超出配额的请求直接返回 429 错误，客户端需要自行处理重试
• queue（排队）：超出配额的请求进入等待队列，在超时时间内有配额释放则继续处理
• degrade（降级）：超出配额后自动降级服务——使用更小的模型、减少工具调用、跳过记忆检索等

2.5 会话生命周期管理

2.5.1 生命周期状态机

┌───────────┐
创建 ────> │ Active │ <──── 恢复
└─────┬─────┘
│
┌───────────┼───────────┐
│ │ │
v v v
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Idle │ │ Paused │ │ Migrating│
│ (超时等待) │ │ (用户暂停) │ │ (节点迁移) │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
│ 恢复 <───┘ 完成 ───┘
│ │ │
v v v
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Expiring │ │ Active │ │ Active │
│ (即将过期) │ │ │ │ │
└─────┬────┘ └──────────┘ └──────────┘
│
v
┌──────────┐
│ Archived │
│ (已归档) │
└─────┬────┘
│
v
┌──────────┐
│ Deleted │
│ (已删除) │
└──────────┘

/// 会话生命周期状态
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum SessionStatus {
/// 活跃状态，正在处理请求
Active,
/// 空闲状态，等待用户输入
Idle { idle_since: DateTime<Utc> },
/// 暂停状态，用户主动暂停
Paused { paused_at: DateTime<Utc> },
/// 迁移中状态，正在跨节点迁移
Migrating { from_node: NodeId, to_node: NodeId },
/// 即将过期
Expiring { expire_at: DateTime<Utc> },
/// 已归档（从热存储转移到冷存储）
Archived,
/// 已删除
Deleted,
}

/// 会话生命周期管理器
pub struct SessionLifecycleManager {
/// 空闲超时（自动转为 Idle）
idle_timeout: Duration,
/// 过期时间（Idle 后多久转为 Expiring）
expire_timeout: Duration,
/// 归档时间（Expiring 后多久转为 Archived）
archive_timeout: Duration,
/// 删除时间（Archived 后多久物理删除）
delete_timeout: Duration,
/// 热存储
hot_store: Arc<HotSessionStore>,
/// 冷存储
cold_store: Arc<ColdSessionStore>,
}

impl SessionLifecycleManager {
/// 定时扫描会话状态，执行状态转换
pub async fn run_lifecycle_scan(&self) {
let mut interval = tokio::time::interval(Duration::from_secs(30));

loop {
interval.tick().await;

// 1. 扫描 Active 会话，检查是否空闲
let idle_sessions = self.hot_store
.find_sessions_idle_before(Utc::now() – self.idle_timeout)
.await;
for session_id in idle_sessions {
self.transition_to_idle(&session_id).await;
}

// 2. 扫描 Idle 会话，检查是否过期
let expiring_sessions = self.hot_store
.find_sessions_idle_before(Utc::now() – self.expire_timeout)
.await;
for session_id in expiring_sessions {
self.transition_to_expiring(&session_id).await;
}

// 3. 扫描 Expiring 会话，执行归档
let archive_sessions = self.hot_store
.find_sessions_expiring_before(Utc::now())
.await;
for session_id in archive_sessions {
self.archive_session(&session_id).await;
}

// 4. 扫描 Archived 会话，执行物理删除
let delete_sessions = self.cold_store
.find_sessions_archived_before(Utc::now() – self.delete_timeout)
.await;
for session_id in delete_sessions {
self.delete_session(&session_id).await;
}
}
}

/// 恢复已归档的会话
pub async fn restore_session(
&self,
session_id: &SessionId,
) -> Result<SessionContext, LifecycleError> {
// 1. 从冷存储读取会话数据
let session = self.cold_store
.get(session_id)
.await?
.ok_or(LifecycleError::SessionNotFound)?;

// 2. 写入热存储
self.hot_store.put(session_id, &session).await?;

// 3. 更新状态为 Active
let mut session = session;
session.status = SessionStatus::Active;
session.last_active_at = Utc::now();

Ok(session)
}
}

2.5.2 冷热分离存储策略

会话数据在不同生命周期阶段存储在不同介质中，以平衡访问速度和存储成本：

冷热分离的关键设计点：

• 异步迁移：会话从热存储到冷存储的迁移是异步操作，不阻塞主流程
• 透明恢复：当用户恢复已归档会话时，Gateway 自动从冷存储加载数据到热存储，对用户透明
• 预加载：系统可以基于用户行为预测，提前将可能恢复的归档会话预加载到热存储
• 渐进式删除：删除操作先标记为软删除，保留 24 小时的恢复窗口后才执行物理删除

第三章 渠道适配器架构

渠道适配器是 Gateway 与外部世界沟通的桥梁。它将各平台千差万别的消息协议统一翻译为 Gateway 内部标准格式，使得核心处理逻辑完全无需感知渠道差异。这一层的设计质量直接决定了系统的可扩展性和多渠道接入效率。

3.1 适配器模式设计

3.1.1 核心接口定义

Hermes Agent 定义了一套统一的渠道适配器 SPI（Service Provider Interface），所有渠道适配器必须实现该接口：

/// 渠道适配器核心接口
#[async_trait]
pub trait ChannelAdapter: Send + Sync {
/// 适配器名称
fn name(&self) -> &str;

/// 支持的渠道类型
fn channel_type(&self) -> ChannelType;

/// 初始化适配器
async fn initialize(&mut self, config: &AdapterConfig)
-> Result<(), AdapterError>;

/// 启动适配器（开始接收消息）
async fn start(&mut self) -> Result<(), AdapterError>;

/// 停止适配器
async fn stop(&mut self) -> Result<(), AdapterError>;

// ========== 入站消息处理 ==========

/// 接收外部消息，转换为内部标准格式
async fn receive_message(
&self,
raw_message: RawMessage,
) -> Result<InboundMessage, AdapterError>;

/// 验证消息来源合法性（签名校验等）
async fn verify_message(
&self,
raw_message: &RawMessage,
) -> Result<bool, AdapterError>;

// ========== 出站消息处理 ==========

/// 将内部标准格式消息转换为目标渠道格式
async fn send_message(
&self,
message: OutboundMessage,
) -> Result<DeliveryReceipt, AdapterError>;

/// 发送流式消息（SSE/WebSocket）
async fn send_stream(
&self,
stream: BoxStream<'static, StreamChunk>,
target: &str,
) -> Result<(), AdapterError>;

// ========== 渠道能力查询 ==========

/// 查询渠道支持的能力
fn capabilities(&self) -> ChannelCapabilities;

/// 健康检查
async fn health_check(&self) -> Result<HealthStatus, AdapterError>;
}

渠道能力描述结构体定义了每个平台支持的功能特性，适配器在初始化时声明自身能力，Gateway 据此做能力感知路由：

/// 渠道能力描述
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ChannelCapabilities {
/// 支持文本消息
pub text: bool,
/// 支持图片消息
pub image: bool,
/// 支持文件消息
pub file: bool,
/// 支持音频消息
pub audio: bool,
/// 支持视频消息
pub video: bool,
/// 支持流式输出
pub streaming: bool,
/// 支持卡片消息
pub card: bool,
/// 支持按钮交互
pub buttons: bool,
/// 支持Markdown
pub markdown: bool,
/// 最大消息长度
pub max_message_length: usize,
/// 支持消息编辑
pub edit: bool,
/// 支持消息删除
pub delete: bool,
/// 支持消息引用回复
pub reply: bool,
/// 支持At提及
pub mention: bool,
}

3.1.2 适配器注册与加载

适配器通过工厂模式创建实例，工厂在 Gateway 启动时自动注册：

/// 适配器注册中心
pub struct AdapterRegistry {
/// 已注册的适配器工厂
factories: RwLock<HashMap<ChannelType, Box<dyn AdapterFactory>>>,
/// 活跃的适配器实例
instances: RwLock<HashMap<ChannelType, Arc<dyn ChannelAdapter>>>,
/// 配置中心
config_source: Arc<dyn ConfigSource>,
}

#[async_trait]
pub trait AdapterFactory: Send + Sync {
/// 创建适配器实例
fn create(&self) -> Box<dyn ChannelAdapter>;
/// 适配器类型标识
fn channel_type(&self) -> ChannelType;
}

impl AdapterRegistry {
/// 注册适配器工厂
pub async fn register_factory(
&self,
factory: Box<dyn AdapterFactory>,
) -> Result<(), RegistryError> {
let channel_type = factory.channel_type();
let mut factories = self.factories.write().await;
factories.insert(channel_type, factory);
tracing::info!(channel = ?channel_type, "Adapter factory registered");
Ok(())
}

/// 获取或创建适配器实例
pub async fn get_or_create(
&self,
channel_type: &ChannelType,
) -> Result<Arc<dyn ChannelAdapter>, RegistryError> {
// 先检查是否有活跃实例
{
let instances = self.instances.read().await;
if let Some(instance) = instances.get(channel_type) {
return Ok(instance.clone());
}
}

// 创建新实例
let factories = self.factories.read().await;
let factory = factories.get(channel_type)
.ok_or(RegistryError::FactoryNotFound(*channel_type))?;

let mut adapter = factory.create();
let config = self.config_source
.get_adapter_config(channel_type)
.await?;

adapter.initialize(&config).await?;
adapter.start().await?;

let adapter: Arc<dyn ChannelAdapter> = Arc::from(adapter);
let mut instances = self.instances.write().await;
instances.insert(*channel_type, adapter.clone());

Ok(adapter)
}

/// 优雅停止指定渠道的适配器
pub async fn stop_adapter(
&self,
channel_type: &ChannelType,
) -> Result<(), RegistryError> {
let mut instances = self.instances.write().await;
if let Some(adapter) = instances.remove(channel_type) {
// 等待正在处理的消息完成
adapter.stop().await?;
tracing::info!(channel = ?channel_type, "Adapter stopped gracefully");
}
Ok(())
}
}

3.1.3 适配器实现示例：飞书适配器

以下是一个完整的飞书渠道适配器实现示例，展示了如何将飞书的事件回调机制适配为 Gateway 标准接口：

/// 飞书适配器工厂
pub struct FeishuAdapterFactory;

#[async_trait]
impl AdapterFactory for FeishuAdapterFactory {
fn create(&self) -> Box<dyn ChannelAdapter> {
Box::new(FeishuAdapter::new())
}
fn channel_type(&self) -> ChannelType {
ChannelType::Feishu
}
}

/// 飞书渠道适配器
pub struct FeishuAdapter {
app_id: String,
app_secret: String,
encrypt_key: Option<String>,
verification_token: String,
http_client: reqwest::Client,
tenant_access_token: Arc<RwLock<Option<(String, Instant)>>>,
}

#[async_trait]
impl ChannelAdapter for FeishuAdapter {
fn name(&self) -> &str { "feishu" }
fn channel_type(&self) -> ChannelType { ChannelType::Feishu }

async fn initialize(&mut self, config: &AdapterConfig)
-> Result<(), AdapterError>
{
self.app_id = config.get("app_id")?;
self.app_secret = config.get("app_secret")?;
self.encrypt_key = config.get_optional("encrypt_key");
self.verification_token = config.get("verification_token")?;
Ok(())
}

async fn receive_message(
&self,
raw: RawMessage,
) -> Result<InboundMessage, AdapterError> {
let feishu_event: FeishuEvent = serde_json::from_slice(&raw.body)
.map_err(|e| AdapterError::ParseError(e.to_string()))?;

match feishu_event.event_type.as_str() {
"im.message.receive_v1" => {
let msg_data = &feishu_event.event.message;

// 解析消息内容（飞书消息内容为JSON字符串）
let content: FeishuMessageContent = serde_json::from_str(
&msg_data.content
).unwrap_or_default();

let normalized = NormalizedContent {
text: content.text.or_else(|| {
serde_json::from_str::<serde_json::Value>(&msg_data.content)
.ok()
.and_then(|v| v.get("text")
.and_then(|t| t.as_str())
.map(String::from))
}),
attachments: self.parse_feishu_attachments(msg_data)
.unwrap_or_default(),
mentions: self.parse_mentions(&feishu_event.event)
.unwrap_or_default(),
reply_to: msg_data.parent_id.clone(),
};

Ok(InboundMessage {
source_message_id: msg_data.message_id.clone(),
session_id: None,
sender: SenderInfo {
user_id: feishu_event.event.sender.sender_id.open_id,
user_name: None,
avatar: None,
},
content: normalized,
metadata: InboundMetadata {
channel: ChannelType::Feishu,
chat_id: msg_data.chat_id.clone(),
chat_type: self.parse_chat_type(&msg_data.chat_type),
timestamp: msg_data.create_time.parse().unwrap_or(0),
raw_ref: raw.id,
},
})
}
_ => Err(AdapterError::UnsupportedEventType(
feishu_event.event_type
))
}
}

async fn send_message(
&self,
message: OutboundMessage,
) -> Result<DeliveryReceipt, AdapterError> {
let token = self.get_tenant_access_token().await?;

// 根据渲染提示选择消息格式
let (msg_type, content) = match message.render_hint.format {
MessageFormat::Markdown => {
("text", serde_json::json!({
"text": message.content.text.unwrap_or_default()
}).to_string())
}
MessageFormat::Interactive => {
("interactive",
self.build_feishu_card(&message).await?.to_string())
}
_ => {
("text", serde_json::json!({
"text": message.content.text.unwrap_or_default()
}).to_string())
}
};

let req_body = serde_json::json!({
"receive_id": message.target,
"msg_type": msg_type,
"content": content,
});

let resp = self.http_client
.post("https://open.feishu.cn/open-apis/im/v1/messages")
.header("Authorization", format!("Bearer {}", token))
.header("Content-Type", "application/json")
.query(&[("receive_id_type", "chat_id")])
.json(&req_body)
.send()
.await
.map_err(|e| AdapterError::SendError(e.to_string()))?;

let resp_body: serde_json::Value = resp.json().await
.map_err(|e| AdapterError::SendError(e.to_string()))?;

let message_id = resp_body
.get("data")
.and_then(|d| d.get("message_id"))
.and_then(|m| m.as_str())
.unwrap_or("")
.to_string();

Ok(DeliveryReceipt {
message_id,
status: DeliveryStatus::Sent,
timestamp: Utc::now(),
})
}

fn capabilities(&self) -> ChannelCapabilities {
ChannelCapabilities {
text: true,
image: true,
file: true,
audio: false,
video: false,
streaming: false,
card: true,
buttons: true,
markdown: true,
max_message_length: 30000,
edit: true,
delete: true,
reply: true,
mention: true,
}
}

async fn health_check(&self) -> Result<HealthStatus, AdapterError> {
// 尝试获取 tenant_access_token 作为健康检查
match self.get_tenant_access_token().await {
Ok(_) => Ok(HealthStatus::Healthy),
Err(e) => Ok(HealthStatus::Unhealthy {
reason: e.to_string(),
since: Utc::now(),
}),
}
}
}

impl FeishuAdapter {
/// 获取或刷新 tenant_access_token
async fn get_tenant_access_token(&self) -> Result<String, AdapterError> {
// 检缓存的 token 是否有效
{
let cache = self.tenant_access_token.read().await;
if let Some((token, expiry)) = cache.as_ref() {
if expiry > &Instant::now() {
return Ok(token.clone());
}
}
}

// 申请新 token
let resp = self.http_client
.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal")
.json(&serde_json::json!({
"app_id": self.app_id,
"app_secret": self.app_secret,
}))
.send()
.await
.map_err(|e| AdapterError::AuthError(e.to_string()))?;

let body: serde_json::Value = resp.json().await
.map_err(|e| AdapterError::AuthError(e.to_string()))?;

let token = body.get("tenant_access_token")
.and_then(|t| t.as_str())
.ok_or(AdapterError::AuthError("Missing token".into()))?
.to_string();

let expire = body.get("expire")
.and_then(|e| e.as_u64())
.unwrap_or(7200);

// 缓存 token（提前60秒过期）
let mut cache = self.tenant_access_token.write().await;
*cache = Some((
token.clone(),
Instant::now() + Duration::from_secs(expire – 60),
));

Ok(token)
}

/// 构建飞书交互卡片
async fn build_feishu_card(
&self,
message: &OutboundMessage,
) -> Result<serde_json::Value, AdapterError> {
let text = message.content.text.as_deref().unwrap_or("");

Ok(serde_json::json!({
"config": {
"wide_screen_mode": true
},
"elements": [
{
"tag": "div",
"text": {
"content": text,
"tag": "lark_md"
}
}
]
}))
}
}

3.2 消息格式归一化流程

3.2.1 归一化管线

消息从外部渠道进入 Gateway 后，需要经过一条归一化管线，将异构格式统一为内部标准格式：

┌──────────────────────────────────────────────────────────────────────┐
│ 消息格式归一化管线 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 原始消息 │──>│ 安全验证 │──>│ 格式解析 │──>│ 内容归一化│ │
│ │ (Raw) │ │ (Verify) │ │ (Parse) │ │ (Normalize)│ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │
│ v │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 消息分发 │<──│ 元数据 │<──│ 内容增强 │<──│ 标准消息 │ │
│ │ (Dispatch)│ │ (Enrich) │ │ (Enhance)│ │ (Standard)│ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │
│ 归一化步骤详解: │
│ │
│ 1. 安全验证: 签名校验 → 频率限制 → 内容安全检查 │
│ 2. 格式解析: JSON/XML/Form 解析 → 渠道特定结构提取 │
│ 3. 内容归一化: 文本提取 → 附件下载 → 多模态转换 │
│ 4. 内容增强: 语言检测 → 意图预判 → 上下文关联 │
│ 5. 元数据填充: 时间戳标准化 → 用户信息补全 → 会话映射 │
│ 6. 消息分发: 路由到会话管理器 → 进入处理管线 │
└──────────────────────────────────────────────────────────────────────┘

3.2.2 归一化处理器实现

归一化管线采用责任链模式，每个阶段独立处理并传递给下一阶段：

/// 消息归一化管线
pub struct NormalizationPipeline {
stages: Vec<Box<dyn NormalizationStage>>,
}

#[async_trait]
pub trait NormalizationStage: Send + Sync {
fn name(&self) -> &str;
async fn process(
&self,
message: &mut PipelineMessage,
) -> Result<(), PipelineError>;
}

/// 归一化管线消息（在管线各阶段间传递）
pub struct PipelineMessage {
pub raw: RawMessage,
pub verified: bool,
pub parsed: Option<ParsedMessage>,
pub normalized: Option<NormalizedContent>,
pub enriched: Option<EnrichedContent>,
pub inbound: Option<InboundMessage>,
}

impl NormalizationPipeline {
pub async fn execute(
&self,
raw: RawMessage,
) -> Result<InboundMessage, PipelineError> {
let mut message = PipelineMessage {
raw,
verified: false,
parsed: None,
normalized: None,
enriched: None,
inbound: None,
};

for stage in &self.stages {
stage.process(&mut message).await.map_err(|e| {
tracing::error!(
stage = stage.name(),
error = %e,
"Pipeline stage failed"
);
e
})?;
}

message.inbound.ok_or(PipelineError::IncompletePipeline)
}
}

以下是两个关键阶段的实现：

/// 阶段1：安全验证
pub struct SecurityVerificationStage {
rate_limiter: Arc<RateLimiter>,
content_safety: Arc<ContentSafetyChecker>,
}

#[async_trait]
impl NormalizationStage for SecurityVerificationStage {
fn name(&self) -> &str { "security_verification" }

async fn process(&self, msg: &mut PipelineMessage)
-> Result<(), PipelineError>
{
// 频率限制检查
self.rate_limiter.check(&msg.raw.source).await?;

// 内容安全检查（敏感词、恶意链接等）
if let Some(text) = &msg.raw.text_preview {
self.content_safety.check(text).await?;
}

msg.verified = true;
Ok(())
}
}

/// 阶段3：内容归一化
pub struct ContentNormalizationStage {
attachment_downloader: Arc<AttachmentDownloader>,
media_processor: Arc<MediaProcessor>,
}

#[async_trait]
impl NormalizationStage for ContentNormalizationStage {
fn name(&self) -> &str { "content_normalization" }

async fn process(&self, msg: &mut PipelineMessage)
-> Result<(), PipelineError>
{
let parsed = msg.parsed.as_ref().unwrap();
let mut attachments = Vec::new();

for att in &parsed.attachments {
match att.att_type {
AttachmentType::Image => {
let downloaded = self.attachment_downloader
.download(&att.url).await?;
let processed = self.media_processor
.process_image(&downloaded).await?;
attachments.push(NormalizedAttachment {
att_type: AttachmentType::Image,
url: processed.url,
thumbnail_url: processed.thumbnail_url,
metadata: Some(serde_json::json!({
"width": processed.width,
"height": processed.height,
"size": processed.size,
})),
});
}
AttachmentType::File => {
let downloaded = self.attachment_downloader
.download(&att.url).await?;
attachments.push(NormalizedAttachment {
att_type: AttachmentType::File,
url: downloaded.url,
thumbnail_url: None,
metadata: Some(serde_json::json!({
"filename": att.filename,
"size": att.size,
"mime_type": att.mime_type,
})),
});
}
_ => {
attachments.push(NormalizedAttachment {
att_type: att.att_type,
url: att.url.clone(),
thumbnail_url: None,
metadata: None,
});
}
}
}

msg.normalized = Some(NormalizedContent {
text: parsed.text.clone(),
attachments,
mentions: parsed.mentions.clone(),
reply_to: parsed.reply_to.clone(),
});

Ok(())
}
}

3.2.3 内容增强阶段

内容增强是归一化管线中最具智能性的阶段，它在消息进入核心处理之前完成预分析：

/// 阶段4：内容增强
pub struct ContentEnhancementStage {
language_detector: Arc<LanguageDetector>,
intent_predictor: Arc<IntentPredictor>,
session_resolver: Arc<SessionResolver>,
}