企业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

架构关键决策点：

1. DNS智能解析：通过健康检查结果自动切换流量，故障切换时间取决于DNS TTL
2. 无状态服务：Agent服务本身不保存会话状态，状态外置到Redis
3. 模型层双集群：每个区域独立部署模型集群，避免单集群故障影响全量
4. 数据库主从：异步复制，从库可承担读流量

对比分析：同城双活 vs 跨区域多活的成本与收益。同城双活：两个可用区部署，延迟 < 2ms，数据库可同步复制保证 RPO=0。月成本约为单区域部署的 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(endpoint=url)

    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",
                    timeout=aiohttp.ClientTimeout(total=5)
                ) 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, 
                            f"HTTP {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_ascii=False)

        # 主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 None

3.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

四、容灾演练的必要性

架构画得再好，没有演练就是纸上谈兵。每个季度至少执行一次容灾演练，验证以下场景：

1. 单区域故障：手动切断一个区域的流量，验证另一个区域能否独立承载
2. 数据库切换：执行主从切换，记录切换时间和数据一致性
3. 模型集群降级：关闭一个模型集群，验证降级策略（使用更小模型/限流）

演练必须记录三个指标：

五、总结

企业Agent的高可用建设有三层：接入层通过DNS+负载均衡做流量分发，服务层通过无状态设计+熔断器做故障隔离，数据层通过主从复制+会话迁移做状态保护。

对于创业团队，先用最小成本实现高可用：同区域双实例 + 数据库主从 + Redis Sentinel。等有了第一个企业客户后，再逐步扩展到跨区域多活。可用性建设是为客户的业务连续性买的保险，这个保险不能等出险之后才想起买。

要点提炼：

1. 先做到同城双活。两个可用区 + 健康检查 + 自动切换，覆盖 95% 故障场景。
2. Agent 服务必须无状态。会话状态外置到 Redis，实例故障后其他节点可接管。
3. 健康检查用连续失败阈值。单次失败可能是网络抖动，连续 3 次才算真故障。
4. 熔断器保护模型调用。GPU 资源不可弹性扩展，熔断能防止雪崩。
5. 每季度做容灾演练。记录故障检测时间、流量切换时间、服务恢复时间三个指标。
6. 会话状态跨区域同步不能阻塞主链路。备用 Redis 写入失败记录告警，不可阻断用户请求。