Agent 高可用设计：多实例部署下的状态同步和故障转移策略

一、深度引言与场景痛点

大家好，我是赵咕咕。一个真实的故障故事：凌晨两点，我们的 AI 客服 Agent 服务因为一个第三方地图 API 的超时，导致整个 Agent 进程卡死。因为这个 Agent 是单实例部署的，它一挂，所有的用户会话全断了。用户那边显示的是永远的"AI 正在思考中…"。

这个故障暴露了三个问题：

1. 单实例没有任何冗余，挂了就是全体掉线
2. Agent 的会话状态都在内存里，即使能快速重启一个新的，老会话也回不去了
3. 没有故障转移机制，用户在断线后只能从头开始对话

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 Prompt<br/>工具定义<br/>模型配置] E --> E2[特点：低频变更<br/>所有实例相同<br/>配置中心管理] style C2 fill:#ffcdd2 style D2 fill:#fff9c4 style E2 fill:#c8e6c9

Agent 的高可用设计，本质上是把会话状态和任务状态从进程内存外移到外部存储，然后让任何健康的实例都能接手。

2.2 多实例架构

flowchart TD LB[负载均衡器<br/>Nginx/Traefik] --> A1[Agent 实例 1] LB --> A2[Agent 实例 2] LB --> A3[Agent 实例 N] A1 --> Redis[Redis<br/>会话状态存储<br/>分布式锁] A2 --> Redis A3 --> Redis A1 --> DB[(PostgreSQL<br/>持久化存储<br/>任务日志)] A2 --> DB A3 --> DB A1 -.->|心跳| HW[Health Watcher<br/>健康检查/故障转移] A2 -.->|心跳| HW A3 -.->|心跳| HW HW -->|检测到故障| LB HW -->|故障通知| Alert[告警系统] style Redis fill:#dc3545,color:white style DB fill:#336791,color:white style HW fill:#ffc107

2.3 故障转移流程

sequenceDiagram participant U as 用户 participant LB as 负载均衡器 participant A1 as Agent 实例 1<br/>(运行中) participant A2 as Agent 实例 2<br/>(备用) 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_factory=lambda: str(uuid.uuid4())) user_id: str = "" status: SessionStatus = SessionStatus.ACTIVE # 对话历史 messages: list[dict] = field(default_factory=list) # 当前任务状态 current_step: int = 0 pending_tool_calls: list[dict] = field(default_factory=list) current_plan: list[dict] = field(default_factory=list) # 实例信息 owner_instance: str = "" # 当前负责的实例 ID created_at: str = field( default_factory=lambda: datetime.now(timezone.utc).isoformat() ) updated_at: str = field( default_factory=lambda: 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, mapping=data) # 设置过期时间防止僵尸会话 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_id=data.get("session_id", session_id), user_id=data.get("user_id", ""), status=SessionStatus(data.get("status", "active")), messages=json.loads(data.get("messages", "[]")), current_step=int(data.get("current_step", 0)), pending_tool_calls=json.loads(data.get("pending_tool_calls", "[]")), current_plan=json.loads(data.get("current_plan", "[]")), owner_instance=data.get("owner_instance", ""), version=int(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, nx=True, ex=ttl ) 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, match=f"{self.SESSION_PREFIX}*", count=100 ) 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_id=session_id, owner_instance=self._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( f"Agent 执行失败，已重试 {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 高可用的核心是状态外移 + 分布式锁 + 故障检测，三位一体：

1. 状态外移：所有会话状态、任务状态持久化到 Redis，Agent 实例本身无状态
2. 分布式锁：用 Redis 锁保证同一会话同一时间只有一个实例在处理
3. 故障检测：心跳检测 + 孤儿会话认领，实现 5-15 秒级故障转移

这不是微服务高可用的简单复制。Agent 的"有状态"特性让它的高可用设计更复杂，但思路是清晰的：让状态活在外面的保险箱里，让 Agent 变成随时可替换的执行单元。

记住：凌晨两点的故障不会因为你准备了预案就不来。但它来了之后，你的系统是在 10 秒内自愈，还是等你从被窝里爬起来手动重启——这取决于你今天有没有做高可用设计。

下一篇预告：Agent 服务化架构，将 Agent 封装为微服务的接口设计与性能考量。