设想一个线上故障诊断服务工程师提交一段脱敏日志系统结合规则和大模型给出诊断方向、证据和下一步建议。刚上线时最容易写出的代码是业务服务读取配置调用某家模型 SDK解析回答然后返回结果。这段代码通常能跑但它会在生产环境里很快变得脆弱供应商限流、网络超时、密钥轮换、模型格式漂移、成本上涨、A/B 实验和离线评测都会穿透到本该只关心诊断策略的业务层。本文不讨论某个项目的实现而是提炼一套可迁移的生产工程模式让业务规则定义需要什么能力让外部适配器决定如何调用模型。TL;DR10 秒核心要点业务服务只决定“何时调用模型、失败后怎么办”不处理 URL、Key、HTTP 或供应商 JSON。用小型Protocol描述业务需要的能力真实模型客户端、测试替身和未来供应商适配器都可以满足它。模型返回的是候选不是事实涉及诊断、风控、运维等场景时必须经过本地验证和安全回退。生产中真正要监控的是质量、回退、延迟和成本不只是“模型接口是否返回 200”。一个典型生产事故某次模型供应商返回429 Too Many Requests。如果业务代码直接捕获 SDK 异常常见的后果有两种要么把原始异常直接展示给用户要么在多个业务函数里各自加一段重试。前者泄露内部细节后者会造成重试风暴、行为不一致和难以测试的分支。真正需要的业务策略其实很朴素规则命中时直接返回规则不确定时才请求模型模型暂时不可用时返回保守结论并记录可观测事件模型给出候选后仍要由本地代码验证证据。要把这件事做稳关键不是选哪个模型而是让变化停留在合适的边界。核心架构隔离大模型的五个关键概念概念它解决什么问题生产中的典型位置业务规则定义确认、拒绝和降级的产品策略领域服务、规则引擎、校验器契约接口用稳定业务语言描述外部能力ModelClient、Retriever、PaymentGateway适配器翻译供应商 SDK、HTTP、鉴权和响应格式OpenAICompatibleModelClient等依赖注入不让业务对象自己创建外部依赖构造函数、函数参数组合根集中创建具体对象并完成组装Web 应用启动代码、Worker 启动代码、CLI 入口依赖方向应该是组合根 - 读取运行配置创建具体适配器 - 把适配器注入业务服务 ​ 业务服务 - 只依赖领域规则、结果类型、能力契约 ​ 具体适配器 - 依赖 HTTP/SDK、运行配置、能力契约业务服务可以调用契约但不应 import 某家供应商客户端。适配器可以依赖契约因为它要承诺实现这个能力。这就是“依赖向内”的实际含义。图中的关键约束是业务服务依赖契约适配器依赖契约并连接基础设施业务服务不直接依赖外部模型服务。什么时候应该隔离外部依赖满足以下任意两项就值得隔离外部能力可能更换例如云模型、本地模型、不同数据库或支付渠道。它会超时、限流、失败或返回不稳定格式。它需要密钥、计费、权限控制或合规审计。业务策略必须在无网络环境下稳定测试。同一种能力会被 API、异步任务、批处理等多个入口使用。反过来一次性脚本、单一依赖、没有测试和替换需求的场景直接调用 API 完全合理。抽象有成本多一个契约、多一些对象和更多阅读负担。最佳实践不是“永远抽象”而是让抽象成本小于未来变化成本。Step 1用 Protocol 定义业务契约杜绝供应商泄漏契约应该由消费者定义因为消费者最清楚自己需要什么。它要小、稳定、面向业务语言不能把messages、temperature、chat/completions等供应商概念泄漏到业务服务。Python 中优先使用Protocol表达这种能力from dataclasses import dataclass from typing import Protocol ​ ​ dataclass(frozenTrue) 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让服务只消费契约统一策略与降级服务层负责策略而不是组装 SDKclass 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(sourcemodel_fallback) ​ return validate_candidate(error_log, candidate) or DiagnosisResult.undetermined( sourcevalidation_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 errorraise ... 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 只能保证“容易解析”不能保证结论真实。线上系统必须把模型输出视为候选并在本地验证字段是否完整、类型是否正确引用的证据是否逐字存在于输入中证据是否满足业务语义而不只是碰巧包含相同关键词校验失败或信息不足时是否安全地返回“不确定”。这个边界尤其适用于日志诊断、风控审核、合规问答和运维建议。模型可以扩大覆盖面但最终信任必须由可测试的本地规则决定。验证也要按风险分级字段类型、证据是否存在等轻量检查应同步完成只有不影响用户安全结论的重量级分析才考虑异步执行、采样或灰度。凡是验证结果决定“能否给用户执行建议”就不能为了省一点延迟而绕过主流程。Step 6用测试替身验证策略不让每次测试都调用模型有了契约单元测试不必请求真实模型class StubModelClient: # Stub提供可控输出用于验证服务策略。 # 若额外记录调用次数它同时承担轻量 Spy 的职责。 def diagnose(self, error_log: str) - DiagnosisCandidate: return DiagnosisCandidate( statusconfirmed, suspected_causemissing_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 个致命反模式契约泄漏供应商术语generate(messages, temperature, tools)只是把某家 API 原样搬进业务层。更好的契约是diagnose(error_log)或retrieve(query)。万能客户端一个既聊天、又嵌入、又重排、又管理工具调用的LLMClient往往会膨胀。按业务能力拆小契约更稳定。供应商异常直达用户原始异常、SDK 错误码或响应文本不应成为用户体验的一部分。全局单例偷偷创建依赖它让测试替换、配置加载时机和生命周期管理变得不透明。过早引入框架和容器依赖注入首先是一种组装方式一个构造函数和应用启动代码往往已经足够。上线前检查清单10 个问题快速自查业务真正需要的最小能力是什么能否用小契约描述业务服务是否 import 了供应商 SDK、HTTP 库或配置文件能否移到适配器外部异常是否被转换为稳定错误并由服务统一决定重试、降级或提示模型候选是否经过本地事实和语义校验能否用替身在无网络、无密钥的情况下测试主要策略具体实现是否集中在组合根而不是散落在业务函数中是否记录了质量、延迟、成本、错误码和回退率同时保护敏感输入是否设置超时、预算、并发上限和有界重试这层抽象是否处于高频路径是否有重复请求等可测量成本如果担心性能是否已有指标或剖析证据当前规模真的需要进一步抽象吗结语依赖注入不是为了把小服务写得像大系统而是为了让业务规则不被外部工具绑住。AI 系统尤其需要这层边界模型会更换、接口会失败、成本需要控制而面向用户的结论必须稳定、可解释、可验证。先让内层定义需要什么能力再让外层选择怎样实现它。把模型放在边界把最终判断留在可测试的业务层系统才有持续演进的余地。