Agent 的守护进程模式：长周期运行 Agent 的状态管理和心跳健康检查

Agent 的守护进程模式长周期运行 Agent 的状态管理和心跳健康检查一、深度引言与场景痛点把 Agent 部署到生产环境后最让人血压飙升的体验是什么不是模型推理慢不是检索召回差而是你周一早上打开监控面板发现 Agent 从周六凌晨就悄无声息地停止了工作——没有报错、没有告警、没有 panic 堆栈就是单纯地不干活了。这就是长周期运行 Agent 面临的核心问题。短任务 Agent 可以用完即焚请求进来、推理完成、返回结果生命周期很短。但守护进程模式的 Agent 需要 7x24 小时在线它要持续监听消息队列、定时执行巡检任务、维持 WebSocket 长连接、管理上下文缓存。时间一长各种问题就冒出来了。首先是状态丢失。Agent 运行中维护了用户会话状态、工具调用历史、中间推理结果这些都在内存里。进程一旦重启所有上下文灰飞烟灭用户下一轮对话时 Agent 像个失忆患者一样从头开始。其次是僵尸进程。Agent 内部可能启动了子进程去调用外部工具如果子进程 hang 住或者异常退出后没有被正确回收进程表里就会积累一堆 defunct 进程慢慢消耗系统资源。然后是资源泄漏。长周期运行下数据库连接没有正确归还、文件句柄忘记关闭、日志缓冲区无限增长最终 OOM 被 k8s 无情 kill。这些问题的共同根源是Agent 缺乏一种系统级的运行时管理机制。它不像普通的 Web 服务有成熟的框架和平台兜底Agent 的运行态更复杂、状态更多样、故障模式也更隐蔽。我们需要一个守护进程模式让 Agent 具备自我感知和自我修复能力。二、底层机制与原理深度剖析守护进程模式的核心设计可以拆成三个正交的组件状态管理器负责追踪 Agent 的当前状态和转换条件心跳机制负责持续输出存活信号恢复控制器负责在检测到异常时触发修复动作。状态机的设计要点在于明确每个状态的进入条件和退出动作。Initializing 阶段要做依赖检查向量库能不能连通、模型端点是否可达、消息队列有没有堆积。Running 状态下心跳每 5 秒发送一次同时监控内存使用率、goroutine/线程数、pending 任务队列长度。Degraded 是一个灰色地带——Agent 还活着但某个能力降级了比如 LLM 超时后回退到了本地缓存检索。Unhealthy 是明确的故障态必须触发恢复。心跳不只是一行print(alive)。生产级的心跳要携带状态摘要当前运行的 step_id、活跃会话数、最近处理的消息 ID、队列深度。这些信息同时写入 Redis用于外部监控读取和本地文件用于进程重启后恢复上下文。心跳丢失的判断需要容忍短暂的网络抖动通常用连续 3 次超时作为阈值避免误触发重启。三、生产级代码实现把上面这些概念变成能跑在生产环境的代码核心是 asyncio 的事件循环加上结构化并发。import asyncio import json import logging import os import signal from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any import aiofiles import redis.asyncio as aioredis logger logging.getLogger(__name__) class AgentStatus(str, Enum): INITIALIZING initializing RUNNING running DEGRADED degraded UNHEALTHY unhealthy RESTARTING restarting STOPPING stopping dataclass class HeartbeatPayload: agent_id: str status: AgentStatus active_sessions: int queue_depth: int memory_mb: float last_message_id: str | None None timestamp: str field(default_factorylambda: datetime.now(timezone.utc).isoformat()) class AgentDaemon: Agent 守护进程状态管理 心跳 自动恢复 def __init__( self, agent_id: str, redis_url: str redis://localhost:6379, heartbeat_interval: int 5, max_retries: int 3, ) - None: self.agent_id agent_id self.status AgentStatus.INITIALIZING self._heartbeat_interval heartbeat_interval self._max_retries max_retries self._retry_count 0 self._redis: aioredis.Redis | None None self._tasks: list[asyncio.Task[Any]] [] self._state_file f/tmp/agent_{agent_id}_state.json async def start(self) - None: 启动守护进程初始化依赖 → 恢复状态 → 开始工作循环 try: # 初始化阶段 self.status AgentStatus.INITIALIZING self._redis await aioredis.from_url( redis://localhost:6379, max_connections20, socket_timeout3.0, retry_on_timeoutTrue, ) await self._redis.ping() # 尝试从文件恢复上次状态 await self._load_state() # 启动核心工作循环 self.status AgentStatus.RUNNING self._tasks.append(asyncio.create_task(self._heartbeat_loop())) self._tasks.append(asyncio.create_task(self._health_check_loop())) self._tasks.append(asyncio.create_task(self._work_loop())) logger.info(Agent daemon started: %s, self.agent_id) except Exception: logger.exception(Failed to start agent daemon) self.status AgentStatus.UNHEALTHY raise async def _heartbeat_loop(self) - None: 心跳循环每 heartbeat_interval 秒发送一次状态快照 consecutive_failures 0 while self.status not in (AgentStatus.STOPPING,): try: payload HeartbeatPayload( agent_idself.agent_id, statusself.status, active_sessionslen(self._tasks), queue_depth0, # 实际场景从队列获取 memory_mbself._get_memory_usage(), ) if self._redis: key fagent:heartbeat:{self.agent_id} await self._redis.setex( key, self._heartbeat_interval * 3, # TTL 为心跳间隔的 3 倍 json.dumps(payload.__dict__), ) # 同步写入本地状态文件 await self._save_state(payload.__dict__) consecutive_failures 0 except Exception as e: consecutive_failures 1 logger.warning(Heartbeat failed (%d/3): %s, consecutive_failures, e) if consecutive_failures 3: await self._trigger_recovery(heartbeat_lost) break await asyncio.sleep(self._heartbeat_interval) async def _health_check_loop(self) - None: 健康检查循环监控内存、连接池、pending 任务 while self.status not in (AgentStatus.STOPPING, AgentStatus.UNHEALTHY): memory_mb self._get_memory_usage() if memory_mb 2048: # 超过 2GB 触发降级 logger.warning(Memory threshold exceeded: %.2f MB, memory_mb) self.status AgentStatus.DEGRADED await asyncio.sleep(30) async def _work_loop(self) - None: 主工作循环从消息队列拉取并处理任务 while self.status in (AgentStatus.RUNNING, AgentStatus.DEGRADED): try: # 实际场景中从 Kafka/Redis 消费消息 await asyncio.sleep(1) except asyncio.CancelledError: break except Exception: logger.exception(Work loop error) # 单次任务失败不触发整体降级由上层决定 async def _trigger_recovery(self, reason: str) - None: 触发自动恢复记录原因 → 尝试重启 → 超限则终止 self._retry_count 1 logger.error( Recovery triggered: %s (attempt %d/%d), reason, self._retry_count, self._max_retries, ) if self._retry_count self._max_retries: logger.critical(Max retries exceeded, agent exiting) self.status AgentStatus.STOPPING return self.status AgentStatus.RESTARTING await self._cleanup() # 等待一段时间后重新初始化 await asyncio.sleep(5) await self.start() async def _cleanup(self) - None: 清理资源取消子任务、关闭连接 for task in self._tasks: task.cancel() if self._redis: await self._redis.aclose() self._tasks.clear() async def _save_state(self, state: dict[str, Any]) - None: 持久化状态到本地文件 try: async with aiofiles.open(self._state_file, w) as f: await f.write(json.dumps(state)) except OSError: logger.exception(Failed to save state file) async def _load_state(self) - None: 从本地文件恢复上次状态 try: if os.path.exists(self._state_file): async with aiofiles.open(self._state_file) as f: content await f.read() state json.loads(content) logger.info(State restored from %s, self._state_file) except (OSError, json.JSONDecodeError): logger.warning(Failed to load state, starting fresh) staticmethod def _get_memory_usage() - float: 获取当前进程内存使用量MB try: import resource return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 except ImportError: return 0.0 async def main() - None: daemon AgentDaemon(agent_idagent-inspector-01) def shutdown_handler(signum: int, _frame: Any) - None: logger.info(Received signal %d, shutting down, signum) daemon.status AgentStatus.STOPPING loop asyncio.get_running_loop() for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler(sig, shutdown_handler, sig, None) try: await daemon.start() # 等待所有子任务完成 await asyncio.gather(*daemon._tasks, return_exceptionsTrue) except KeyboardInterrupt: pass finally: await daemon._cleanup() if __name__ __main__: logging.basicConfig(levellogging.INFO) asyncio.run(main())这段代码有几个生产级的考量状态先写本地文件再写 Redis保证进程重启后能从文件恢复而不是依赖外部存储心跳 TTL 设为间隔的 3 倍既给监控足够的容忍度又不会让故障窗口太大内存阈值触发降级而不是直接标记 Unhealthy给自愈留出空间。_work_loop中的异常不会直接让 Agent 挂掉单次任务失败的容错和整体状态判断是分离的——这是守护进程和普通请求处理最大的区别。四、边界分析与架构权衡守护进程模式解决了长周期运行的状态管理问题但它引入了额外的复杂度。首先状态文件如果损坯可能导致 Agent 反复重启无法恢复需要增加 checksum 校验和备份机制。其次心跳间隔的选择是个两难——间隔太短增加 Redis 写入压力间隔太长导致故障检测延迟。最后自动恢复本身可能掩盖系统性问题如果 Agent 因为某个工具接口的 bug 反复崩溃和重启自愈逻辑会变成一个无限循环消耗资源而不解决问题。和 cron 的对比cron 适合无状态的定时任务每次执行都是全新的环境没有状态管理负担但不适合需要维持上下文的长对话场景。和 k8s Job 的对比k8s Job 保证任务最终成功执行有完善的失败重试和资源隔离但调度延迟通常在秒级不适合对实时性有要求的 Agent。守护进程模式在即时性和状态连续性上更好但需要自己处理故障恢复。和 k8s Deployment liveness probe 的对比Deployment 的存活探针是外部的、粗粒度的判断进程是否在而守护进程的自检是内部的、细粒度的业务是否正常。两者不矛盾可以组合使用k8s 负责进程级存活守护进程负责业务级健康。一个实用的决策建议是如果 Agent 需要维护跨请求的状态会话、缓存、推理链用守护进程模式如果 Agent 是无状态的请求-响应模式直接用 HTTP 服务 k8s 部署更简单可靠。五、总结Agent 守护进程模式解决的是长周期运行下的状态管理、故障检测和自动恢复问题。核心设计包括三个正交组件状态机定义生命周期的合法转换心跳机制提供外部可观测的存活信号恢复控制器在异常时触发自愈。实现上asyncio 的事件循环天然适合这种多协程协同的架构状态持久化做到本地文件优先、Redis 兜底避免外部依赖的单点故障。守护进程模式不是银弹它在即时性、状态连续性上有优势但复杂度也更高。对于无状态 Agentk8s Deployment 是更简单的选择对于定时任务cron 足矣。选择守护进程模式的前提是你的 Agent 确实需要 7x24 小时在线并维持跨请求的上下文。