7.1 模型抽象层设计

模型抽象层连接智能体应用与具体 LLM API，支持灵活的多模型或单模型策略。本节介绍配置管理系统、故障转移链路、Provider 接口设计和模型选择引擎。

7.1.1 核心概念

模型抽象层是连接智能体应用与具体 LLM API 的中间层。其目标是：

1. 统一接口：提供一致的 API，支持 Claude、GPT、DeepSeek、Gemini 等多种模型

2. 灵活切换：动态选择或回退模型，无需修改业务逻辑

3. 配置驱动：通过配置文件管理模型选择、默认值、故障转移策略

4. 错误隔离：单个模型的故障不影响整个系统

7.1.2 设计权衡：多模型 vs 单模型绑定

Claude Code 方案（单模型绑定）

• 绑定 Claude 模型家族(Opus/Sonnet/Haiku)

• 专注于版本管理与推理预算调优

• 深度集成 Adaptive Thinking 等 Claude 特性

• 场景：需要最佳性能、深度功能集成、模型特性充分利用

OpenClaw 方案（多模型支持）

• 支持多供应商切换(Claude/GPT/DeepSeek/Gemini/Ollama)

• 供应商认证与接入配置(openclaw.json)

• 模型故障转移链路(fallback mechanism)

• 场景：需要成本优化、供应商多元化、灰度迁移

选择建议：

• 小团队初期选择单模型绑定降低复杂度

• 成熟产品需要多模型以应对供应商依赖和成本波动

7.1.3 配置管理系统

模型选择策略

模型选择的实现方式如下：

from enum import Enum
from dataclasses import dataclass
from typing import Optional, List

class ModelProviderType(Enum):
    CLAUDE = "claude"
    OPENAI = "openai"
    DEEPSEEK = "deepseek"
    GEMINI = "gemini"
    OLLAMA = "ollama"

@dataclass
class ModelConfig:
    """模型配置对象"""
    provider: ModelProviderType
    model_id: str
    api_key: Optional[str] = None
    api_endpoint: Optional[str] = None
    timeout: int = 30
    max_tokens: int = 4096
    temperature: Optional[float] = None  # 默认不传；部分 Claude 模型拒绝非默认采样参数

@dataclass
class ModelSelectionPolicy:
    """模型选择策略"""
    primary: ModelConfig
    fallback_chain: List[ModelConfig] = None
    cost_threshold: float = None  # 成本上限,超过则切换
    latency_threshold: int = None  # 延迟上限,超过则切换

故障转移链路

故障转移链路定义了模型失败时的替代方案：

实现示例：

import time

class CircuitBreaker:
    """熔断器,跟踪模型健康状态"""
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "closed"  # closed, open, half-open
        self.last_failure_time = None

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            self.last_failure_time = time.time()

    def is_available(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open 允许尝试

Provider 接口设计

Provider 接口定义了模型的核心操作。使用 Python Protocol 提供灵活的鸭子类型：

from typing import Protocol, List, Any, Optional

class Message:
    def __init__(self, role: str, content: str):
        self.role = role  # "user", "assistant"
        self.content = content

class ProviderResponse:
    def __init__(self, content: str, tokens_used: int, model: str):
        self.content = content
        self.tokens_used = tokens_used
        self.model = model

class ModelProvider(Protocol):
    """LLM 供应商的统一接口"""

    def complete(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ) -> ProviderResponse:
        """完整的模型调用(无流式)"""
        ...

    def stream(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ):
        """流式模型调用,逐块返回"""
        ...

    def estimate_tokens(self, text: str) -> int:
        """估算文本的 token 数"""
        ...

    def validate_config(self) -> bool:
        """验证配置有效性(API 密钥等)"""
        ...

具体实现：Claude Provider

Claude Provider 的具体实现如下：

import anthropic
from typing import Generator, Optional

class ClaudeProvider:
    def __init__(self, config: ModelConfig):
        self.config = config
        self.client = anthropic.Anthropic(api_key=config.api_key)

    def complete(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ) -> ProviderResponse:
        """Claude 完整调用"""
        api_messages = [
            {"role": msg.role, "content": msg.content}
            for msg in messages
        ]
        kwargs = {
            "model": self.config.model_id,
            "max_tokens": max_tokens,
            "messages": api_messages,
        }
        if temperature is not None:
            kwargs["temperature"] = temperature
        response = self.client.messages.create(**kwargs)
        return ProviderResponse(
            content=response.content[0].text,
            tokens_used=(
                response.usage.input_tokens
                + getattr(response.usage, "cache_creation_input_tokens", 0)
                + getattr(response.usage, "cache_read_input_tokens", 0)
                + response.usage.output_tokens
            ),
            model=self.config.model_id,
        )

    def stream(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ) -> Generator[str, None, None]:
        """Claude 流式调用"""
        api_messages = [
            {"role": msg.role, "content": msg.content}
            for msg in messages
        ]
        kwargs = {
            "model": self.config.model_id,
            "max_tokens": max_tokens,
            "messages": api_messages,
        }
        if temperature is not None:
            kwargs["temperature"] = temperature
        with self.client.messages.stream(**kwargs) as stream:
            for text in stream.text_stream:
                yield text

    def estimate_tokens(self, text: str) -> int:
        """Claude token 计数"""
        response = self.client.messages.count_tokens(
            model=self.config.model_id,
            messages=[{"role": "user", "content": text}]
        )
        return response.input_tokens

    def validate_config(self) -> bool:
        try:
            self.estimate_tokens("test")
            return True
        except:
            return False

模型选择引擎

模型选择引擎的实现方式如下：

class ModelSelectionEngine:
    def __init__(self, policy: ModelSelectionPolicy):
        self.policy = policy
        self.breakers = {}
        self._init_breakers()

    def _init_breakers(self):
        for config in [self.policy.primary] + (
            self.policy.fallback_chain or []
        ):
            self.breakers[config.model_id] = CircuitBreaker()

    def select_model(self) -> ModelProvider:
        """根据健康状态选择可用模型"""
        candidates = [self.policy.primary] + (
            self.policy.fallback_chain or []
        )
        for config in candidates:
            breaker = self.breakers[config.model_id]
            if breaker.is_available():
                return self._create_provider(config)
        raise Exception("所有模型不可用")

    def mark_failure(self, model_id: str):
        """记录模型故障"""
        if model_id in self.breakers:
            self.breakers[model_id].record_failure()

    def mark_success(self, model_id: str):
        """记录模型成功"""
        if model_id in self.breakers:
            self.breakers[model_id].record_success()

    def _create_provider(self, config: ModelConfig):
        if config.provider == ModelProviderType.CLAUDE:
            return ClaudeProvider(config)
        # ... 其他 provider 实现

配置文件管理

配置文件的结构示例如下：

{
  "model_selection": {
    "primary": {
      "provider": "claude",
      "model_id": "claude-sonnet-4-6",
      "timeout": 30,
      "max_tokens": 4096
    },
    "fallback_chain": [
      {
        "provider": "openai",
        "model_id": "gpt-5.4",
        "timeout": 30,
        "max_tokens": 4096
      },
      {
        "provider": "deepseek",
        "model_id": "deepseek-chat",
        "timeout": 30,
        "max_tokens": 4096
      }
    ],
    "cost_threshold": 0.05,
    "latency_threshold": 5000
  }
}

模型抽象层架构图

模型抽象层的整体架构如下所示：

图 7-1：模型抽象层架构 —— 应用通过统一的路由器访问多个模型提供商

总结

模型抽象层通过 Provider 接口、配置驱动的选择策略和故障转移机制，实现了：

• 灵活的多模型支持或单模型绑定

• 透明的故障切换，对上层应用无感

• 可观测的模型健康度和选择决策

• 成本与性能的可控权衡

这为后续的输出解析、质量门控奠定了基础。

7.1.4 能力降级与权限收缩

前面的故障转移链路解决的是可用性问题：主模型不可用时换一个能用的，让请求继续得到响应。但 select_model() 的判据只有 CircuitBreaker.is_available()，也就是「这个模型现在能不能调通」。它没有回答另一个问题：换上来的模型，是否仍然有能力承担原来那些自动执行的决策。

这两件事被混为一谈时，会出现一种很难察觉的生产状态：

维度	状态
服务可用性	正常，仪表板全绿
模型能力	已下降（上下文更小、结构化输出更弱、工具调用不稳定）
智能体权限	未变，仍可自动执行

主模型故障是一次显性事件，会触发告警、进入值班流程；而回退成功之后，告警恢复、可用性指标回到 99.99%，能力下降却不会触发任何东西。系统看起来已经恢复了。

问题出在能力下降会静默改变决策的输入。假设回退模型的上下文预算更小，抽象层为了让工作流继续跑通，只能截掉一部分证据；智能体于是在一份不完整的材料上做判断。如果被截掉的恰好是一条限制性条款，而剩下的材料看起来都很正常，模型会给出一个自信、无幻觉、推理链条完整的结论——然后被自动放行。这里没有任何一个组件报错：模型没编造，回退机制按设计工作，熔断器状态正常。真正的缺陷是系统把能力降级当成了基础设施细节，而不是一次权限变更。

因此，回退策略除了定义「主模型失败后换谁」，还应该定义「能力变化时权限如何跟着变」：

from dataclasses import dataclass, field

@dataclass
class CapabilityProfile:
    """描述一个模型能稳定承担什么，而不只是能否调通"""
    context_tokens: int
    structured_output: str        # high | moderate | limited
    tool_calling: str             # high | moderate | limited
    evidence_reconciliation: str  # high | moderate | limited

@dataclass
class AuthorityProfile:
    """该能力档位下允许的最高权限"""
    may_recommend: bool = True
    may_auto_approve: bool = False
    may_execute_external: bool = False

# 与 ModelConfig 一一对应：能力档位决定权限上限
CAPABILITY_AUTHORITY = {
    "primary": AuthorityProfile(
        may_recommend=True, may_auto_approve=True, may_execute_external=True
    ),
    "fallback": AuthorityProfile(
        may_recommend=True, may_auto_approve=False, may_execute_external=False
    ),
}

对应地，模型选择引擎返回的就不再只是一个 provider，而是「provider + 当前可用权限」这一对值：

class AuthorityAwareSelectionEngine(ModelSelectionEngine):
    def select_with_authority(self):
        """返回 (provider, authority, degraded)，让上层无法忽略降级事实"""
        primary = self.policy.primary
        if self.breakers[primary.model_id].is_available():
            return self._create_provider(primary), CAPABILITY_AUTHORITY["primary"], False

        for config in self.policy.fallback_chain or []:
            if self.breakers[config.model_id].is_available():
                # 关键：换模型的同时换权限，二者不能分开决策
                return self._create_provider(config), CAPABILITY_AUTHORITY["fallback"], True

        raise RuntimeError("所有模型不可用")

这样一来，回退期间智能体仍然可以读取、分析、生成建议，只是不再拥有自动批准和对外执行的权限——服务没有中断，风险敞口却收窄了。

这里的核心判断是：可用性回退不应该静默地变成权限回退。为无状态服务做冗余时，副本和主库提供的是等价能力，切换是透明的；而不同模型之间除了性能差异，能安全承担的决策范围也不一样。所以回退链路本身属于部署策略的一部分，不只是一项基础设施配置。

由此还可以推出一个更基本的取舍：不是所有能力都应该「优雅降级」。

类别	典型任务	回退期间的处理
可降级	提取、摘要、草稿、低风险建议	继续执行
不可降级	高价值执行、不可逆操作、受监管决策	暂停自动路径，转人工

对不可降级的任务，正确的行为不是「保持服务可用」，而是明确返回「自动决策当前不可用」并转入人工审核。这与 11.3 容错模式 中「失败要显式、不要静默吞掉」是同一条原则在权限维度上的体现。

最后，降级期间的运行事实需要被记录下来，否则事后无法回答「这段时间到底发生了什么」：哪些请求是在降级状态下处理的、其中有没有本该走人工的决策被自动放行、恢复后是否需要回溯复核。可用性仪表板上的 99.99% 并不能替代这份记录，因为它衡量的是「系统有没有响应」，而不是「系统响应时是否具备原有的保障」。恢复主模型时也同样需要一道显式的门槛——健康检查通过、抽样回归通过、降级期间积压的高风险决策已经处理完毕，之后才把权限调回原位。