Agent 心跳与健康检查：长连接场景下的会话状态监控

Agent 连着连着就没了反应——你不知道它是真的在思考，还是已经悄悄挂了。

一、场景痛点

你的 Agent 系统用 WebSocket 维持长连接，用户发一条消息后 Agent 需要调用多个工具，耗时可能 10-60 秒。问题来了：这 60 秒里，用户不知道 Agent 是在处理还是已经挂了。如果 Agent 进程崩溃，WebSocket 连接不会自动断开——客户端一直等着，等到超时才报错，但此时用户已经以为系统卡死了。

你加了一个进度条，每 5 秒推送一条"还在处理中"的消息。但 Agent 真的挂了的时候，进度条还在转——因为推送线程和 Agent 执行线程是独立的，Agent 挂了推送线程还在跑。

更棘手的是会话恢复。Agent 处理到第 3 步时崩溃了，用户重连后从第 1 步重新开始——前两步的结果全部丢失。如果是付费场景（每次调用消耗 token），重复执行的成本直接翻倍。

核心矛盾：长连接场景下，Agent 的存活状态和处理进度必须被持续监控，不能靠"连接还在就以为还活着"。

二、底层机制与原理剖析

2.1 心跳机制的层次

2.2 心跳数据模型

进程心跳不只是"我还活着"，它包含处理状态：

字段	含义	用途
agent_id	Agent 实例标识	关联会话与实例
session_id	会话标识	恢复会话时使用
status	idle/processing/error	区分状态
current_step	当前执行步骤	进度追踪
total_steps	总步骤数	进度百分比计算
cpu_percent	CPU 使用率	资源监控
memory_mb	内存占用	资源监控
last_tool_call	最近一次工具调用信息	卡住定位
timestamp	心跳时间戳	判断是否过期

2.3 健康检查的判定逻辑

健康判定不是简单的"心跳在就健康"。需要根据心跳间隔和业务状态综合判断：

• 健康：心跳间隔 < 预期间隔 × 2，status 不是 error
• 亚健康：心跳间隔 > 预期间隔 × 2 但 < 预期间隔 × 5，status 是 processing 但 current_step 长时间不变
• 不健康：心跳间隔 > 预期间隔 × 5，或 status 是 error，或连续 3 次心跳缺失

三、生产级代码实现

3.1 Agent 心跳上报器

// agent-heartbeat.ts —— Agent 进程心跳上报器
import { EventEmitter } from 'events';

export enum AgentStatus {
  IDLE = 'idle',        // 等待用户输入
  PROCESSING = 'processing',  // 正在处理用户请求
  ERROR = 'error',      // 处理出错，等待恢复
  TERMINATING = 'terminating',  // 正在优雅关闭
}

export interface HeartbeatPayload {
  agent_id: string;
  session_id: string;
  status: AgentStatus;
  current_step: number;
  total_steps: number;
  cpu_percent: number;
  memory_mb: number;
  last_tool_call: string | null;
  timestamp: number;  // Unix 时间戳（毫秒）
}

export class AgentHeartbeat extends EventEmitter {
  private agentId: string;
  private sessionId: string;
  private status: AgentStatus = AgentStatus.IDLE;
  private currentStep: number = 0;
  private totalSteps: number = 0;
  private lastToolCall: string | null = null;

  private intervalMs: number;  // 心跳间隔
  private maxMissedHeartbeats: number;  // 允许缺失的最大心跳数
  private heartbeatTimer: NodeJS.Timeout | null = null;

  // 心跳发送通道：WebSocket / HTTP / 消息队列
  private sender: (payload: HeartbeatPayload) => Promise<void>;

  constructor(
    agentId: string,
    sessionId: string,
    intervalMs: number = 5000,  // 默认 5 秒心跳间隔
    maxMissedHeartbeats: number = 3,
    sender: (payload: HeartbeatPayload) => Promise<void>
  ) {
    super();
    this.agentId = agentId;
    this.sessionId = sessionId;
    this.intervalMs = intervalMs;
    this.maxMissedHeartbeats = maxMissedHeartbeats;
    this.sender = sender;
  }

  /** 启动心跳循环：定时上报状态 */
  start(): void {
    if (this.heartbeatTimer) return;  // 已启动则不重复启动

    // 定时发送心跳：intervalMs 间隔
    // 心跳是"推"模式，不是"拉"模式——服务端不需要轮询检查 Agent 状态
    this.heartbeatTimer = setInterval(() => {
      this.sendHeartbeat();
    }, this.intervalMs);

    // 立即发送一次心跳：启动时让服务端知道 Agent 已上线
    this.sendHeartbeat();
  }

  /** 停止心跳循环：优雅关闭前调用 */
  stop(): void {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }

    // 发送最终心跳：标记为 terminating，服务端知道 Agent 正在关闭
    this.status = AgentStatus.TERMINATING;
    this.sendHeartbeat();
  }

  /** 更新处理状态：Agent 每完成一步调用此方法 */
  updateProgress(currentStep: number, totalSteps: number, toolCall: string): void {
    this.currentStep = currentStep;
    this.totalSteps = totalSteps;
    this.lastToolCall = toolCall;
    this.status = AgentStatus.PROCESSING;

    // 状态变化时立即发送一次心跳（不等定时器）
    // 用户在等待结果，状态变化应该第一时间告知服务端
    this.sendHeartbeat();
  }

  /** 标记错误状态 */
  markError(): void {
    this.status = AgentStatus.ERROR;
    this.sendHeartbeat();
  }

  /** 标记空闲状态 */
  markIdle(): void {
    this.status = AgentStatus.IDLE;
    this.currentStep = 0;
    this.totalSteps = 0;
    this.lastToolCall = null;
    this.sendHeartbeat();
  }

  /** 发送心跳：组装 payload 并调用 sender */
  private async sendHeartbeat(): void {
    const payload: HeartbeatPayload = {
      agent_id: this.agentId,
      session_id: this.sessionId,
      status: this.status,
      current_step: this.currentStep,
      total_steps: this.totalSteps,
      cpu_percent: this.getCpuUsage(),
      memory_mb: this.getMemoryUsage(),
      last_tool_call: this.lastToolCall,
      timestamp: Date.now(),
    };

    try {
      await this.sender(payload);
      this.emit('heartbeat:sent', payload);
    } catch (err) {
      // 心跳发送失败：不中断 Agent 处理流程
      // 心跳是辅助功能，核心业务不能因为心跳通道故障而停止
      this.emit('heartbeat:failed', { error: err, payload });
    }
  }

  /** 获取 CPU 使用率：简化实现，生产环境用 process.cpuUsage() */
  private getCpuUsage(): number {
    // Node.js 的 process.cpuUsage() 返回微秒级的 CPU 时间
    const usage = process.cpuUsage();
    const totalUsec = usage.user + usage.system;
    // 转换为百分比（近似值，需要采样间隔才能精确计算）
    return Math.min(totalUsec / 1000 / this.intervalMs, 100);
  }

  /** 获取内存使用量 */
  private getMemoryUsage(): number {
    return process.memoryUsage().heapUsed / 1024 / 1024;  // MB
  }
}

3.2 服务端健康检查监控器

// health-monitor.ts —— 服务端 Agent 健康检查监控器
// 监控所有 Agent 实例的心跳，判断健康状态，触发告警和会话恢复

export enum HealthStatus {
  HEALTHY = 'healthy',
  DEGRADED = 'degraded',
  UNHEALTHY = 'unhealthy',
  DEAD = 'dead',
}

interface AgentHealthRecord {
  agentId: string;
  sessionId: string;
  lastHeartbeat: HeartbeatPayload;
  lastHeartbeatTime: number;
  missedHeartbeats: number;
  healthStatus: HealthStatus;
  // 停滞检测：current_step 连续 N 次心跳未变化
  stagnantCount: number;
}

export class AgentHealthMonitor extends EventEmitter {
  private agents: Map<string, AgentHealthRecord> = new Map();
  private intervalMs: number;
  private maxMissed: number;
  private stagnantThreshold: number;  // 心跳停滞阈值：step 不变的次数
  private checkTimer: NodeJS.Timeout | null = null;

  constructor(
    intervalMs: number = 10000,  // 每 10 秒检查一次所有 Agent
    maxMissed: number = 3,
    stagnantThreshold: number = 5  // 5 次心跳 step 不变判定为停滞
  ) {
    super();
    this.intervalMs = intervalMs;
    this.maxMissed = maxMissed;
    this.stagnantThreshold = stagnantThreshold;
  }

  /** 接收 Agent 心跳：更新健康记录 */
  receiveHeartbeat(payload: HeartbeatPayload): void {
    const existing = this.agents.get(payload.agent_id);

    if (existing) {
      // 检查 current_step 是否变化：停滞检测
      if (payload.current_step === existing.lastHeartbeat.current_step
          && payload.status === AgentStatus.PROCESSING) {
        existing.stagnantCount++;
      } else {
        existing.stagnantCount = 0;
      }

      // 更新记录
      existing.lastHeartbeat = payload;
      existing.lastHeartbeatTime = payload.timestamp;
      existing.missedHeartbeats = 0;

      // 重新评估健康状态
      this.evaluateHealth(existing);
    } else {
      // 新 Agent 上线：初始化健康记录
      this.agents.set(payload.agent_id, {
        agentId: payload.agent_id,
        sessionId: payload.session_id,
        lastHeartbeat: payload,
        lastHeartbeatTime: payload.timestamp,
        missedHeartbeats: 0,
        healthStatus: HealthStatus.HEALTHY,
        stagnantCount: 0,
      });
      this.emit('agent:registered', { agentId: payload.agent_id });
    }
  }

  /** 启动健康检查循环 */
  start(): void {
    this.checkTimer = setInterval(() => {
      this.checkAllAgents();
    }, this.intervalMs);
  }

  /** 检查所有 Agent 的健康状态 */
  private checkAllAgents(): void {
    const now = Date.now();
    const expectedInterval = 5000;  // Agent 心跳间隔

    for (const [agentId, record] of this.agents) {
      const elapsed = now - record.lastHeartbeatTime;

      // 心跳缺失检测：超过预期间隔则计数 +1
      if (elapsed > expectedInterval * 2) {
        record.missedHeartbeats++;
      }

      // 停滞检测：step 不变的次数超过阈值
      if (record.stagnantCount >= this.stagnantThreshold) {
        // 工具调用卡住：Agent 还活着但处理停滞
        this.emit('agent:stagnant', {
          agentId,
          sessionId: record.sessionId,
          currentStep: record.lastHeartbeat.current_step,
          lastToolCall: record.lastHeartbeat.last_tool_call,
        });
      }

      // 重新评估健康状态
      this.evaluateHealth(record);

      // 不健康或死亡：触发告警
      if (record.healthStatus === HealthStatus.UNHEALTHY) {
        this.emit('agent:unhealthy', {
          agentId,
          sessionId: record.sessionId,
          missedHeartbeats: record.missedHeartbeats,
        });
      }

      if (record.healthStatus === HealthStatus.DEAD) {
        this.emit('agent:dead', {
          agentId,
          sessionId: record.sessionId,
        });

        // 死亡 Agent 从监控列表移除：不再等待心跳
        // 但会话状态保留，用于后续恢复
        this.agents.delete(agentId);
      }
    }
  }

  /** 评估单个 Agent 的健康状态 */
  private evaluateHealth(record: AgentHealthRecord): void {
    const previousStatus = record.healthStatus;

    if (record.missedHeartbeats >= this.maxMissed * 2) {
      // 连续缺失超过 2 倍阈值：判定死亡
      record.healthStatus = HealthStatus.DEAD;
    } else if (record.missedHeartbeats >= this.maxMissed) {
      // 连续缺失超过阈值：判定不健康
      record.healthStatus = HealthStatus.UNHEALTHY;
    } else if (record.missedHeartbeats > 0 || record.stagnantCount >= this.stagnantThreshold) {
      // 有缺失但未超阈值，或处理停滞：亚健康
      record.healthStatus = HealthStatus.DEGRADED;
    } else {
      // 正常心跳且处理推进中：健康
      record.healthStatus = HealthStatus.HEALTHY;
    }

    // 状态变化时发出事件：外部系统可以订阅做自动恢复
    if (previousStatus !== record.healthStatus) {
      this.emit('health:changed', {
        agentId: record.agentId,
        from: previousStatus,
        to: record.healthStatus,
      });
    }
  }
}

3.3 会话恢复与断点续传

# session_recovery.py —— Agent 崩溃后的会话恢复与断点续传
import json
import logging
import time
from datetime import datetime

logger = logging.getLogger('session-recovery')

class SessionRecoveryManager:
    """会话恢复管理器：Agent 崩溃后从断点继续处理"""

    def __init__(self, storage_client, heartbeat_monitor):
        self.storage = storage_client
        self.monitor = heartbeat_monitor

    def save_checkpoint(self, session_id: str, step_index: int, step_results: dict):
        """保存检查点：每完成一步就保存，崩溃后从检查点恢复"""
        checkpoint = {
            'session_id': session_id,
            'step_index': step_index,
            'step_results': step_results,
            'timestamp': datetime.utcnow().isoformat(),
        }
        # 检查点存到对象存储：比数据库更快，且不影响业务表
        key = f"checkpoints/{session_id}/step_{step_index}.json"
        self.storage.put(key, json.dumps(checkpoint))

    def recover_session(self, session_id: str) -> dict:
        """从最新检查点恢复会话"""
        # 查找该会话的所有检查点，取最新的
        pattern = f"checkpoints/{session_id}/step_*.json"
        checkpoints = self.storage.list(pattern)

        if not checkpoints:
            logger.warning(f"No checkpoints found for session {session_id}")
            return {'step_index': 0, 'step_results': {}}

        # 取最新的检查点（step_index 最大的）
        latest = max(checkpoints, key=lambda k: int(k.split('step_')[1].split('.')[0]))
        checkpoint_data = self.storage.get(latest)

        checkpoint = json.loads(checkpoint_data)
        logger.info(
            f"Recovered session {session_id} from step {checkpoint['step_index']}"
        )
        return checkpoint

    def handle_dead_agent(self, agent_id: str, session_id: str):
        """处理死亡 Agent：恢复会话并分配新 Agent"""
        # 1. 从检查点恢复会话状态
        checkpoint = self.recover_session(session_id)

        # 2. 创建新 Agent 实例，传入恢复的检查点
        # 新 Agent 从断点继续，不从第 0 步重新开始
        new_agent = self.create_agent_with_checkpoint(
            session_id, checkpoint
        )

        # 3. 通知用户：会话恢复，从第 N 步继续
        logger.info(
            f"Session {session_id} recovered: "
            f"new agent {new_agent.agent_id}, "
            f"resuming from step {checkpoint['step_index']}"
        )

        # 4. 清理旧 Agent 的残留资源（内存中的会话数据等）
        self.cleanup_agent_resources(agent_id)

        return new_agent

    def create_agent_with_checkpoint(self, session_id: str, checkpoint: dict):
        """创建新 Agent 并注入检查点数据"""
        # 新 Agent 启动时接收检查点，
        # 从 checkpoint['step_index'] + 1 开始执行
        # 前面步骤的结果从 checkpoint['step_results'] 中读取
        agent_config = {
            'session_id': session_id,
            'resume_from_step': checkpoint['step_index'] + 1,
            'previous_results': checkpoint['step_results'],
        }
        # 调用 Agent 启动接口
        return self.start_new_agent(agent_config)

    def cleanup_agent_resources(self, agent_id: str):
        """清理死亡 Agent 的残留资源"""
        # 释放内存中的会话缓存、关闭未完成的工具调用连接等
        logger.info(f"Cleaning up resources for dead agent {agent_id}")

四、边界分析与架构权衡

4.1 心跳间隔的权衡

心跳间隔太短（1 秒）：网络开销大，服务端处理压力大。心跳间隔太长（30 秒）：Agent 挂了 30 秒你才知道，用户已经等了 30 秒才发现系统没响应。

折中：基础心跳 5 秒（覆盖大部分场景），状态变化时立即发送一次即时心跳。这样正常情况下每 5 秒一次心跳，状态变化时秒级感知。

4.2 检查点的存储频率

每完成一步就保存检查点，意味着每步都有一次存储写入。如果步骤执行很快（每步 1 秒），写入频率就是每秒一次。对象存储的写入延迟约 50-100ms，不影响步骤执行。

但如果步骤执行很慢（每步 10 秒），保存频率是每 10 秒一次，崩溃后最多丢失 10 秒的工作量。

4.3 适用边界与禁用场景

• 适用：WebSocket/SSE 长连接的 Agent 系统、多步骤工具调用链路、需要会话恢复的付费场景
• 禁用：单次请求-响应的简单 Agent（不需要心跳）、短连接 HTTP API（不需要长连接监控）、离线批处理 Agent（不需要实时状态）

4.4 心跳通道与业务通道的隔离

心跳消息和业务消息走同一个 WebSocket 连接时，如果业务消息阻塞（比如大结果传输），心跳也会延迟。解决方案：心跳走独立连接或独立的消息类型（WebSocket 的 ping 帧与数据帧是独立的）。

五、总结

Agent 长连接场景的健康检查需要三层心跳：连接层检测网络可达、进程层检测 Agent 存活与资源状态、业务层检测处理进度是否推进。单靠连接层心跳无法区分"在思考"和"已挂掉"。核心设计：5 秒基础心跳 + 状态变化即时心跳、停滞检测（step 不变的次数超阈值判定卡住）、缺失检测（连续 N 次心跳缺失判定死亡）。检查点机制保证崩溃后断点续传：每完成一步保存结果，恢复时从最新检查点继续，不从头重跑。心跳通道与业务通道隔离，避免业务阻塞影响心跳延迟。