Agent 高可用设计多实例部署下的状态同步和故障转移策略一、深度引言与场景痛点大家好我是赵咕咕。一个真实的故障故事凌晨两点我们的 AI 客服 Agent 服务因为一个第三方地图 API 的超时导致整个 Agent 进程卡死。因为这个 Agent 是单实例部署的它一挂所有的用户会话全断了。用户那边显示的是永远的AI 正在思考中…。这个故障暴露了三个问题单实例没有任何冗余挂了就是全体掉线Agent 的会话状态都在内存里即使能快速重启一个新的老会话也回不去了没有故障转移机制用户在断线后只能从头开始对话Agent 的高可用设计和微服务的高可用还不完全一样。Agent 多了一个独特的挑战——有状态。用户的对话上下文、工具调用历史、中间结果——这些不是简单的实例重启就能恢复。今天我们来聊聊多实例部署的 Agent 系统怎么保证状态不丢、故障自愈、用户无感。二、底层机制与原理深度剖析2.1 Agent 状态分类flowchart TD A[Agent 状态] -- B{状态类型} B -- C[会话状态/Session State] B -- D[任务状态/Task State] B -- E[配置状态/Config State] C -- C1[对话历史br/工具调用记录br/中间推理结果] C -- C2[特点高频写入br/必须持久化br/跨实例共享] D -- D1[当前执行步骤br/待处理工具调用br/执行计划] D -- D2[特点生命周期短br/需要幂等保证br/故障后可重放] E -- E1[System Promptbr/工具定义br/模型配置] E -- E2[特点低频变更br/所有实例相同br/配置中心管理] style C2 fill:#ffcdd2 style D2 fill:#fff9c4 style E2 fill:#c8e6c9Agent 的高可用设计本质上是把会话状态和任务状态从进程内存外移到外部存储然后让任何健康的实例都能接手。2.2 多实例架构flowchart TD LB[负载均衡器br/Nginx/Traefik] -- A1[Agent 实例 1] LB -- A2[Agent 实例 2] LB -- A3[Agent 实例 N] A1 -- Redis[Redisbr/会话状态存储br/分布式锁] A2 -- Redis A3 -- Redis A1 -- DB[(PostgreSQLbr/持久化存储br/任务日志)] A2 -- DB A3 -- DB A1 -.-|心跳| HW[Health Watcherbr/健康检查/故障转移] A2 -.-|心跳| HW A3 -.-|心跳| HW HW --|检测到故障| LB HW --|故障通知| Alert[告警系统] style Redis fill:#dc3545,color:white style DB fill:#336791,color:white style HW fill:#ffc1072.3 故障转移流程sequenceDiagram participant U as 用户 participant LB as 负载均衡器 participant A1 as Agent 实例 1br/(运行中) participant A2 as Agent 实例 2br/(备用) participant Redis as Redis participant HW as Health Watcher U-LB: 发送消息 LB-A1: 路由到实例1 Note over A1: 处理请求中... A1-Redis: 保存会话快照 A1--xA1: ❌ 进程崩溃 HW-A1: 心跳超时 HW-LB: 标记实例1为不健康 HW-Alert: 发送告警 U-LB: 重试请求 LB-A2: 路由到实例2 A2-Redis: 加载会话状态 Redis--A2: 会话快照 历史 A2-A2: 重放未完成的任务 A2--U: 无缝恢复继续对话三、生产级代码实现import asyncio import json import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any, Optional import redis.asyncio as redis # ── 1. 会话状态数据模型 ─────────────────────────── class SessionStatus(Enum): ACTIVE active PAUSED paused # 暂停等待工具返回 FAILING_OVER failing_over # 故障转移中 COMPLETED completed dataclass class AgentSession: Agent 会话的完整状态 session_id: str field(default_factorylambda: str(uuid.uuid4())) user_id: str status: SessionStatus SessionStatus.ACTIVE # 对话历史 messages: list[dict] field(default_factorylist) # 当前任务状态 current_step: int 0 pending_tool_calls: list[dict] field(default_factorylist) current_plan: list[dict] field(default_factorylist) # 实例信息 owner_instance: str # 当前负责的实例 ID created_at: str field( default_factorylambda: datetime.now(timezone.utc).isoformat() ) updated_at: str field( default_factorylambda: datetime.now(timezone.utc).isoformat() ) version: int 0 # 乐观锁版本号 # ── 2. 会话状态管理器 ───────────────────────────── class SessionStateManager: 基于 Redis 的会话状态持久化管理器 SESSION_PREFIX agent:session: LOCK_PREFIX agent:lock:session: HEARTBEAT_PREFIX agent:heartbeat: def __init__(self, redis_client: redis.Redis): self._redis redis_client self._instance_id str(uuid.uuid4())[:8] async def save_session(self, session: AgentSession) - bool: 持久化会话状态到 Redis session.updated_at datetime.now(timezone.utc).isoformat() session.version 1 key f{self.SESSION_PREFIX}{session.session_id} data { session_id: session.session_id, user_id: session.user_id, status: session.status.value, messages: json.dumps(session.messages), current_step: str(session.current_step), pending_tool_calls: json.dumps(session.pending_tool_calls), current_plan: json.dumps(session.current_plan), owner_instance: session.owner_instance, version: str(session.version), updated_at: session.updated_at, } await self._redis.hset(key, mappingdata) # 设置过期时间防止僵尸会话 await self._redis.expire(key, 3600) return True async def load_session(self, session_id: str) - Optional[AgentSession]: 从 Redis 加载会话状态 key f{self.SESSION_PREFIX}{session_id} data await self._redis.hgetall(key) if not data: return None return AgentSession( session_iddata.get(session_id, session_id), user_iddata.get(user_id, ), statusSessionStatus(data.get(status, active)), messagesjson.loads(data.get(messages, [])), current_stepint(data.get(current_step, 0)), pending_tool_callsjson.loads(data.get(pending_tool_calls, [])), current_planjson.loads(data.get(current_plan, [])), owner_instancedata.get(owner_instance, ), versionint(data.get(version, 0)), ) async def acquire_session_lock( self, session_id: str, ttl: int 30 ) - bool: 获取会话的分布式锁 lock_key f{self.LOCK_PREFIX}{session_id} return await self._redis.set( lock_key, self._instance_id, nxTrue, exttl ) async def release_session_lock(self, session_id: str): 释放会话锁仅当锁属于当前实例时 lock_key f{self.LOCK_PREFIX}{session_id} # Lua 脚本保证原子性 script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end await self._redis.eval(script, 1, lock_key, self._instance_id) async def update_heartbeat(self): 更新实例心跳 key f{self.HEARTBEAT_PREFIX}{self._instance_id} await self._redis.setex(key, 15, alive) async def is_instance_alive(self, instance_id: str) - bool: 检查实例是否存活 key f{self.HEARTBEAT_PREFIX}{instance_id} return await self._redis.exists(key) 0 async def claim_orphan_sessions(self) - list[str]: 认领孤立的会话原 owner 已挂 orphan_ids [] cursor 0 while True: cursor, keys await self._redis.scan( cursor, matchf{self.SESSION_PREFIX}*, count100 ) for key in keys: owner await self._redis.hget(key, owner_instance) if owner and not await self.is_instance_alive(owner): session_id key.decode().replace(self.SESSION_PREFIX, ) orphan_ids.append(session_id) if cursor 0: break return orphan_ids # ── 3. 高可用 Agent ──────────────────────────────── class HighAvailabilityAgent: 具备故障转移能力的高可用 Agent def __init__( self, state_manager: SessionStateManager, max_retry_steps: int 3, session_ttl: int 3600, ): self._state state_manager self._max_retry max_retry_steps self._session_ttl session_ttl async def process_message( self, session_id: str, user_msg: str ) - dict: 处理用户消息带故障转移 # 1. 获取会话锁 locked await self._state.acquire_session_lock(session_id) if not locked: return {error: 会话正被其他实例处理请稍后重试} try: # 2. 加载或创建会话 session await self._state.load_session(session_id) if session is None: session AgentSession( session_idsession_id, owner_instanceself._state._instance_id, ) else: session.owner_instance self._state._instance_id # 3. 追加用户消息 session.messages.append({ role: user, content: user_msg }) # 4. 执行 Agent 推理 response await self._execute_with_retry(session) # 5. 持久化状态 session.messages.append({ role: assistant, content: response }) session.status SessionStatus.COMPLETED await self._state.save_session(session) return {session_id: session_id, response: response} finally: await self._state.release_session_lock(session_id) async def _execute_with_retry(self, session: AgentSession) - str: 带重试的 Agent 执行 last_error None for attempt in range(self._max_retry): try: # 保存检查点 await self._state.save_session(session) # 执行推理这里接入实际的 LLM 调用 result await self._do_inference(session) return result except Exception as exc: last_error exc if attempt self._max_retry - 1: # 恢复到最后一次检查点 restored await self._state.load_session( session.session_id ) if restored: session restored await asyncio.sleep(2 ** attempt) raise RuntimeError( fAgent 执行失败已重试 {self._max_retry} 次: {last_error} ) async def _do_inference(self, session: AgentSession) - str: 实际的 Agent 推理逻辑简化版 # 这里接入 LLM、工具调用等 await asyncio.sleep(0.1) # 模拟推理延迟 return f处理完成 (步骤 {session.current_step 1}) # ── 4. 故障检测和自动转移 ───────────────────────── class FailoverWatcher: 故障检测和自动转移引擎 def __init__( self, state_manager: SessionStateManager, check_interval: float 5.0, ): self._state state_manager self._interval check_interval async def watch_loop(self): 持续监控并执行故障转移 while True: try: # 1. 更新自身心跳 await self._state.update_heartbeat() # 2. 查找孤儿会话 orphans await self._state.claim_orphan_sessions() # 3. 接管孤儿会话 for session_id in orphans: await self._failover_session(session_id) await asyncio.sleep(self._interval) except Exception as exc: # 故障检测自身不能挂 await asyncio.sleep(self._interval) async def _failover_session(self, session_id: str): 执行单个会话的故障转移 session await self._state.load_session(session_id) if session is None: return # 标记为故障转移中 session.status SessionStatus.FAILING_OVER await self._state.save_session(session) # 恢复到最后一个检查点 # 实际生产环境中这一步会重新执行最后一步操作 session.owner_instance self._state._instance_id session.status SessionStatus.ACTIVE await self._state.save_session(session) # ── 5. 启动入口 ──────────────────────────────────── async def start_ha_agent(): 启动高可用 Agent 服务 redis_client redis.from_url(redis://localhost:6379) state_manager SessionStateManager(redis_client) agent HighAvailabilityAgent(state_manager) watcher FailoverWatcher(state_manager) # 启动故障检测 asyncio.create_task(watcher.watch_loop()) # 启动心跳 asyncio.create_task(_heartbeat_loop(state_manager)) # 启动 HTTP 服务FastAPI 等 # ... async def _heartbeat_loop(state_manager: SessionStateManager): while True: await state_manager.update_heartbeat() await asyncio.sleep(10)四、边界分析与架构权衡一致性与可用性的抉择多实例 Agent 的高可用本质上是一个 CAP 问题。当网络分区发生时两个实例可能同时认为对方挂了都去接管同一个会话。这时必须用分布式锁保证同一时刻只有一个实例处理同一个会话。代价是极端情况下会牺牲一点可用性——锁超时时间内会话不可用。状态同步的粒度每次 Agent 执行步骤后都全量保存会话状态延迟敏感度高的场景受不了。折中方案是增量 全量高频步骤只追加增量新消息每秒做一次全量快照。如果故障发生从最近的全量快照恢复 增量重放。有工具调用正在进行时的故障转移如果 Agent 正在调用一个外部 API比如发起退款调用还没返回就挂了新实例接手后不能简单地重调。退款 API 可能已经执行了——重调会导致重复退款。解决方案所有的外部调用都要做幂等设计或者在调用前先记录意图到 Redis新实例检查到未完成的意图后再决定是重试还是查询结果。故障转移的延迟从故障检测到新会话完全恢复通常有 5-15 秒的延迟心跳超时 10s 故障转移处理 2-5s。这期间用户看到的可能是正在重连。如果业务不能接受这个延迟需要缩短心跳间隔但代价是更多的心跳网络开销。五、总结Agent 高可用的核心是状态外移 分布式锁 故障检测三位一体状态外移所有会话状态、任务状态持久化到 RedisAgent 实例本身无状态分布式锁用 Redis 锁保证同一会话同一时间只有一个实例在处理故障检测心跳检测 孤儿会话认领实现 5-15 秒级故障转移这不是微服务高可用的简单复制。Agent 的有状态特性让它的高可用设计更复杂但思路是清晰的让状态活在外面的保险箱里让 Agent 变成随时可替换的执行单元。记住凌晨两点的故障不会因为你准备了预案就不来。但它来了之后你的系统是在 10 秒内自愈还是等你从被窝里爬起来手动重启——这取决于你今天有没有做高可用设计。下一篇预告Agent 服务化架构将 Agent 封装为微服务的接口设计与性能考量。
Agent 高可用设计：多实例部署下的状态同步和故障转移策略
发布时间：2026/7/13 16:05:55
Agent 高可用设计多实例部署下的状态同步和故障转移策略一、深度引言与场景痛点大家好我是赵咕咕。一个真实的故障故事凌晨两点我们的 AI 客服 Agent 服务因为一个第三方地图 API 的超时导致整个 Agent 进程卡死。因为这个 Agent 是单实例部署的它一挂所有的用户会话全断了。用户那边显示的是永远的AI 正在思考中…。这个故障暴露了三个问题单实例没有任何冗余挂了就是全体掉线Agent 的会话状态都在内存里即使能快速重启一个新的老会话也回不去了没有故障转移机制用户在断线后只能从头开始对话Agent 的高可用设计和微服务的高可用还不完全一样。Agent 多了一个独特的挑战——有状态。用户的对话上下文、工具调用历史、中间结果——这些不是简单的实例重启就能恢复。今天我们来聊聊多实例部署的 Agent 系统怎么保证状态不丢、故障自愈、用户无感。二、底层机制与原理深度剖析2.1 Agent 状态分类flowchart TD A[Agent 状态] -- B{状态类型} B -- C[会话状态/Session State] B -- D[任务状态/Task State] B -- E[配置状态/Config State] C -- C1[对话历史br/工具调用记录br/中间推理结果] C -- C2[特点高频写入br/必须持久化br/跨实例共享] D -- D1[当前执行步骤br/待处理工具调用br/执行计划] D -- D2[特点生命周期短br/需要幂等保证br/故障后可重放] E -- E1[System Promptbr/工具定义br/模型配置] E -- E2[特点低频变更br/所有实例相同br/配置中心管理] style C2 fill:#ffcdd2 style D2 fill:#fff9c4 style E2 fill:#c8e6c9Agent 的高可用设计本质上是把会话状态和任务状态从进程内存外移到外部存储然后让任何健康的实例都能接手。2.2 多实例架构flowchart TD LB[负载均衡器br/Nginx/Traefik] -- A1[Agent 实例 1] LB -- A2[Agent 实例 2] LB -- A3[Agent 实例 N] A1 -- Redis[Redisbr/会话状态存储br/分布式锁] A2 -- Redis A3 -- Redis A1 -- DB[(PostgreSQLbr/持久化存储br/任务日志)] A2 -- DB A3 -- DB A1 -.-|心跳| HW[Health Watcherbr/健康检查/故障转移] A2 -.-|心跳| HW A3 -.-|心跳| HW HW --|检测到故障| LB HW --|故障通知| Alert[告警系统] style Redis fill:#dc3545,color:white style DB fill:#336791,color:white style HW fill:#ffc1072.3 故障转移流程sequenceDiagram participant U as 用户 participant LB as 负载均衡器 participant A1 as Agent 实例 1br/(运行中) participant A2 as Agent 实例 2br/(备用) participant Redis as Redis participant HW as Health Watcher U-LB: 发送消息 LB-A1: 路由到实例1 Note over A1: 处理请求中... A1-Redis: 保存会话快照 A1--xA1: ❌ 进程崩溃 HW-A1: 心跳超时 HW-LB: 标记实例1为不健康 HW-Alert: 发送告警 U-LB: 重试请求 LB-A2: 路由到实例2 A2-Redis: 加载会话状态 Redis--A2: 会话快照 历史 A2-A2: 重放未完成的任务 A2--U: 无缝恢复继续对话三、生产级代码实现import asyncio import json import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any, Optional import redis.asyncio as redis # ── 1. 会话状态数据模型 ─────────────────────────── class SessionStatus(Enum): ACTIVE active PAUSED paused # 暂停等待工具返回 FAILING_OVER failing_over # 故障转移中 COMPLETED completed dataclass class AgentSession: Agent 会话的完整状态 session_id: str field(default_factorylambda: str(uuid.uuid4())) user_id: str status: SessionStatus SessionStatus.ACTIVE # 对话历史 messages: list[dict] field(default_factorylist) # 当前任务状态 current_step: int 0 pending_tool_calls: list[dict] field(default_factorylist) current_plan: list[dict] field(default_factorylist) # 实例信息 owner_instance: str # 当前负责的实例 ID created_at: str field( default_factorylambda: datetime.now(timezone.utc).isoformat() ) updated_at: str field( default_factorylambda: datetime.now(timezone.utc).isoformat() ) version: int 0 # 乐观锁版本号 # ── 2. 会话状态管理器 ───────────────────────────── class SessionStateManager: 基于 Redis 的会话状态持久化管理器 SESSION_PREFIX agent:session: LOCK_PREFIX agent:lock:session: HEARTBEAT_PREFIX agent:heartbeat: def __init__(self, redis_client: redis.Redis): self._redis redis_client self._instance_id str(uuid.uuid4())[:8] async def save_session(self, session: AgentSession) - bool: 持久化会话状态到 Redis session.updated_at datetime.now(timezone.utc).isoformat() session.version 1 key f{self.SESSION_PREFIX}{session.session_id} data { session_id: session.session_id, user_id: session.user_id, status: session.status.value, messages: json.dumps(session.messages), current_step: str(session.current_step), pending_tool_calls: json.dumps(session.pending_tool_calls), current_plan: json.dumps(session.current_plan), owner_instance: session.owner_instance, version: str(session.version), updated_at: session.updated_at, } await self._redis.hset(key, mappingdata) # 设置过期时间防止僵尸会话 await self._redis.expire(key, 3600) return True async def load_session(self, session_id: str) - Optional[AgentSession]: 从 Redis 加载会话状态 key f{self.SESSION_PREFIX}{session_id} data await self._redis.hgetall(key) if not data: return None return AgentSession( session_iddata.get(session_id, session_id), user_iddata.get(user_id, ), statusSessionStatus(data.get(status, active)), messagesjson.loads(data.get(messages, [])), current_stepint(data.get(current_step, 0)), pending_tool_callsjson.loads(data.get(pending_tool_calls, [])), current_planjson.loads(data.get(current_plan, [])), owner_instancedata.get(owner_instance, ), versionint(data.get(version, 0)), ) async def acquire_session_lock( self, session_id: str, ttl: int 30 ) - bool: 获取会话的分布式锁 lock_key f{self.LOCK_PREFIX}{session_id} return await self._redis.set( lock_key, self._instance_id, nxTrue, exttl ) async def release_session_lock(self, session_id: str): 释放会话锁仅当锁属于当前实例时 lock_key f{self.LOCK_PREFIX}{session_id} # Lua 脚本保证原子性 script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end await self._redis.eval(script, 1, lock_key, self._instance_id) async def update_heartbeat(self): 更新实例心跳 key f{self.HEARTBEAT_PREFIX}{self._instance_id} await self._redis.setex(key, 15, alive) async def is_instance_alive(self, instance_id: str) - bool: 检查实例是否存活 key f{self.HEARTBEAT_PREFIX}{instance_id} return await self._redis.exists(key) 0 async def claim_orphan_sessions(self) - list[str]: 认领孤立的会话原 owner 已挂 orphan_ids [] cursor 0 while True: cursor, keys await self._redis.scan( cursor, matchf{self.SESSION_PREFIX}*, count100 ) for key in keys: owner await self._redis.hget(key, owner_instance) if owner and not await self.is_instance_alive(owner): session_id key.decode().replace(self.SESSION_PREFIX, ) orphan_ids.append(session_id) if cursor 0: break return orphan_ids # ── 3. 高可用 Agent ──────────────────────────────── class HighAvailabilityAgent: 具备故障转移能力的高可用 Agent def __init__( self, state_manager: SessionStateManager, max_retry_steps: int 3, session_ttl: int 3600, ): self._state state_manager self._max_retry max_retry_steps self._session_ttl session_ttl async def process_message( self, session_id: str, user_msg: str ) - dict: 处理用户消息带故障转移 # 1. 获取会话锁 locked await self._state.acquire_session_lock(session_id) if not locked: return {error: 会话正被其他实例处理请稍后重试} try: # 2. 加载或创建会话 session await self._state.load_session(session_id) if session is None: session AgentSession( session_idsession_id, owner_instanceself._state._instance_id, ) else: session.owner_instance self._state._instance_id # 3. 追加用户消息 session.messages.append({ role: user, content: user_msg }) # 4. 执行 Agent 推理 response await self._execute_with_retry(session) # 5. 持久化状态 session.messages.append({ role: assistant, content: response }) session.status SessionStatus.COMPLETED await self._state.save_session(session) return {session_id: session_id, response: response} finally: await self._state.release_session_lock(session_id) async def _execute_with_retry(self, session: AgentSession) - str: 带重试的 Agent 执行 last_error None for attempt in range(self._max_retry): try: # 保存检查点 await self._state.save_session(session) # 执行推理这里接入实际的 LLM 调用 result await self._do_inference(session) return result except Exception as exc: last_error exc if attempt self._max_retry - 1: # 恢复到最后一次检查点 restored await self._state.load_session( session.session_id ) if restored: session restored await asyncio.sleep(2 ** attempt) raise RuntimeError( fAgent 执行失败已重试 {self._max_retry} 次: {last_error} ) async def _do_inference(self, session: AgentSession) - str: 实际的 Agent 推理逻辑简化版 # 这里接入 LLM、工具调用等 await asyncio.sleep(0.1) # 模拟推理延迟 return f处理完成 (步骤 {session.current_step 1}) # ── 4. 故障检测和自动转移 ───────────────────────── class FailoverWatcher: 故障检测和自动转移引擎 def __init__( self, state_manager: SessionStateManager, check_interval: float 5.0, ): self._state state_manager self._interval check_interval async def watch_loop(self): 持续监控并执行故障转移 while True: try: # 1. 更新自身心跳 await self._state.update_heartbeat() # 2. 查找孤儿会话 orphans await self._state.claim_orphan_sessions() # 3. 接管孤儿会话 for session_id in orphans: await self._failover_session(session_id) await asyncio.sleep(self._interval) except Exception as exc: # 故障检测自身不能挂 await asyncio.sleep(self._interval) async def _failover_session(self, session_id: str): 执行单个会话的故障转移 session await self._state.load_session(session_id) if session is None: return # 标记为故障转移中 session.status SessionStatus.FAILING_OVER await self._state.save_session(session) # 恢复到最后一个检查点 # 实际生产环境中这一步会重新执行最后一步操作 session.owner_instance self._state._instance_id session.status SessionStatus.ACTIVE await self._state.save_session(session) # ── 5. 启动入口 ──────────────────────────────────── async def start_ha_agent(): 启动高可用 Agent 服务 redis_client redis.from_url(redis://localhost:6379) state_manager SessionStateManager(redis_client) agent HighAvailabilityAgent(state_manager) watcher FailoverWatcher(state_manager) # 启动故障检测 asyncio.create_task(watcher.watch_loop()) # 启动心跳 asyncio.create_task(_heartbeat_loop(state_manager)) # 启动 HTTP 服务FastAPI 等 # ... async def _heartbeat_loop(state_manager: SessionStateManager): while True: await state_manager.update_heartbeat() await asyncio.sleep(10)四、边界分析与架构权衡一致性与可用性的抉择多实例 Agent 的高可用本质上是一个 CAP 问题。当网络分区发生时两个实例可能同时认为对方挂了都去接管同一个会话。这时必须用分布式锁保证同一时刻只有一个实例处理同一个会话。代价是极端情况下会牺牲一点可用性——锁超时时间内会话不可用。状态同步的粒度每次 Agent 执行步骤后都全量保存会话状态延迟敏感度高的场景受不了。折中方案是增量 全量高频步骤只追加增量新消息每秒做一次全量快照。如果故障发生从最近的全量快照恢复 增量重放。有工具调用正在进行时的故障转移如果 Agent 正在调用一个外部 API比如发起退款调用还没返回就挂了新实例接手后不能简单地重调。退款 API 可能已经执行了——重调会导致重复退款。解决方案所有的外部调用都要做幂等设计或者在调用前先记录意图到 Redis新实例检查到未完成的意图后再决定是重试还是查询结果。故障转移的延迟从故障检测到新会话完全恢复通常有 5-15 秒的延迟心跳超时 10s 故障转移处理 2-5s。这期间用户看到的可能是正在重连。如果业务不能接受这个延迟需要缩短心跳间隔但代价是更多的心跳网络开销。五、总结Agent 高可用的核心是状态外移 分布式锁 故障检测三位一体状态外移所有会话状态、任务状态持久化到 RedisAgent 实例本身无状态分布式锁用 Redis 锁保证同一会话同一时间只有一个实例在处理故障检测心跳检测 孤儿会话认领实现 5-15 秒级故障转移这不是微服务高可用的简单复制。Agent 的有状态特性让它的高可用设计更复杂但思路是清晰的让状态活在外面的保险箱里让 Agent 变成随时可替换的执行单元。记住凌晨两点的故障不会因为你准备了预案就不来。但它来了之后你的系统是在 10 秒内自愈还是等你从被窝里爬起来手动重启——这取决于你今天有没有做高可用设计。下一篇预告Agent 服务化架构将 Agent 封装为微服务的接口设计与性能考量。
