企业Agent的高可用架构：多活部署、故障转移与容灾演练

企业Agent的高可用架构多活部署、故障转移与容灾演练一、可用性是企业Agent的第一道门槛B端客户对AI产品有一个隐性的假设它应该像一个API一样可靠。当企业把Agent接入他们的核心业务流程后可用性问题就不再是内部故障而是客户的生产事故。一次服务中断可能意味着客户的工单系统挂了客户的问答机器人停止响应客户的自动化流程卡住了。这对B端产品是毁灭性的——一次中断可能毁掉三个月的信任建设。企业Agent的高可用不是一个以后再做的任务。它应该在你签下第一个企业客户之前就完成基本建设。二、企业Agent的高可用架构graph TD subgraph 用户层 DNS[DNS智能解析] end subgraph 接入层 DNS -- LB1[负载均衡 - 区域A] DNS -- LB2[负载均衡 - 区域B] LB1 -- GW1[API网关集群A] LB2 -- GW2[API网关集群B] end subgraph 服务层 GW1 -- SvcA1[Agent服务实例A1] GW1 -- SvcA2[Agent服务实例A2] GW2 -- SvcB1[Agent服务实例B1] GW2 -- SvcB2[Agent服务实例B2] end subgraph 模型层 SvcA1 -- LLM1[LLM集群A] SvcA2 -- LLM1 SvcB1 -- LLM2[LLM集群B] SvcB2 -- LLM2 end subgraph 数据层 DB1[(主库 - 区域A)] DB2[(从库 - 区域B)] DB1 --|异步复制| DB2 Cache1[(Redis A)] Cache2[(Redis B)] end subgraph 调度层 HB[健康检查] -- |心跳检测| SvcA1 HB -- |心跳检测| SvcB1 HB -- |自动切换| DNS end style DNS fill:#ff6b6b,color:#fff style HB fill:#ffd93d style LB1 fill:#6c5ce7,color:#fff style LB2 fill:#6c5ce7,color:#fff架构关键决策点DNS智能解析通过健康检查结果自动切换流量故障切换时间取决于DNS TTL无状态服务Agent服务本身不保存会话状态状态外置到Redis模型层双集群每个区域独立部署模型集群避免单集群故障影响全量数据库主从异步复制从库可承担读流量对比分析同城双活 vs 跨区域多活的成本与收益。同城双活两个可用区部署延迟 2ms数据库可同步复制保证 RPO0。月成本约为单区域部署的 1.8 倍。跨区域多活两个城市部署延迟 20-50ms数据库只能异步复制RPO 可能有数秒丢失。月成本约为单区域部署的 2.5 倍。关键差异不在成本而在取舍同城双活的单城市级灾难地震、大面积断电仍然会导致全站不可用。跨区域多活能抗城市级灾难但代价是数据一致性只能做到最终一致。创业团队的第一目标应该是同城双活同一云厂商的两个可用区成本可控能覆盖 95% 的故障场景单机房断电、网络分区、单机故障。只有当客户合同里写了城市级灾备 SLA时才考虑跨区域多活。三、生产级实现3.1 健康检查与自动故障转移import asyncio import aiohttp import time from typing import Dict, List, Optional from dataclasses import dataclass, field from collections import defaultdict dataclass class HealthStatus: 服务健康状态 endpoint: str healthy: bool True last_success: float 0.0 last_failure: float 0.0 consecutive_failures: int 0 response_time_ms: float 0.0 class HealthCheckEngine: 健康检查引擎 核心职责周期性探测所有服务端点维护健康状态。 为什么用consecutive_failures而非single_failure 网络抖动可能导致单次失败连续失败才是真正的故障。 这个阈值防止误切换。 def __init__(self, check_interval: int 5): self.interval check_interval # 检查间隔秒数 self.fail_threshold 3 # 连续失败阈值 self.endpoints: Dict[str, HealthStatus] {} self.callbacks: List[callable] [] # 状态变更回调 def register_endpoint(self, name: str, url: str) - None: 注册需要监控的端点 self.endpoints[name] HealthStatus(endpointurl) def on_status_change(self, callback): 注册状态变更回调 self.callbacks.append(callback) async def start(self): 启动健康检查循环 while True: tasks [] for name, status in self.endpoints.items(): tasks.append(self._check_one(name, status)) await asyncio.gather(*tasks) await asyncio.sleep(self.interval) async def _check_one(self, name: str, status: HealthStatus): 检查单个端点 start time.time() try: async with aiohttp.ClientSession() as session: # 健康检查端点必须轻量级 # 为什么是/health而非/api # 检查应测试连接性和基本响应不应触发业务逻辑 async with session.get( f{status.endpoint}/health, timeoutaiohttp.ClientTimeout(total5) ) as resp: if resp.status 200: latency (time.time() - start) * 1000 was_unhealthy not status.healthy status.healthy True status.consecutive_failures 0 status.last_success time.time() status.response_time_ms latency # 从故障中恢复触发回调 if was_unhealthy: for cb in self.callbacks: await cb(recovered, name, latency) else: await self._handle_failure(name, status, fHTTP {resp.status}) except Exception as e: await self._handle_failure(name, status, str(e)) async def _handle_failure( self, name: str, status: HealthStatus, error: str ): 处理健康检查失败 status.consecutive_failures 1 status.last_failure time.time() if status.consecutive_failures self.fail_threshold: if status.healthy: status.healthy False # 触发故障转移回调 for cb in self.callbacks: await cb(unhealthy, name, error)3.2 会话状态的跨区域迁移import json from typing import Dict, Optional from redis import Redis from datetime import datetime class SessionMigrationManager: 跨区域会话迁移管理 为什么Agent会话需要跨区域迁移 如果用户在区域A建立的会话区域A故障后切换到区域B 用户不应该丢失上下文。这就要求会话状态跨区域同步。 def __init__( self, primary_redis: Redis, secondary_redis: Redis, sync_batch_size: int 100 ): self.primary primary_redis self.secondary secondary_redis self.batch_size sync_batch_size self.SESSION_PREFIX agent:session: self.SESSION_TTL 3600 # 会话过期时间1小时 def save_session( self, session_id: str, context: Dict, messages: list ) - bool: 保存会话状态到主Redis并同步到备用Redis。 为什么双向写入而非异步复制 Redis Cluster的异步复制在跨区域场景延迟较高。 双向写入可以在切换后保证数据已就位。 代价是写入延迟翻倍对于非高频更新可容忍。 session_key f{self.SESSION_PREFIX}{session_id} session_data json.dumps({ context: context, messages: messages[-20:], # 只保留最近20条消息 # 为什么截断消息历史 # 全量消息在跨区域同步时占用大量带宽。 # 近20条足够恢复对话上下文。 updated_at: datetime.utcnow().isoformat() }, ensure_asciiFalse) # 主Redis写入 self.primary.setex(session_key, self.SESSION_TTL, session_data) # 备用Redis异步写入 # 为什么用异步 # 备用写入失败不应阻塞主链路。 try: self.secondary.setex( session_key, self.SESSION_TTL, session_data ) except Exception: # 异步写入失败记录到监控系统 # 不能阻塞主链路但需要记录以便同步补偿 pass return True def restore_session( self, session_id: str, prefer_primary: bool True ) - Optional[Dict]: 恢复会话状态。 prefer_primary: 优先从主Redis读失败则从备用读。 为什么需要fallback 正常场景下主Redis响应更快但故障时需要能从备用恢复。 session_key f{self.SESSION_PREFIX}{session_id} # 优先主Redis if prefer_primary: try: data self.primary.get(session_key) if data: return json.loads(data) except Exception: pass # 降级备用Redis try: data self.secondary.get(session_key) if data: return json.loads(data) except Exception: pass return None3.3 熔断器保护import time import asyncio from enum import Enum from functools import wraps class CircuitState(Enum): CLOSED closed # 正常通行 OPEN open # 断路直接拒绝 HALF_OPEN half_open # 半开尝试恢复 class CircuitBreaker: 熔断器 为什么LLM调用需要熔断器 GPU资源不可弹性扩展。如果模型服务压力过大导致响应变慢 调用方持续堆积请求会加速崩溃。 熔断器可以阻止雪崩。 def __init__( self, failure_threshold: int 5, # 连续失败阈值 timeout: float 60.0, # 断路后的恢复等待时间 half_open_max_calls: int 3 # 半开状态允许的试探调用数 ): self.failure_threshold failure_threshold self.timeout timeout self.half_open_max_calls half_open_max_calls self.state CircuitState.CLOSED self.failure_count 0 self.last_failure_time 0.0 self.half_open_calls 0 async def call(self, func, *args, **kwargs): 受熔断器保护的调用。 状态机转换逻辑 CLOSED → (连续失败到达阈值) → OPEN OPEN → (超时后) → HALF_OPEN HALF_OPEN → (成功) → CLOSED HALF_OPEN → (失败) → OPEN if self.state CircuitState.OPEN: if time.time() - self.last_failure_time self.timeout: # 超时后进入半开状态 self.state CircuitState.HALF_OPEN self.half_open_calls 0 else: raise CircuitBreakerOpenError( f熔断器已断开{self.timeout - (time.time() - self.last_failure_time):.0f}秒后恢复 ) if self.state CircuitState.HALF_OPEN: if self.half_open_calls self.half_open_max_calls: raise CircuitBreakerOpenError(熔断器试探调用已达上限) try: result await func(*args, **kwargs) # 调用成功 if self.state CircuitState.HALF_OPEN: self.state CircuitState.CLOSED self.failure_count 0 elif self.state CircuitState.CLOSED: # 成功调用重置失败计数 self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if (self.state CircuitState.CLOSED and self.failure_count self.failure_threshold): self.state CircuitState.OPEN if self.state CircuitState.HALF_OPEN: self.state CircuitState.OPEN self.last_failure_time time.time() raise class CircuitBreakerOpenError(Exception): 熔断器断开异常 pass四、容灾演练的必要性架构画得再好没有演练就是纸上谈兵。每个季度至少执行一次容灾演练验证以下场景单区域故障手动切断一个区域的流量验证另一个区域能否独立承载数据库切换执行主从切换记录切换时间和数据一致性模型集群降级关闭一个模型集群验证降级策略使用更小模型/限流演练必须记录三个指标指标目标原因故障检测时间 30秒健康检查间隔连续失败阈值决定流量切换时间 60秒DNS TTL 路由更新服务恢复时间 5分钟从检测到恢复正常的时间五、总结企业Agent的高可用建设有三层接入层通过DNS负载均衡做流量分发服务层通过无状态设计熔断器做故障隔离数据层通过主从复制会话迁移做状态保护。对于创业团队先用最小成本实现高可用同区域双实例 数据库主从 Redis Sentinel。等有了第一个企业客户后再逐步扩展到跨区域多活。可用性建设是为客户的业务连续性买的保险这个保险不能等出险之后才想起买。要点提炼先做到同城双活。两个可用区 健康检查 自动切换覆盖 95% 故障场景。Agent 服务必须无状态。会话状态外置到 Redis实例故障后其他节点可接管。健康检查用连续失败阈值。单次失败可能是网络抖动连续 3 次才算真故障。熔断器保护模型调用。GPU 资源不可弹性扩展熔断能防止雪崩。每季度做容灾演练。记录故障检测时间、流量切换时间、服务恢复时间三个指标。会话状态跨区域同步不能阻塞主链路。备用 Redis 写入失败记录告警不可阻断用户请求。