设想一个线上故障诊断服务：工程师提交一段脱敏日志，系统结合规则和大模型给出诊断方向、证据和下一步建议。刚上线时，最容易写出的代码是：业务服务读取配置，调用某家模型 SDK，解析回答，然后返回结果。

这段代码通常能跑，但它会在生产环境里很快变得脆弱：供应商限流、网络超时、密钥轮换、模型格式漂移、成本上涨、A/B 实验和离线评测，都会穿透到本该只关心诊断策略的业务层。

本文不讨论某个项目的实现，而是提炼一套可迁移的生产工程模式：让业务规则定义需要什么能力，让外部适配器决定如何调用模型。

TL;DR：10 秒核心要点

• 业务服务只决定“何时调用模型、失败后怎么办”，不处理 URL、Key、HTTP 或供应商 JSON。

• 用小型 Protocol 描述业务需要的能力；真实模型客户端、测试替身和未来供应商适配器都可以满足它。

• 模型返回的是候选，不是事实；涉及诊断、风控、运维等场景时，必须经过本地验证和安全回退。

• 生产中真正要监控的是质量、回退、延迟和成本，不只是“模型接口是否返回 200”。

一个典型生产事故

某次模型供应商返回 429 Too Many Requests。如果业务代码直接捕获 SDK 异常，常见的后果有两种：要么把原始异常直接展示给用户，要么在多个业务函数里各自加一段重试。前者泄露内部细节，后者会造成重试风暴、行为不一致和难以测试的分支。

真正需要的业务策略其实很朴素：规则命中时直接返回；规则不确定时才请求模型；模型暂时不可用时返回保守结论并记录可观测事件；模型给出候选后，仍要由本地代码验证证据。

要把这件事做稳，关键不是选哪个模型，而是让变化停留在合适的边界。

核心架构：隔离大模型的五个关键概念

概念	它解决什么问题	生产中的典型位置
业务规则	定义确认、拒绝和降级的产品策略	领域服务、规则引擎、校验器
契约（接口）	用稳定业务语言描述外部能力	ModelClient、Retriever、PaymentGateway
适配器	翻译供应商 SDK、HTTP、鉴权和响应格式	OpenAICompatibleModelClient 等
依赖注入	不让业务对象自己创建外部依赖	构造函数、函数参数
组合根	集中创建具体对象并完成组装	Web 应用启动代码、Worker 启动代码、CLI 入口

依赖方向应该是：

组合根
  -> 读取运行配置，创建具体适配器
  -> 把适配器注入业务服务
​
业务服务
  -> 只依赖领域规则、结果类型、能力契约
​
具体适配器
  -> 依赖 HTTP/SDK、运行配置、能力契约

业务服务可以调用契约，但不应 import 某家供应商客户端。适配器可以依赖契约，因为它要承诺实现这个能力。这就是“依赖向内”的实际含义。

图中的关键约束是：业务服务依赖契约，适配器依赖契约并连接基础设施；业务服务不直接依赖外部模型服务。

什么时候应该隔离外部依赖

满足以下任意两项，就值得隔离：

1. 外部能力可能更换，例如云模型、本地模型、不同数据库或支付渠道。

2. 它会超时、限流、失败或返回不稳定格式。

3. 它需要密钥、计费、权限控制或合规审计。

4. 业务策略必须在无网络环境下稳定测试。

5. 同一种能力会被 API、异步任务、批处理等多个入口使用。

反过来，一次性脚本、单一依赖、没有测试和替换需求的场景，直接调用 API 完全合理。抽象有成本：多一个契约、多一些对象和更多阅读负担。最佳实践不是“永远抽象”，而是让抽象成本小于未来变化成本。

Step 1：用 Protocol 定义业务契约，杜绝供应商泄漏

契约应该由消费者定义，因为消费者最清楚自己需要什么。它要小、稳定、面向业务语言，不能把 messages、temperature、chat/completions 等供应商概念泄漏到业务服务。

Python 中，优先使用 Protocol 表达这种能力：

from dataclasses import dataclass
from typing import Protocol
​
​
@dataclass(frozen=True)
class DiagnosisCandidate:
    status: str
    suspected_cause: str | None
    evidence_fragments: tuple[str, ...]
​
​
class ModelClientError(RuntimeError):
    """外部模型失败的统一业务错误。"""
​
​
class ModelClient(Protocol):
    """业务只要求：根据输入返回未信任的诊断候选。"""
​
    def diagnose(self, error_log: str) -> DiagnosisCandidate:
        ...

Protocol 是结构化类型：一个类不必显式继承它，只要实现了兼容的 diagnose 方法，就可被类型检查器视为符合契约。这很适合第三方 SDK 包装器和测试替身。

Protocol 还是 ABC？

默认选 Protocol。它只描述能力，不强制继承，耦合更低。

当多个实现确实共享行为、生命周期或状态，或者你需要在运行时强制继承关系时，再选 ABC（抽象基类）。不要因为“接口就该有父类”而引入 ABC。Protocol 的约束主要发生在静态检查阶段，关键业务边界仍必须有运行时校验。

两者在适配第三方客户端时的差别，可以用这段极小代码看清：

from abc import ABC, abstractmethod
​
​
class ABCModelClient(ABC):
    @abstractmethod
    def diagnose(self, error_log: str) -> DiagnosisCandidate: ...
​
​
class VendorWrapper(ABCModelClient):  # ABC：必须显式继承
    def diagnose(self, error_log: str) -> DiagnosisCandidate: ...
​
​
class ThirdPartyWrapper:  # Protocol：不必继承，只要方法签名兼容即可
    def diagnose(self, error_log: str) -> DiagnosisCandidate: ...

这里的取舍不是“Protocol 更高级”。当你无法或不愿让第三方包装器继承你的父类时，结构化类型更轻；当多个实现需要共享重试、连接管理或生命周期代码时，ABC 才可能更合适。

流式响应不要塞进同步契约

生产中聊天、长报告等场景常需要 Streaming。不要让一个 diagnose() 同时承担“返回完整结果”和“逐块输出”两种语义；为流式能力单独定义契约更清楚：

from collections.abc import AsyncIterator
from typing import Protocol
​
​
class StreamingModelClient(Protocol):
    def stream_answer(self, prompt: str) -> AsyncIterator[str]:
        """业务层只消费文本块，不关心 SSE、HTTP 或 SDK。"""
        ...

业务层消费 AsyncIterator，适配器负责把 SSE 或供应商流式协议翻译成文本块。诊断这类必须先完成证据校验才能给结论的工作流，通常不应把未经验证的流式结论直接推给用户。

Step 2：让服务只消费契约，统一策略与降级

服务层负责策略，而不是组装 SDK：

class DiagnosisService:
    def __init__(self, model_client: ModelClient | None = None) -> None:
        self._model_client = model_client
​
    def diagnose(self, error_log: str) -> DiagnosisResult:
        rule_result = run_deterministic_rules(error_log)
        if rule_result.is_confirmed or self._model_client is None:
            return rule_result
​
        try:
            candidate = self._model_client.diagnose(error_log)
        except ModelClientError:
            return DiagnosisResult.undetermined(source="model_fallback")
​
        return validate_candidate(error_log, candidate) or DiagnosisResult.undetermined(
            source="validation_fallback"
        )

这个服务决定规则优先、何时调用模型、模型失败时怎么降级。它不关心 Base URL、认证头或具体 JSON 结构。任何 API、Worker 或命令行入口都可以复用这套策略。

Step 3：用适配器隔离供应商、异常与协议细节

适配器是唯一可以了解 HTTP、SDK、鉴权和供应商响应格式的地方。它负责双向翻译：把业务输入转换为供应商请求，再把供应商响应转换为 DiagnosisCandidate。

异常转换是隔离层的核心价值。不要让 HTTPError、RateLimitError 或供应商 SDK 异常穿透到服务层：

try:
    # 1. 基础设施层负责发送请求和解析供应商协议。
    raw_response = send_request(request, timeout_seconds)
    response = json.loads(raw_response.decode("utf-8"))
    content = response["choices"][0]["message"]["content"]
    candidate_data = json.loads(content)
except HTTPError as error:
    # 2. 上层只接触稳定的领域错误，不解析 HTTP 或 SDK 异常。
    raise ModelClientError(f"模型服务返回 HTTP {error.code}。") from error
except (URLError, OSError, TimeoutError) as error:
    raise ModelClientError("模型服务网络请求失败。") from error
except (UnicodeDecodeError, json.JSONDecodeError, KeyError, IndexError, TypeError) as error:
    raise ModelClientError("模型服务响应格式无效。") from error

raise ... from error 保留底层原因，方便受控日志和排障；向上层暴露的却是稳定错误，不泄露密钥、内部 URL 或供应商响应细节。这类隔离层常被称为防腐层：外部系统的怪异命名和易变格式不应污染领域语言。

进阶：让错误分类驱动策略

如果所有模型失败都统一降级，简单的 ModelClientError 已足够。只有当策略真的不同，才增加结构化错误码：

from enum import Enum
​
​
class ModelClientErrorCode(str, Enum):
    NETWORK = "network"
    AUTH = "auth"
    FORMAT = "format"
    RATE_LIMIT = "rate_limit"
​
​
class ModelClientError(RuntimeError):
    def __init__(self, message: str, code: ModelClientErrorCode) -> None:
        super().__init__(message)
        self.code = code

例如，RATE_LIMIT 可以触发有限次数、带抖动的退避重试；FORMAT 应直接回退并告警；AUTH 通常停止重试并提示配置问题。底层异常不必另存为自定义 cause 字段，仍使用 raise ModelClientError(...) from error 保留 Python 标准异常链。

错误码不是“自动重试”的许可证。重试必须有总时限、最大次数、退避与抖动，并确认操作可安全重试。否则高峰期的重试会放大供应商故障。

Step 4：在组合根组装依赖，而不是在业务函数里 new 客户端

组合根只负责创建和连接对象：加载运行配置，选择模型适配器，传入服务，启动入口。

config = load_runtime_config()
# 只在最外层决定使用哪家模型或哪种适配器。
client = OpenAICompatibleModelClient(config)
service = DiagnosisService(client)
app = create_http_app(service)

小型服务把这段代码放在应用启动处即可。不要为了“企业级”过早引入依赖注入容器或复杂工厂。只有模型种类、运行环境和配置组合明显增多时，再抽出一个小型客户端工厂或组装模块。

Step 5：模型输出是候选，不是事实

结构化 JSON 只能保证“容易解析”，不能保证结论真实。线上系统必须把模型输出视为候选，并在本地验证：

1. 字段是否完整、类型是否正确；

2. 引用的证据是否逐字存在于输入中；

3. 证据是否满足业务语义，而不只是碰巧包含相同关键词；

4. 校验失败或信息不足时，是否安全地返回“不确定”。

这个边界尤其适用于日志诊断、风控审核、合规问答和运维建议。模型可以扩大覆盖面，但最终信任必须由可测试的本地规则决定。

验证也要按风险分级：字段类型、证据是否存在等轻量检查应同步完成；只有不影响用户安全结论的重量级分析，才考虑异步执行、采样或灰度。凡是验证结果决定“能否给用户执行建议”，就不能为了省一点延迟而绕过主流程。

Step 6：用测试替身验证策略，不让每次测试都调用模型

有了契约，单元测试不必请求真实模型：

class StubModelClient:
    # Stub：提供可控输出，用于验证服务策略。
    # 若额外记录调用次数，它同时承担轻量 Spy 的职责。
    def diagnose(self, error_log: str) -> DiagnosisCandidate:
        return DiagnosisCandidate(
            status="confirmed",
            suspected_cause="missing_environment_variable",
            evidence_fragments=("PAYMENTS_API_TOKEN is not set",),
        )
​
​
service = DiagnosisService(StubModelClient())

这让核心策略测试摆脱网络、密钥、供应商可用性和潜在费用。应重点验证：规则命中时是否跳过模型、超时后是否安全降级、编造证据是否被拒绝、错误分类是否触发正确策略。

替身的术语不必教条化：Stub 提供受控输出，Spy 记录实际交互，Mock 预先声明并验证交互预期。一个测试对象可能兼具多种职责，关键是测试意图清楚。

真实 API 调用仍要保留，但放在少量、显式、脱敏的端到端检查与离线评测中。不要让每次单元测试都触网。

生产补强：可观测性、Trace ID 与成本控制

可观测性：没有证据，就无法运营

每次模型调用至少应记录：请求 ID、模型与提示词版本、超时与错误分类、是否回退、校验是否通过、延迟、输入输出 token 或计费单位。日志内容和模型原始输出可能包含敏感信息，应脱敏、采样、设置访问控制和保留期限。

指标至少覆盖成功率、错误码分布、回退率、校验拒绝率、端到端延迟、成本和人工确认后的正确率。只监控“模型调用成功率”会掩盖模型成功返回但被本地校验拒绝的质量问题。

Trace ID 应从入口层贯穿到模型适配器：可通过 ContextVar 或 OpenTelemetry 的上下文传播，把同一个请求标识写入应用日志、模型调用 Span 和降级事件。不要把原始日志、密钥或用户标识塞进 Trace 属性；Trace 的价值是把一次请求的事件串起来，不是复制敏感数据。

成本与可靠性：先测量，再优化

对大多数 LLM 调用，主要延迟和成本来自网络与模型推理，不是多创建了一个 Python 对象。真正要防的是重复序列化、重复请求、无界重试和把每个用户动作都送进大模型。

先设置超时、预算、并发上限和回退策略；再用指标和剖析证据决定是否缓存、批处理、缩短上下文或切换更小模型。没有测量就做微优化，通常只会增加复杂度。

为第二个模型和 A/B 实验留出空间

增加第二个模型时，只需提供另一个符合 ModelClient 的适配器，并在组合根选择或路由它。规则、服务策略和校验器不应修改。

真正的 A/B 实验不只是加一个 if：还要定义稳定分流、记录模型和提示词版本、关联用户反馈、比较质量、延迟和成本，并避免让实验结果污染保留集。隔离不会自动完成实验，但它把实验变化限制在合适边界。

避坑指南：AI 应用架构的 5 个致命反模式

1. 契约泄漏供应商术语：generate(messages, temperature, tools) 只是把某家 API 原样搬进业务层。更好的契约是 diagnose(error_log) 或 retrieve(query)。

2. 万能客户端：一个既聊天、又嵌入、又重排、又管理工具调用的 LLMClient 往往会膨胀。按业务能力拆小契约更稳定。

3. 供应商异常直达用户：原始异常、SDK 错误码或响应文本不应成为用户体验的一部分。

4. 全局单例偷偷创建依赖：它让测试替换、配置加载时机和生命周期管理变得不透明。

5. 过早引入框架和容器：依赖注入首先是一种组装方式，一个构造函数和应用启动代码往往已经足够。

上线前检查清单：10 个问题快速自查

1. 业务真正需要的最小能力是什么？能否用小契约描述？

2. 业务服务是否 import 了供应商 SDK、HTTP 库或配置文件？能否移到适配器？

3. 外部异常是否被转换为稳定错误，并由服务统一决定重试、降级或提示？

4. 模型候选是否经过本地事实和语义校验？

5. 能否用替身在无网络、无密钥的情况下测试主要策略？

6. 具体实现是否集中在组合根，而不是散落在业务函数中？

7. 是否记录了质量、延迟、成本、错误码和回退率，同时保护敏感输入？

8. 是否设置超时、预算、并发上限和有界重试？

9. 这层抽象是否处于高频路径，是否有重复请求等可测量成本？如果担心性能，是否已有指标或剖析证据？

10. 当前规模真的需要进一步抽象吗？

结语

依赖注入不是为了把小服务写得像大系统，而是为了让业务规则不被外部工具绑住。AI 系统尤其需要这层边界：模型会更换、接口会失败、成本需要控制，而面向用户的结论必须稳定、可解释、可验证。

先让内层定义需要什么能力，再让外层选择怎样实现它。把模型放在边界，把最终判断留在可测试的业务层，系统才有持续演进的余地。