AI Agent Harness Engineering 的降级策略：核心能力保底与优雅降级

副标题：构建高可用性AI智能体系统的工程实践与方法论

---

第一部分：引言与基础

1. 摘要/引言

在当今快速发展的人工智能领域，AI智能体（AI Agents）正逐渐成为各种应用系统的核心组件。从智能客服到自动化决策系统，从个人助手到工业控制，AI智能体正在改变我们与技术互动的方式。然而，随着AI智能体在关键业务场景中的广泛应用，系统可靠性和稳定性问题日益突出。

问题陈述

AI智能体系统面临着多重挑战：模型性能波动、API调用失败、资源限制、网络问题、数据异常等，这些都可能导致系统性能下降甚至完全失效。在传统软件系统中，我们已经有成熟的降级策略来应对类似问题，但AI智能体系统的特殊性使得这些传统方法不能直接套用。

核心方案

本文提出了一套专门针对AI Agent Harness Engineering的降级策略框架，重点关注核心能力保底与优雅降级两个关键维度。我们将探讨如何在AI智能体系统中设计和实现有效的降级机制，确保系统在面对各种异常情况时仍能提供基本服务，并随着条件改善逐步恢复全部功能。

主要成果/价值

阅读本文后，你将：

• 理解AI智能体系统中降级策略的重要性和特殊性
• 掌握核心能力保底与优雅降级的设计原则和实现方法
• 学习如何评估和选择适合特定场景的降级策略
• 获取可直接应用于实际项目的代码示例和最佳实践
• 了解AI智能体降级策略的未来发展趋势

文章导览

本文将分为四个主要部分：首先介绍AI智能体降级策略的基础概念和背景；然后深入探讨核心能力保底与优雅降级的设计与实现；接着讨论验证方法、性能优化和常见问题；最后总结全文并展望未来发展方向。

---

2. 目标读者与前置知识

目标读者

本文主要面向以下读者群体：

• AI/ML工程师：负责设计和实现AI智能体系统的专业人士
• 后端/系统工程师：关注系统可靠性和高可用性的工程师
• 技术架构师：需要设计弹性AI系统架构的决策者
• DevOps/SRE工程师：负责AI系统运维和可靠性保障的专业人员
• 对AI系统工程实践感兴趣的技术爱好者

前置知识

为了更好地理解本文内容，建议读者具备以下基础知识：

• 基本的机器学习和深度学习概念
• 熟悉Python编程
• 了解微服务架构和API设计基础
• 对系统可靠性工程有基本认识
• 了解容器化技术（如Docker）和编排工具（如Kubernetes）将是加分项

---

3. 文章目录

1. 第一部分：引言与基础

   • 摘要/引言
   • 目标读者与前置知识
   • 文章目录

2. 第二部分：核心内容

   • 问题背景与动机
   • 核心概念与理论基础
   • 环境准备
   • 分步实现：核心能力保底设计
   • 分步实现：优雅降级机制
   • 关键代码解析与深度剖析

3. 第三部分：验证与扩展

   • 结果展示与验证
   • 性能优化与最佳实践
   • 常见问题与解决方案
   • 未来展望与扩展方向

4. 第四部分：总结与附录

   • 总结
   • 参考资料
   • 附录

---

第二部分：核心内容

4. 问题背景与动机

AI智能体系统的普及与挑战

近年来，AI智能体技术取得了长足进步，从简单的规则驱动系统发展到基于大语言模型（LLMs）的复杂智能体。这些智能体被广泛应用于客户服务、内容生成、数据分析、自动化决策等多个领域。根据Gartner的预测，到2025年，超过50%的企业将在其业务流程中部署某种形式的AI智能体。

然而，随着AI智能体在关键业务场景中的深入应用，我们也面临着前所未有的挑战：

1. 模型不确定性：与传统软件系统不同，AI模型的输出具有概率性，即使在正常情况下也可能产生不准确或不适当的结果。

2. 资源依赖性：现代AI模型，特别是大型语言模型，通常需要大量的计算资源和内存，这使得它们对资源波动非常敏感。

3. 外部服务依赖：许多AI智能体系统依赖于外部API服务（如OpenAI、Anthropic等），这些服务可能会遭遇 downtime、速率限制或性能下降。

4. 数据质量问题：AI模型的性能高度依赖于输入数据的质量，异常或低质量的数据可能导致模型表现急剧下降。

5. 复杂性导致的脆弱性：现代AI智能体系统通常由多个组件组成（模型、向量数据库、工具调用模块等），系统的复杂性增加了故障点。

传统降级策略的局限性

在传统软件系统中，我们已经发展出了一套成熟的降级策略，如：

• 熔断机制（Circuit Breaker）
• 限流（Rate Limiting）
• 备用服务（Fallback）
• 负载 shedding

然而，AI智能体系统的特殊性使得这些传统方法不能直接套用：

1. AI系统的输出不是二元的：传统系统通常可以明确区分"工作"和"不工作"状态，而AI系统的性能是连续的，从"完美"到"完全无用"之间有很多中间状态。

2. 降级的目标不同：传统系统降级通常关注"保持系统运行"，而AI系统降级还需要考虑"保持输出质量在可接受范围内"。

3. 降级决策更复杂：AI系统的降级决策需要考虑更多因素，如模型性能指标、用户体验影响、业务价值等。

为什么我们需要专门的AI Agent降级策略

基于以上分析，我们可以清楚地看到，AI智能体系统需要一套专门设计的降级策略，原因如下：

1. 业务连续性保障：在关键业务场景中，即使AI智能体不能提供最优服务，也需要确保基本功能可用。

2. 用户体验管理：智能地降级可以帮助管理用户期望，避免用户因突然的性能下降而感到失望。

3. 资源优化：通过降级策略，可以在资源受限的情况下优先保障核心功能。

4. 成本控制：AI服务通常按使用量计费，有效的降级策略可以帮助控制成本。

5. 风险 mitigation：在模型输出质量下降时，及时降级可以避免潜在的业务风险。

---

5. 核心概念与理论基础

在深入探讨AI Agent降级策略的实现之前，我们首先需要明确一些核心概念和理论基础。

核心概念

5.1 AI Agent Harness Engineering

AI Agent Harness Engineering是指设计、构建和管理AI智能体系统的工程实践，重点关注系统的可靠性、可维护性和可扩展性。这个概念强调将软件工程的最佳实践应用于AI系统开发，弥补了传统AI研究与实际生产部署之间的差距。

5.2 降级策略（Degradation Strategy）

降级策略是指当系统组件出现故障或性能下降时，系统采取的一系列措施，以确保核心功能仍然可用，并尽可能保持良好的用户体验。在AI智能体系统中，降级策略不仅要处理完全故障的情况，还要处理性能下降的情况。

5.3 核心能力保底（Core Capability Assurance）

核心能力保底是指识别AI智能体系统的最关键功能，并确保即使在系统其他部分出现问题时，这些核心功能仍然能够正常工作。这需要对业务价值和用户需求有深入的理解。

5.4 优雅降级（Graceful Degradation）

优雅降级是指系统在面对问题时，不是突然完全失效，而是逐步降低功能级别，尽可能保持系统可用，并随着条件改善逐步恢复全部功能。优雅降级的目标是平滑过渡，减少对用户的负面影响。

5.5 性能阈值（Performance Threshold）

性能阈值是指系统性能指标的可接受范围，当性能指标超出这个范围时，系统将触发降级策略。在AI智能体系统中，性能阈值可能包括模型准确度、响应时间、置信度等多个维度。

5.6 降级级别（Degradation Levels）

降级级别是指系统从全功能到最小可用功能之间的不同状态。每个级别定义了一组可用的功能和性能特征。定义清晰的降级级别是实现优雅降级的基础。

概念结构与核心要素组成

AI Agent降级策略系统由以下核心要素组成：

1. 监控模块：负责持续监控系统性能指标、资源使用情况和外部依赖状态。
2. 评估模块：根据监控数据评估系统当前状态，判断是否需要触发降级。
3. 降级策略库：存储不同场景下的降级策略和配置。
4. 执行引擎：负责实际执行降级策略，调整系统行为。
5. 恢复机制：监控系统状态改善情况，触发功能恢复。
6. 反馈循环：收集降级后的系统表现和用户反馈，用于优化降级策略。

我们可以用以下Mermaid架构图来表示这些核心要素之间的关系：

概念之间的关系：属性维度对比

为了更好地理解AI Agent降级策略中的关键概念，我们可以从多个维度对核心能力保底和优雅降级进行对比：

维度	核心能力保底	优雅降级
主要目标	确保最关键功能始终可用	平滑过渡，最小化用户体验影响
关注重点	功能完整性	性能连续性
触发时机	严重故障时	性能下降时
变化模式	二元变化（有/无）	连续变化（级别调整）
决策依据	业务重要性	性能指标和用户体验
恢复策略	快速恢复到完整功能	逐步恢复，每个级别验证
复杂度	相对简单	较高，需要多级状态管理
用户可见性	通常用户可见	尽可能对用户透明

AI Agent降级策略的ER实体关系图

为了更清晰地表示AI Agent降级策略中各实体之间的关系，我们可以使用以下ER图：

数学模型：降级决策的量化方法

在AI Agent降级策略中，我们需要一种量化方法来决定何时触发降级以及降级到哪个级别。我们可以使用多准则决策分析（MCDA）来构建这个模型。

首先，我们定义一组评估指标 X={x1,x2,...,xn}X = \{x_1, x_2, ..., x_n\}X={x1​,x2​,...,xn​}，每个指标代表系统性能的一个方面，如响应时间、模型置信度、资源使用率等。

然后，我们为每个指标分配权重 W={w1,w2,...,wn}W = \{w_1, w_2, ..., w_n\}W={w1​,w2​,...,wn​}，表示该指标在降级决策中的重要性，满足 ${\sum }_{i=1}^n{w}_i=1$。

接下来，我们为每个指标定义一个隶属函数 ${\mu }_i(x_i)$，表示指标值 $x_i$ 对系统健康度的隶属程度，取值范围为[0, 1]，其中1表示最佳状态，0表示最差状态。

系统的整体健康度 $H$ 可以计算为：

$$H=\sum _{i=1}^n{w}_i\cdot {\mu }_i(x_i)$$

最后，我们定义一组健康度阈值 T={t1,t2,...,tm}T = \{t_1, t_2, ..., t_m\}T={t1​,t2​,...,tm​}，对应不同的降级级别 L={L0,L1,...,Lm}L = \{L_0, L_1, ..., L_m\}L={L0​,L1​,...,Lm​}，其中 $L_0$ 表示全功能状态，$L_m$ 表示最小可用状态。降级决策规则为：

$$\text{选择级别 }{L}_k\text{ 当且仅当 }{t}_k\le H<t_{k-1}$$

其中 $t_0=1$，$t_m=0$。

这个模型为我们提供了一个量化的框架，可以根据系统状态做出客观的降级决策。

---

6. 环境准备

在开始实现AI Agent降级策略之前，我们需要准备一个合适的开发环境。本节将介绍所需的软件、库和工具，并提供配置指南。

软件与工具清单

我们将使用以下软件和工具来构建我们的AI Agent降级策略系统：

1. Python 3.9+：我们的主要开发语言
2. Docker & Docker Compose：用于容器化部署和服务编排
3. Prometheus：用于监控指标收集
4. Grafana：用于监控数据可视化
5. Redis：用于状态存储和缓存
6. FastAPI：用于构建API服务
7. OpenAI SDK：用于与大语言模型交互（示例用）
8. pytest：用于测试

Python依赖库

我们将使用以下Python库：

fastapi==0.104.1
uvicorn==0.24.0
python-dotenv==1.0.0
openai==1.3.5
redis==5.0.1
prometheus-client==0.19.0
pydantic==2.5.2
tenacity==8.2.3
structlog==23.2.0

你可以将这些依赖保存到 requirements.txt 文件中，然后使用以下命令安装：

pip install -r requirements.txt

Docker Compose配置

为了简化环境设置，我们将使用Docker Compose来管理我们的服务。以下是一个基本的 docker-compose.yml 文件：

version: '3.8'

services:
  ai-agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - redis
      - prometheus
    volumes:
      - ./:/app

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  prometheus:
    image: prom/prometheus:v2.45.0
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus

  grafana:
    image: grafana/grafana:10.2.0
    ports:
      - "3000:3000"
    volumes:
      - grafana_data:/var/lib/grafana
    depends_on:
      - prometheus

volumes:
  redis_data:
  prometheus_data:
  grafana_data:

同时，我们需要创建一个Prometheus配置文件 prometheus.yml：

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'ai-agent'
    static_configs:
      - targets: ['ai-agent:8000']

项目结构

我们将使用以下项目结构：

ai-agent-degradation/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI应用入口
│   ├── agent/
│   │   ├── __init__.py
│   │   ├── base.py             # 基础Agent类
│   │   ├── core.py             # 核心Agent实现
│   │   └── degraded.py         # 降级Agent实现
│   ├── degradation/
│   │   ├── __init__.py
│   │   ├── manager.py          # 降级管理器
│   │   ├── strategies.py       # 降级策略实现
│   │   └── levels.py           # 降级级别定义
│   ├── monitoring/
│   │   ├── __init__.py
│   │   ├── metrics.py          # 指标定义
│   │   └── collector.py        # 指标收集器
│   └── utils/
│       ├── __init__.py
│       ├── config.py           # 配置管理
│       └── state.py            # 状态管理
├── tests/
│   ├── __init__.py
│   ├── test_degradation.py     # 降级策略测试
│   └── test_agent.py           # Agent测试
├── Dockerfile
├── docker-compose.yml
├── prometheus.yml
├── requirements.txt
└── .env.example

在接下来的部分中，我们将逐步实现这个系统的各个组件。

---

7. 分步实现：核心能力保底设计

核心能力保底是AI Agent降级策略的基础，它确保即使在系统出现严重问题时，最关键的功能仍然可用。在本节中，我们将详细介绍如何设计和实现核心能力保底机制。

7.1 识别核心能力

在设计核心能力保底机制之前，我们首先需要明确哪些是AI Agent的核心能力。这通常需要与业务利益相关者合作，基于以下几个维度进行评估：

1. 业务价值：该功能对业务的重要性程度
2. 用户期望：用户对该功能的依赖程度
3. 实现复杂度：在资源受限情况下实现该功能的难度
4. 替代方案：如果该功能不可用，是否有其他方式满足需求

让我们以一个客户服务AI Agent为例，它可能具有以下功能：

1. 回答常见问题（高价值，高用户期望）
2. 处理退货请求（高价值，中用户期望）
3. 产品推荐（中价值，中用户期望）
4. 情感分析与个性化回应（低价值，低用户期望）
5. 多语言翻译（中价值，低用户期望）

在这个例子中，"回答常见问题"显然是核心能力，我们需要确保即使在系统降级时，这个功能仍然可用。

7.2 定义核心能力级别

一旦我们识别了核心能力，我们需要为其定义不同的实现级别，以便在资源受限的情况下能够提供不同程度的功能。对于"回答常见问题"这个核心能力，我们可以定义以下级别：

1. 级别0（完整功能）：使用最新的LLM模型，能够理解复杂问题，提供个性化回答
2. 级别1（简化功能）：使用较小的模型或规则系统，能够回答标准问题，但可能不够灵活
3. 级别2（最小功能）：基于关键词匹配的预定义回答，只能处理最常见的问题

7.3 实现核心能力保底框架

现在，让我们开始实现核心能力保底框架。首先，我们将定义降级级别：

# app/degradation/levels.py
from enum import Enum
from pydantic import BaseModel, Field
from typing import Dict, Any, Optional, Callable


class DegradationLevel(Enum):
    FULL = 0          # 完整功能
    REDUCED = 1       # 简化功能
    MINIMAL = 2       # 最小功能
    UNAVAILABLE = 3   # 不可用


class LevelCapabilities(BaseModel):
    """定义每个降级级别的能力"""
    level: DegradationLevel
    name: str
    description: str
    # 该级别支持的功能列表
    capabilities: list[str]
    # 性能指标阈值
    performance_thresholds: Dict[str, float]
    # 资源需求
    resource_requirements: Dict[str, float]
    # 实现类或函数的引用
    implementation: Optional[Callable] = None


# 定义各个级别的能力
FULL_CAPABILITIES = LevelCapabilities(
    level=DegradationLevel.FULL,
    name="Full Functionality",
    description="Complete AI agent capabilities with advanced features",
    capabilities=[
        "complex_query_handling",
        "contextual_understanding",
        "personalized_responses",
        "multi_turn_conversation",
        "tool_integration"
    ],
    performance_thresholds={
        "response_time": 5.0,  # 最大响应时间（秒）
        "confidence": 0.7,     # 最小置信度
        "availability": 0.99   # 最小可用性
    },
    resource_requirements={
        "cpu": 2.0,
        "memory": 4.0,  # GB
        "model_size": "large"
    }
)

REDUCED_CAPABILITIES = LevelCapabilities(
    level=DegradationLevel.REDUCED,
    name="Reduced Functionality",
    description="Core AI agent capabilities with some advanced features disabled",
    capabilities=[
        "standard_query_handling",
        "basic_context",
        "predefined_responses"
    ],
    performance_thresholds={
        "response_time": 2.0,
        "confidence": 0.6,
        "availability": 0.995
    },
    resource_requirements={
        "cpu": 1.0,
        "memory": 2.0,  # GB
        "model_size": "medium"
    }
)

MINIMAL_CAPABILITIES = LevelCapabilities(
    level=DegradationLevel.MINIMAL,
    name="Minimal Functionality",
    description="Only the most essential capabilities using simple rule-based systems",
    capabilities=[
        "faq_matching",
        "basic_information"
    ],
    performance_thresholds={
        "response_time": 0.5,
        "confidence": 0.5,
        "availability": 1.0
    },
    resource_requirements={
        "cpu": 0.5,
        "memory": 0.5,  # GB
        "model_size": "none"
    }
)

# 创建级别映射
LEVEL_CAPABILITIES_MAP = {
    DegradationLevel.FULL: FULL_CAPABILITIES,
    DegradationLevel.REDUCED: REDUCED_CAPABILITIES,
    DegradationLevel.MINIMAL: MINIMAL_CAPABILITIES
}

接下来，我们将实现不同级别的Agent：

# app/agent/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from pydantic import BaseModel


class AgentRequest(BaseModel):
    """Agent请求模型"""
    query: str
    context: Optional[Dict[str, Any]] = None
    user_id: Optional[str] = None
    session_id: Optional[str] = None


class AgentResponse(BaseModel):
    """Agent响应模型"""
    response: str
    confidence: float
    sources: Optional[list[str]] = None
    metadata: Optional[Dict[str, Any]] = None
    level_used: int  # 使用的降级级别


class BaseAgent(ABC):
    """Agent基类"""
    
    @abstractmethod
    async def process(self, request: AgentRequest) -> AgentResponse:
        """处理请求并返回响应"""
        pass
    
    @abstractmethod
    def get_capabilities(self) -> list[str]:
        """获取该Agent支持的能力列表"""
        pass
    
    @abstractmethod
    def get_level(self) -> int:
        """获取该Agent对应的降级级别"""
        pass

现在，我们实现三个不同级别的Agent：

# app/agent/core.py
import os
import openai
from typing import Dict, Any, Optional
from .base import BaseAgent, AgentRequest, AgentResponse
from ..degradation.levels import DegradationLevel
from ..utils.config import settings
import structlog

logger = structlog.get_logger()


class FullFunctionalityAgent(BaseAgent):
    """全功能Agent - 使用高级LLM"""
    
    def __init__(self):
        self.client = openai.AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
        self.model = settings.OPENAI_MODEL or "gpt-4"
        logger.info("FullFunctionalityAgent initialized", model=self.model)
    
    async def process(self, request: AgentRequest) -> AgentResponse:
        """使用高级LLM处理请求"""
        try:
            # 构建消息
            messages = [
                {"role": "system", "content": "You are a helpful customer service assistant."}
            ]
            
            # 添加上下文
            if request.context:
                context_str = "\n".join([f"{k}: {v}" for k, v in request.context.items()])
                messages.append({"role": "system", "content": f"Context:\n{context_str}"})
            
            # 添加用户查询
            messages.append({"role": "user", "content": request.query})
            
            # 调用LLM
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.7,
                max_tokens=1000
            )
            
            # 生成响应
            return AgentResponse(
                response=response.choices[0].message.content,
                confidence=0.85,  # 假设的置信度，实际应该基于模型输出计算
                sources=["openai_gpt4"],
                metadata={"model": self.model, "usage": response.usage.model_dump()},
                level_used=DegradationLevel.FULL.value
            )
            
        except Exception as e:
            logger.error("Error in FullFunctionalityAgent", error=str(e))
            raise
    
    def get_capabilities(self) -> list[str]:
        """获取该Agent支持的能力列表"""
        return [
            "complex_query_handling",
            "contextual_understanding",
            "personalized_responses",
            "multi_turn_conversation",
            "tool_integration"
        ]
    
    def get_level(self) -> int:
        """获取该Agent对应的降级级别"""
        return DegradationLevel.FULL.value


class ReducedFunctionalityAgent(BaseAgent):
    """简化功能Agent - 使用较小模型或简化提示"""
    
    def __init__(self):
        self.client = openai.AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
        self.model = "gpt-3.5-turbo"  # 使用较小的模型
        self.predefined_responses = {
            "shipping": "Standard shipping takes 3-5 business days. Express shipping takes 1-2 business days.",
            "return": "You can return items within 30 days of purchase. Please visit our returns page for more information.",
            "hours": "Our customer service hours are Monday to Friday, 9 AM to 5 PM EST."
        }
        logger.info("ReducedFunctionalityAgent initialized", model=self.model)
    
    async def process(self, request: AgentRequest) -> AgentResponse:
        """使用简化模型处理请求，优先使用预定义响应"""
        try:
            # 首先检查是否有匹配的预定义响应
            query_lower = request.query.lower()
            for keyword, response in self.predefined_responses.items():
                if keyword in query_lower:
                    logger.info("Using predefined response", keyword=keyword)
                    return AgentResponse(
                        response=response,
                        confidence=0.9,
                        sources=["predefined"],
                        level_used=DegradationLevel.REDUCED.value
                    )
            
            # 如果没有预定义响应，使用简化模型
            messages = [
                {"role": "system", "content": "You are a helpful customer service assistant. Keep responses brief and focused."},
                {"role": "user", "content": request.query}
            ]
            
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.5,
                max_tokens=500
            )
            
            return AgentResponse(
                response=response.choices[0].message.content,
                confidence=0.7,
                sources=["openai_gpt35"],
                metadata={"model": self.model, "usage": response.usage.model_dump()},
                level_used=DegradationLevel.REDUCED.value
            )
            
        except Exception as e:
            logger.error("Error in ReducedFunctionalityAgent", error=str(e))
            raise
    
    def get_capabilities(self) -> list[str]:
        """获取该Agent支持的能力列表"""
        return [
            "standard_query_handling",
            "basic_context",
            "predefined_responses"
        ]
    
    def get_level(self) -> int:
        """获取该Agent对应的降级级别"""
        return DegradationLevel.REDUCED.value

最后，我们实现最小功能Agent：

# app/agent/degraded.py
import re
from typing import Dict, Any
from .base import BaseAgent, AgentRequest, AgentResponse
from ..degradation.levels import DegradationLevel
import structlog

logger = structlog.get_logger()


class MinimalFunctionalityAgent(BaseAgent):
    """最小功能Agent - 基于规则的系统"""
    
    def __init__(self):
        # 预定义FAQ
        self.faq_database = [
            {
                "keywords": ["shipping", "delivery", "ship", "deliver"],
                "response": "Standard shipping takes 3-5 business days. Express shipping takes 1-2 business days."
            },
            {
                "keywords": ["return", "refund", "exchange"],
                "response": "You can return items within 30 days of purchase. Please visit our returns page at example.com/returns for more information."
            },
            {
                "keywords": ["hours", "open", "service", "support"],
                "response": "Our customer service hours are Monday to Friday, 9 AM to 5 PM EST."
            },
            {
                "keywords": ["price", "cost", "discount", "sale"],
                "response": "Please visit our website for current pricing and promotions. Our sales team can also be reached at sales@example.com."
            },
            {
                "keywords": ["track", "tracking", "order", "status"],
                "response": "You can track your order by logging into your account and visiting the 'Orders' section."
            }
        ]
        
        # 默认响应
        self.default_response = "I'm sorry, I can only answer basic questions at this time. Please try again later or contact our customer service team at support@example.com."
        logger.info("MinimalFunctionalityAgent initialized")
    
    async def process(self, request: AgentRequest) -> AgentResponse:
        """基于关键词匹配处理请求"""
        query_lower = request.query.lower()
        
        # 尝试匹配FAQ
        for faq in self.faq_database:
            if any(keyword in query_lower for keyword in faq["keywords"]):
                logger.info("Matched FAQ", keywords=faq["keywords"])
                return AgentResponse(
                    response=faq["response"],
                    confidence=0.8,
                    sources=["faq_database"],
                    level_used=DegradationLevel.MINIMAL.value
                )
        
        # 如果没有匹配，返回默认响应
        logger.info("No FAQ match found, returning default response")
        return AgentResponse(
            response=self.default_response,
            confidence=0.5,
            sources=["default"],
            level_used=DegradationLevel.MINIMAL.value
        )
    
    def get_capabilities(self) -> list[str]:
        """获取该Agent支持的能力列表"""
        return [
            "faq_matching",
            "basic_information"
        ]
    
    def get_level(self) -> int:
        """获取该Agent对应的降级级别"""
        return DegradationLevel.MINIMAL.value

7.4 实现核心能力保底机制

现在，我们已经定义了不同级别的Agent，接下来需要实现一个机制来确保核心能力始终可用。我们将创建一个Agent工厂类，负责根据当前系统状态提供合适的Agent实例：

# app/agent/__init__.py
from .base import BaseAgent, AgentRequest, AgentResponse
from .core import FullFunctionalityAgent, ReducedFunctionalityAgent
from .degraded import MinimalFunctionalityAgent
from ..degradation.levels import DegradationLevel
from typing import Optional
import structlog

logger = structlog.get_logger()


class AgentFactory:
    """Agent工厂类 - 根据降级级别创建相应的Agent实例"""
    
    _agents = {}
    
    @classmethod
    def get_agent(cls, level: DegradationLevel) -> BaseAgent:
        """获取指定级别的Agent实例"""
        if level not in cls._agents:
            if level == DegradationLevel.FULL:
                cls._agents[level] = FullFunctionalityAgent()
            elif level == DegradationLevel.REDUCED:
                cls._agents[level] = ReducedFunctionalityAgent()
            elif level == DegradationLevel.MINIMAL:
                cls._agents[level] = MinimalFunctionalityAgent()
            else:
                raise ValueError(f"Unsupported degradation level: {level}")
        
        return cls._agents[level]
    
    @classmethod
    def get_minimal_agent(cls) -> BaseAgent:
        """获取最小功能Agent（确保核心能力保底）"""
        return cls.get_agent(DegradationLevel.MINIMAL)
    
    @classmethod
    def get_highest_available_agent(cls, max_level: DegradationLevel) -> BaseAgent:
        """获取最高可用级别的Agent，不超过指定级别"""
        # 尝试从最高级别开始，直到找到可用的Agent
        levels_to_try = [
            l for l in DegradationLevel 
            if l.value <= max_level.value and l != DegradationLevel.UNAVAILABLE
        ]
        levels_to_try.sort(key=lambda x: x.value)  # 从高级别到低级别
        
        for level in levels_to_try:
            try:
                agent = cls.get_agent(level)
                # 这里可以添加健康检查逻辑
                logger.info(f"Selected agent at level: {level.name}")
                return agent
            except Exception as e:
                logger.warning(f"Failed to initialize agent at level {level.name}", error=str(e))
                continue
        
        # 如果所有级别都失败，返回最小功能Agent（最后保底）
        logger.warning("All higher level agents failed, falling back to minimal agent")
        return cls.get_minimal_agent()

这个工厂类确保了即使所有高级别Agent都不可用，系统仍然可以提供最小功能Agent，从而保证核心能力的可用性。

---

8. 分步实现：优雅降级机制

在上一节中，我们实现了核心能力保底机制，确保了系统在任何情况下都能提供基本功能。现在，我们将实现优雅降级机制，使系统能够根据当前状态平滑地调整功能级别。

8.1 实现监控指标收集

首先，我们需要实现监控指标收集功能，以便系统能够了解当前的运行状态。我们将使用Prometheus客户端库来定义和收集指标：

# app/monitoring/metrics.py
from prometheus_client import (
    Counter, Gauge, Histogram, Summary,
    generate_latest, CONTENT_TYPE_LATEST
)
from fastapi import Response
import time
import structlog

logger = structlog.get_logger()


# 请求计数
REQUEST_COUNT = Counter(
    'ai_agent_requests_total',
    'Total number of requests processed',
    ['agent_level', 'endpoint', 'status']
)

# 响应时间直方图
REQUEST_DURATION = Histogram(
    'ai_agent_request_duration_seconds',
    'Time spent processing requests',
    ['agent_level', 'endpoint'],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

# 当前使用的Agent级别
CURRENT_AGENT_LEVEL = Gauge(
    'ai_agent_current_level',
    'Current agent level in use (0=full, 1=reduced, 2=minimal)',
    ['service']
)

# 系统健康度
SYSTEM_HEALTH = Gauge(
    'ai_agent_system_health',
    'Overall system health score (0-1)',
    ['service']
)

# API错误计数
API_ERRORS = Counter(
    'ai_agent_api_errors_total',
    'Total number of API errors',
    ['service', 'error_type']
)

# 降级事件计数
DEGRADATION_EVENTS = Counter(
    'ai_agent_degradation_events_total',
    'Total number of degradation events',
    ['from_level', 'to_level', 'reason']
)

# 模型置信度
MODEL_CONFIDENCE = Summary(
    'ai_agent_model_confidence',
    'Confidence score of model responses',
    ['agent_level']
)


def record_request(agent_level: int, endpoint: str, status: str, duration: float):
    """记录请求指标"""
    REQUEST_COUNT.labels(agent_level=agent_level, endpoint=endpoint, status=status).inc()
    REQUEST_DURATION.labels(agent_level=agent_level, endpoint=endpoint).observe(duration)


def set_agent_level(level: int, service: str = "ai-agent"):
    """设置当前Agent级别"""
    CURRENT_AGENT_LEVEL.labels(service=service).set(level)


def set_system_health(health: float, service: str = "ai-agent"):
    """设置系统健康度"""
    SYSTEM_HEALTH.labels(service=service).set(health)


def record_api_error(service: str, error_type: str):
    """记录API错误"""
    API_ERRORS.labels(service=service, error_type=error_type).inc()


def record_degradation_event(from_level: str, to_level: str, reason: str):
    """记录降级事件"""
    DEGRADATION_EVENTS.labels(from_level=from_level, to_level=to_level, reason=reason).inc()


def record_model_confidence(agent_level: int, confidence: float):
    """记录模型置信度"""
    MODEL_CONFIDENCE.labels(agent_level=agent_level).observe(confidence)


def get_metrics_response():
    """获取Prometheus指标响应"""
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )

8.2 实现降级管理器

现在，我们将实现降级管理器，它是优雅降级机制的核心组件，负责监控系统状态、评估健康度并做出降级决策：

# app/degradation/manager.py
import asyncio
import time
from typing import Optional, Dict, Any, Callable
from enum import Enum
from dataclasses import dataclass
import structlog

from .levels import DegradationLevel, LevelCapabilities, LEVEL_CAPABILITIES_MAP
from ..monitoring.metrics import (
    set_agent_level, set_system_health, record_degradation_event,
    record_api_error, get_metrics_response
)
from ..agent import AgentFactory, BaseAgent
from ..utils.state import StateManager

logger = structlog.get_logger()


class DegradationReason(Enum):
    """降级原因枚举"""
    HIGH_LATENCY = "high_latency"
    LOW_CONFIDENCE = "low_confidence"
    API_ERROR_RATE = "api_error_rate"
    HIGH_RESOURCE_USAGE = "high_resource_usage"
    MANUAL = "manual"
    RECOVERY = "recovery"


@dataclass
class HealthMetrics:
    """健康指标数据类"""
    timestamp: float
    avg_response_time: float
    error_rate: float
    avg_confidence: float
    cpu_usage: float
    memory_usage: float
    request_rate: float


class DegradationManager:
    """降级管理器"""
    
    def __init__(self, state_manager: StateManager):
        self.state_manager = state_manager
        self.current_level: DegradationLevel = DegradationLevel.FULL
        self.last_level_change_time: float = time.time()
        self.min_level_duration: float = 60.0  # 最小级别持续时间（秒）
        self.health_history: list[HealthMetrics] = []
        self.max_history_size: int = 100
        self.level_coefficients = {
            DegradationLevel.FULL: 1.0,
            DegradationLevel.REDUCED: 0.7,
            DegradationLevel.MINIMAL: 0.4
        }
        # 健康度评估权重
        self.metric_weights = {
            "response_time": 0.25,
            "error_rate": 0.3,
            "confidence": 0.25,
            "resource_usage": 0.2
        }
        # 阈值配置
        self.thresholds = {
            "response_time": {
                DegradationLevel.FULL: 2.0,
                DegradationLevel.REDUCED: 1.0,
                DegradationLevel.MINIMAL: 0.5
            },
            "error_rate": {
                DegradationLevel.FULL: 0.05,
                DegradationLevel.REDUCED: 0.02,
                DegradationLevel.MINIMAL: 0.01
            },
            "confidence": {
                DegradationLevel.FULL: 0.7,
                DegradationLevel.REDUCED: 0.6,
                DegradationLevel.MINIMAL: 0.5
            },
            "cpu_usage": 0.8,
            "memory_usage": 0.85
        }
        
        # 初始化状态
        self._initialize_state()
        logger.info("DegradationManager initialized", initial_level=self.current_level.name)
    
    def _initialize_state(self):
        """初始化状态"""
        # 尝试从状态管理器恢复当前级别
        saved_level = self.state_manager.get("current_degradation_level")
        if saved_level is not None:
            try:
                self.current_level = DegradationLevel(saved_level)
                logger.info("Restored degradation level from state", level=self.current_level.name)
            except ValueError:
                logger.warning("Invalid saved degradation level, using default", saved_level=saved_level)
        
        # 更新指标
        set_agent_level(self.current_level.value)
        self._update_system_health()
    
    def _calculate_system_health(self) -> float:
        """计算系统健康度"""
        if not self.health_history:
            return 1.0  # 如果没有历史数据，假设完全健康
        
        # 使用最近的指标
        latest_metrics = self.health_history[-1]
        
        # 计算各指标的隶属度
        # 响应时间：越短越好
        rt_threshold = self.thresholds["response_time"][DegradationLevel.FULL]
        rt_membership = max(0, 1 - (latest_metrics.avg_response_time / rt_threshold))
        
        # 错误率：越低越好
        er_membership = max(0, 1 - (latest_metrics.error_rate / self.thresholds["error_rate"][DegradationLevel.FULL]))
        
        # 置信度：越高越好