服务端 Agent 生命周期管理创建、暂停、销毁与资源回收一、一个跑了 3 天的 Agent 把服务器内存吃满了Agent 系统上线第三周监控告警凌晨 3 点炸了——一台推理节点的内存从 30% 飙到 95%。排查发现一个用户创建了 Agent 后发了近 300 条对话Agent 的上下文缓存和工具调用历史塞满了内存但用户早就不跟它交互了。Agent 创建了但没有销毁机制。这暴露了一个核心问题Agent 不是普通请求——它有状态、有生命周期。传统 HTTP 服务的请求-响应模型请求处理完资源就释放。Agent 的对话可能持续数小时甚至数天如果不对其生命周期做显式管理内存泄漏、连接泄漏、GPU 资源占用不释放是必然结果。二、Agent 生命周期状态机Agent 的完整生命周期包含 5 个阶段、6 种状态stateDiagram-v2 [*] -- Created: 用户创建 Agent Created -- Active: 首次对话 Active -- Active: 继续对话 Active -- Paused: 用户主动暂停br/或超时无交互 Paused -- Active: 恢复对话 Active -- Destroying: 用户删除/超时强制清理 Paused -- Destroying: 用户删除/超时强制清理 Destroying -- Destroyed: 资源回收完成 Destroyed -- [*] note right of Active: 占用资源:br/- LLM 上下文缓存br/- 工具连接池br/- 对话历史存储br/- Session 内存 note right of Paused: 释放瞬时资源:br/- LLM 上下文缓存br/- 保留:br/- 对话历史br/- Agent 配置 note right of Destroying: 异步清理流程:br/1. 通知所有工具关闭br/2. Wait 30sbr/3. Force Kill关键状态转换规则Created → Active用户首次发送消息时分配 LLM 连接、构建上下文窗口、初始化工具客户端。Active → Paused超过 30 分钟无交互可配置释放 LLM 连接和 GPU 上下文缓存但保留对话历史到持久化存储。降低资源占用的同时不丢失对话状态。Paused → Active用户再次发送消息时从持久化存储重建上下文窗口。恢复耗时通常 2s含一次 LLM 预热调用。任何状态 → Destroying用户主动删除或者闲置超过 24 小时全局 TTL。进入异步销毁流程。三、生产级 Agent 生命周期管理实现import asyncio import time from enum import Enum from typing import Dict, Optional from dataclasses import dataclass, field import redis.asyncio as aioredis class AgentState(Enum): CREATED created ACTIVE active PAUSED paused DESTROYING destroying DESTROYED destroyed ERROR error dataclass class AgentInstance: Agent 实例的核心数据模型 agent_id: str user_id: str state: AgentState AgentState.CREATED created_at: float field(default_factorytime.time) last_active_at: float field(default_factorytime.time) conversation_turns: int 0 # 资源句柄Paused 状态时置空 llm_connection: Optional[object] None context_cache: Optional[object] None # 持久化存储 key history_key: str checkpoint_key: str class AgentLifecycleManager: Agent 生命周期管理器 负责状态转换、资源分配/释放和超时自动回收 def __init__(self, redis_url: str): self.agents: Dict[str, AgentInstance] {} self.redis None # Async Redis 连接池 self.redis_url redis_url # 为什么用 Dict 内存存储而非纯 Redis # Agent 的 LLM 连接句柄无法序列化到 Redis # 内存 Store 持有活跃连接Redis 仅存序列化状态做持久化和多副本同步 # 配置参数 self.pause_ttl_sec 30 * 60 # 30 分钟无交互自动暂停 self.destroy_ttl_sec 24 * 3600 # 24 小时无交互自动销毁 self.max_conversation_turns 500 # 单会话最大对话轮次 async def start(self): 初始化 Redis 连接并启动后台回收循环 self.redis await aioredis.from_url( self.redis_url, encodingutf-8, decode_responsesTrue ) # 启动后台任务定期扫描超时 Agent asyncio.create_task(self._reap_idle_agents()) async def create_agent(self, user_id: str, config: dict) - AgentInstance: 创建 Agent 实例并分配初始资源 agent_id self._generate_agent_id(user_id) agent AgentInstance( agent_idagent_id, user_iduser_id, stateAgentState.CREATED, ) # 在 Redis 中注册 Agent 元信息 # 为什么先写 Redis 再写本地如果进程崩溃 # 其他副本可以从 Redis 检测到孤儿 Agent 并接管清理 await self.redis.hset( fagent:{agent_id}:meta, mapping{ state: AgentState.CREATED.value, user_id: user_id, created_at: str(agent.created_at), } ) await self.redis.expire( fagent:{agent_id}:meta, self.destroy_ttl_sec ) self.agents[agent_id] agent return agent async def activate(self, agent_id: str) - bool: 激活 AgentCreated → Active 或 Paused → Active 负责分配 LLM 连接、重建上下文缓存 agent self.agents.get(agent_id) if not agent: # 本地不存在尝试从 Redis 恢复 # 场景运维重启后需要从 Redis 恢复之前的 Agent agent await self._recover_from_redis(agent_id) if not agent: return False if agent.state AgentState.ACTIVE: # 已是 Active更新活跃时间即可 agent.last_active_at time.time() return True if agent.state AgentState.DESTROYING: return False # 正在销毁拒绝激活 # 分配 LLM 连接可能涉及连接池争用 try: agent.llm_connection await self._acquire_llm_connection() except ConnectionError: agent.state AgentState.ERROR return False # 如果是 Paused 恢复重建上下文缓存 if agent.state AgentState.PAUSED: await self._rebuild_context_cache(agent) agent.state AgentState.ACTIVE agent.last_active_at time.time() await self._sync_state_to_redis(agent) return True async def pause(self, agent_id: str) - None: 暂停 AgentActive → Paused 释放瞬时资源但保留持久状态 agent self.agents.get(agent_id) if not agent or agent.state ! AgentState.ACTIVE: return # 保存对话检查点到持久化存储 await self._checkpoint_conversation(agent) # 释放 LLM 连接和上下文缓存 if agent.llm_connection: await self._release_llm_connection(agent.llm_connection) agent.llm_connection None agent.context_cache None agent.state AgentState.PAUSED await self._sync_state_to_redis(agent) async def destroy(self, agent_id: str) - bool: 销毁 Agent进入异步销毁流程 为什么异步销毁而非同步 Agent 可能持有工具调用中的连接如数据库、远程 API 直接 kill 会造成连接泄漏。需要给工具 30s 的优雅关闭窗口。 agent self.agents.get(agent_id) if not agent: return False agent.state AgentState.DESTROYING await self._sync_state_to_redis(agent) # 阶段 1: 通知所有工具准备关闭优雅窗口 30s await self._notify_tools_shutdown(agent, grace_period30) # 阶段 2: 等待工具响应最多 30s try: await asyncio.wait_for( self._wait_tools_ack(agent), timeout30.0 ) except asyncio.TimeoutError: # 超时则强制释放连接 pass # 阶段 3: 释放所有资源 if agent.llm_connection: await self._release_llm_connection(agent.llm_connection) # 阶段 4: 清理持久化存储 await self._cleanup_persistent_data(agent) agent.state AgentState.DESTROYED await self._sync_state_to_redis(agent) # 从内存中移除 self.agents.pop(agent_id, None) return True async def _reap_idle_agents(self): 后台循环回收超时的空闲 Agent while True: await asyncio.sleep(60) # 每分钟扫描一次 now time.time() idle_agents [] for agent_id, agent in list(self.agents.items()): idle_sec now - agent.last_active_at if agent.state AgentState.ACTIVE: if idle_sec self.destroy_ttl_sec: # 空闲超过 24h → 销毁 asyncio.create_task(self.destroy(agent_id)) elif idle_sec self.pause_ttl_sec: # 空闲超过 30min → 暂停 asyncio.create_task(self.pause(agent_id)) elif agent.state AgentState.PAUSED: if idle_sec self.destroy_ttl_sec: asyncio.create_task(self.destroy(agent_id)) async def _acquire_llm_connection(self): 从 LLM 连接池获取连接 # 实际实现需接入连接池 pass async def _release_llm_connection(self, conn): 归还 LLM 连接到连接池 pass async def _checkpoint_conversation(self, agent: AgentInstance): 保存对话检查点到 Redis pass async def _rebuild_context_cache(self, agent: AgentInstance): 从检查点恢复上下文缓存 pass async def _sync_state_to_redis(self, agent: AgentInstance): 同步 Agent 状态到 Redis if self.redis: await self.redis.hset( fagent:{agent.agent_id}:meta, state, agent.state.value, ) async def _recover_from_redis(self, agent_id: str): 从 Redis 恢复 Agent进程重启场景 if not self.redis: return None meta await self.redis.hgetall(fagent:{agent_id}:meta) if not meta: return None agent AgentInstance( agent_idagent_id, user_idmeta.get(user_id, ), stateAgentState(meta.get(state, created)), ) self.agents[agent_id] agent return agent async def _notify_tools_shutdown(self, agent, grace_period): Send shutdown signal to all tools pass async def _wait_tools_ack(self, agent): Wait for tools to acknowledge shutdown pass async def _cleanup_persistent_data(self, agent): Clean up persistent data if self.redis: await self.redis.delete(fagent:{agent.agent_id}:meta) def _generate_agent_id(self, user_id: str) - str: import uuid return f{user_id}:{uuid.uuid4().hex[:12]}关键配置参数# agent-lifecycle.yaml lifecycle: pause_after_idle_sec: 1800 # 30 分钟无交互暂停 destroy_after_idle_sec: 86400 # 24 小时无交互销毁 max_conversation_turns: 500 # 单次对话最大轮数 graceful_shutdown_timeout: 30 # 优雅关闭等待时间 resource_limits: max_active_agents_per_user: 5 # 单用户最多活跃 Agent max_total_active_agents: 1000 # 全局活跃 Agent 上限 checkpoint: enabled: true interval_turns: 10 # 每 10 轮对话做一次 checkpoint retention_count: 3 # 保留最近 3 个 checkpoint四、生命周期管理的边界与不足缺点死锁风险Paused→Active 的恢复过程依赖外部资源LLM 服务、Redis如果这些服务在恢复时不可用Agent 会永久卡在 Paused 状态。需要增加重试上限和降级策略。checkpoint 一致性对话历史是追加写入的checkpoint 是异步快照。极端情况下进程崩溃恰好发生在 checkpoint 写入一半时可能导致对话历史与 checkpoint 不一致。可通过预写日志WAL解决。分布式同步内存 Store Redis 的双写模式在网络分区时可能出现不一致。需要引入版本号vector clock做冲突检测。禁用场景完全无状态的 Agent每次请求都重新构建上下文生命周期管理的 overhead 超过收益。需要严格的事务一致性的场景Agent 状态和业务事务耦合时两阶段提交的实现复杂度极高。五、总结Agent 的生命周期管理是工程化落地的必修课。显式定义 Created→Active→Paused→Destroying→Destroyed 的状态机用空闲超时和资源上限做自动回收用异步销毁流程避免连接泄漏。两个关键设计Paused 状态区分释放瞬时资源和保留持久状态Destroying 状态提供优雅关闭窗口。不管理生命周期的 Agent 系统迟早被资源泄漏反噬。
服务端 Agent 生命周期管理创建、暂停、销毁与资源回收一、一个跑了 3 天的 Agent 把服务器内存吃满了Agent 系统上线第三周监控告警凌晨 3 点炸了——一台推理节点的内存从 30% 飙到 95%。排查发现一个用户创建了 Agent 后发了近 300 条对话Agent 的上下文缓存和工具调用历史塞满了内存但用户早就不跟它交互了。Agent 创建了但没有销毁机制。这暴露了一个核心问题Agent 不是普通请求——它有状态、有生命周期。传统 HTTP 服务的请求-响应模型请求处理完资源就释放。Agent 的对话可能持续数小时甚至数天如果不对其生命周期做显式管理内存泄漏、连接泄漏、GPU 资源占用不释放是必然结果。二、Agent 生命周期状态机Agent 的完整生命周期包含 5 个阶段、6 种状态stateDiagram-v2 [*] -- Created: 用户创建 Agent Created -- Active: 首次对话 Active -- Active: 继续对话 Active -- Paused: 用户主动暂停br/或超时无交互 Paused -- Active: 恢复对话 Active -- Destroying: 用户删除/超时强制清理 Paused -- Destroying: 用户删除/超时强制清理 Destroying -- Destroyed: 资源回收完成 Destroyed -- [*] note right of Active: 占用资源:br/- LLM 上下文缓存br/- 工具连接池br/- 对话历史存储br/- Session 内存 note right of Paused: 释放瞬时资源:br/- LLM 上下文缓存br/- 保留:br/- 对话历史br/- Agent 配置 note right of Destroying: 异步清理流程:br/1. 通知所有工具关闭br/2. Wait 30sbr/3. Force Kill关键状态转换规则Created → Active用户首次发送消息时分配 LLM 连接、构建上下文窗口、初始化工具客户端。Active → Paused超过 30 分钟无交互可配置释放 LLM 连接和 GPU 上下文缓存但保留对话历史到持久化存储。降低资源占用的同时不丢失对话状态。Paused → Active用户再次发送消息时从持久化存储重建上下文窗口。恢复耗时通常 2s含一次 LLM 预热调用。任何状态 → Destroying用户主动删除或者闲置超过 24 小时全局 TTL。进入异步销毁流程。三、生产级 Agent 生命周期管理实现import asyncio import time from enum import Enum from typing import Dict, Optional from dataclasses import dataclass, field import redis.asyncio as aioredis class AgentState(Enum): CREATED created ACTIVE active PAUSED paused DESTROYING destroying DESTROYED destroyed ERROR error dataclass class AgentInstance: Agent 实例的核心数据模型 agent_id: str user_id: str state: AgentState AgentState.CREATED created_at: float field(default_factorytime.time) last_active_at: float field(default_factorytime.time) conversation_turns: int 0 # 资源句柄Paused 状态时置空 llm_connection: Optional[object] None context_cache: Optional[object] None # 持久化存储 key history_key: str checkpoint_key: str class AgentLifecycleManager: Agent 生命周期管理器 负责状态转换、资源分配/释放和超时自动回收 def __init__(self, redis_url: str): self.agents: Dict[str, AgentInstance] {} self.redis None # Async Redis 连接池 self.redis_url redis_url # 为什么用 Dict 内存存储而非纯 Redis # Agent 的 LLM 连接句柄无法序列化到 Redis # 内存 Store 持有活跃连接Redis 仅存序列化状态做持久化和多副本同步 # 配置参数 self.pause_ttl_sec 30 * 60 # 30 分钟无交互自动暂停 self.destroy_ttl_sec 24 * 3600 # 24 小时无交互自动销毁 self.max_conversation_turns 500 # 单会话最大对话轮次 async def start(self): 初始化 Redis 连接并启动后台回收循环 self.redis await aioredis.from_url( self.redis_url, encodingutf-8, decode_responsesTrue ) # 启动后台任务定期扫描超时 Agent asyncio.create_task(self._reap_idle_agents()) async def create_agent(self, user_id: str, config: dict) - AgentInstance: 创建 Agent 实例并分配初始资源 agent_id self._generate_agent_id(user_id) agent AgentInstance( agent_idagent_id, user_iduser_id, stateAgentState.CREATED, ) # 在 Redis 中注册 Agent 元信息 # 为什么先写 Redis 再写本地如果进程崩溃 # 其他副本可以从 Redis 检测到孤儿 Agent 并接管清理 await self.redis.hset( fagent:{agent_id}:meta, mapping{ state: AgentState.CREATED.value, user_id: user_id, created_at: str(agent.created_at), } ) await self.redis.expire( fagent:{agent_id}:meta, self.destroy_ttl_sec ) self.agents[agent_id] agent return agent async def activate(self, agent_id: str) - bool: 激活 AgentCreated → Active 或 Paused → Active 负责分配 LLM 连接、重建上下文缓存 agent self.agents.get(agent_id) if not agent: # 本地不存在尝试从 Redis 恢复 # 场景运维重启后需要从 Redis 恢复之前的 Agent agent await self._recover_from_redis(agent_id) if not agent: return False if agent.state AgentState.ACTIVE: # 已是 Active更新活跃时间即可 agent.last_active_at time.time() return True if agent.state AgentState.DESTROYING: return False # 正在销毁拒绝激活 # 分配 LLM 连接可能涉及连接池争用 try: agent.llm_connection await self._acquire_llm_connection() except ConnectionError: agent.state AgentState.ERROR return False # 如果是 Paused 恢复重建上下文缓存 if agent.state AgentState.PAUSED: await self._rebuild_context_cache(agent) agent.state AgentState.ACTIVE agent.last_active_at time.time() await self._sync_state_to_redis(agent) return True async def pause(self, agent_id: str) - None: 暂停 AgentActive → Paused 释放瞬时资源但保留持久状态 agent self.agents.get(agent_id) if not agent or agent.state ! AgentState.ACTIVE: return # 保存对话检查点到持久化存储 await self._checkpoint_conversation(agent) # 释放 LLM 连接和上下文缓存 if agent.llm_connection: await self._release_llm_connection(agent.llm_connection) agent.llm_connection None agent.context_cache None agent.state AgentState.PAUSED await self._sync_state_to_redis(agent) async def destroy(self, agent_id: str) - bool: 销毁 Agent进入异步销毁流程 为什么异步销毁而非同步 Agent 可能持有工具调用中的连接如数据库、远程 API 直接 kill 会造成连接泄漏。需要给工具 30s 的优雅关闭窗口。 agent self.agents.get(agent_id) if not agent: return False agent.state AgentState.DESTROYING await self._sync_state_to_redis(agent) # 阶段 1: 通知所有工具准备关闭优雅窗口 30s await self._notify_tools_shutdown(agent, grace_period30) # 阶段 2: 等待工具响应最多 30s try: await asyncio.wait_for( self._wait_tools_ack(agent), timeout30.0 ) except asyncio.TimeoutError: # 超时则强制释放连接 pass # 阶段 3: 释放所有资源 if agent.llm_connection: await self._release_llm_connection(agent.llm_connection) # 阶段 4: 清理持久化存储 await self._cleanup_persistent_data(agent) agent.state AgentState.DESTROYED await self._sync_state_to_redis(agent) # 从内存中移除 self.agents.pop(agent_id, None) return True async def _reap_idle_agents(self): 后台循环回收超时的空闲 Agent while True: await asyncio.sleep(60) # 每分钟扫描一次 now time.time() idle_agents [] for agent_id, agent in list(self.agents.items()): idle_sec now - agent.last_active_at if agent.state AgentState.ACTIVE: if idle_sec self.destroy_ttl_sec: # 空闲超过 24h → 销毁 asyncio.create_task(self.destroy(agent_id)) elif idle_sec self.pause_ttl_sec: # 空闲超过 30min → 暂停 asyncio.create_task(self.pause(agent_id)) elif agent.state AgentState.PAUSED: if idle_sec self.destroy_ttl_sec: asyncio.create_task(self.destroy(agent_id)) async def _acquire_llm_connection(self): 从 LLM 连接池获取连接 # 实际实现需接入连接池 pass async def _release_llm_connection(self, conn): 归还 LLM 连接到连接池 pass async def _checkpoint_conversation(self, agent: AgentInstance): 保存对话检查点到 Redis pass async def _rebuild_context_cache(self, agent: AgentInstance): 从检查点恢复上下文缓存 pass async def _sync_state_to_redis(self, agent: AgentInstance): 同步 Agent 状态到 Redis if self.redis: await self.redis.hset( fagent:{agent.agent_id}:meta, state, agent.state.value, ) async def _recover_from_redis(self, agent_id: str): 从 Redis 恢复 Agent进程重启场景 if not self.redis: return None meta await self.redis.hgetall(fagent:{agent_id}:meta) if not meta: return None agent AgentInstance( agent_idagent_id, user_idmeta.get(user_id, ), stateAgentState(meta.get(state, created)), ) self.agents[agent_id] agent return agent async def _notify_tools_shutdown(self, agent, grace_period): Send shutdown signal to all tools pass async def _wait_tools_ack(self, agent): Wait for tools to acknowledge shutdown pass async def _cleanup_persistent_data(self, agent): Clean up persistent data if self.redis: await self.redis.delete(fagent:{agent.agent_id}:meta) def _generate_agent_id(self, user_id: str) - str: import uuid return f{user_id}:{uuid.uuid4().hex[:12]}关键配置参数# agent-lifecycle.yaml lifecycle: pause_after_idle_sec: 1800 # 30 分钟无交互暂停 destroy_after_idle_sec: 86400 # 24 小时无交互销毁 max_conversation_turns: 500 # 单次对话最大轮数 graceful_shutdown_timeout: 30 # 优雅关闭等待时间 resource_limits: max_active_agents_per_user: 5 # 单用户最多活跃 Agent max_total_active_agents: 1000 # 全局活跃 Agent 上限 checkpoint: enabled: true interval_turns: 10 # 每 10 轮对话做一次 checkpoint retention_count: 3 # 保留最近 3 个 checkpoint四、生命周期管理的边界与不足缺点死锁风险Paused→Active 的恢复过程依赖外部资源LLM 服务、Redis如果这些服务在恢复时不可用Agent 会永久卡在 Paused 状态。需要增加重试上限和降级策略。checkpoint 一致性对话历史是追加写入的checkpoint 是异步快照。极端情况下进程崩溃恰好发生在 checkpoint 写入一半时可能导致对话历史与 checkpoint 不一致。可通过预写日志WAL解决。分布式同步内存 Store Redis 的双写模式在网络分区时可能出现不一致。需要引入版本号vector clock做冲突检测。禁用场景完全无状态的 Agent每次请求都重新构建上下文生命周期管理的 overhead 超过收益。需要严格的事务一致性的场景Agent 状态和业务事务耦合时两阶段提交的实现复杂度极高。五、总结Agent 的生命周期管理是工程化落地的必修课。显式定义 Created→Active→Paused→Destroying→Destroyed 的状态机用空闲超时和资源上限做自动回收用异步销毁流程避免连接泄漏。两个关键设计Paused 状态区分释放瞬时资源和保留持久状态Destroying 状态提供优雅关闭窗口。不管理生命周期的 Agent 系统迟早被资源泄漏反噬。
