多模型时代，如果还在用 if-else 硬编码模型调用，那这套系统基本活不过三个月。我见过太多团队把“换个模型”这件事做成了一场灾难：模型一换，所有调用点跟着改，提示词散落各处，参数格式不统一，流式输出直接崩掉。这个系列写到第七篇，我觉得是时候把目光从单个 Prompt 技巧转向系统架构层面的设计模式了——生成式AI应用一旦进入生产环境，真正的复杂度不在于模型本身，而在于你如何用可维护、可扩展的方式去编排它。
这篇我重点聊多模型路由与工程化集成，用策略模式、工厂模式和适配器模式搭一个支持热切换的路由框架，顺带讲讲链路编排里的模板方法和责任链。适合正在把 LLM 应用从原型推向生产，天天被模型切换、厂商差异、链路不稳定折磨的工程师。内容偏实战，代码都是可落地的骨架，不是理论空谈。
1. 多模型时代，为什么设计模式从“加分项”变成了“生存技能”
先讲一个我真实踩过的坑。去年做一个客服摘要系统，前期只接了某一家大模型，代码写得很“质朴”：每个业务方法里直接 new 一个 client，然后调 chat 接口，参数散落在各个 Service 里。上线后模型效果不好，想换成另一家，结果一改就是三天——不是模型本身难接，而是调用方式、参数名、返回结构全都不一样，每个调用点都得单独适配。改完还不敢保证有没有漏网之鱼，整个团队焦头烂额。
这个问题的本质是什么？是耦合。业务逻辑和具体的模型客户端绑死了，模型厂商一换，业务代码跟着遭殃。生成式AI应用跟传统后端不一样的地方在于，模型是外部依赖，而且是一个会频繁更换、版本迭代极快、厂商差异极大的外部依赖。今天 A 模型效果好，明天 B 模型更便宜，后天公司自研模型上线要灰度。如果你一开始就没把“模型”这个维度抽象出来，后面每一次切换都是一次大手术。
所以，做生成式AI应用，从第一天就应该把设计模式用上。不是那种为了用而用的炫技，而是为了解决几个非常具体的问题：
- 隔离变化：模型供应商、模型版本、参数配置的变化不影响业务层。
- 统一入口：所有模型调用走同一个接口，调用方不用关心背后是哪个模型。
- 灵活路由：根据成本、延迟、效果、用户等级等维度动态选择模型。
- 可测试可回滚：想切回旧模型，改一行配置就行，不用动代码。
这个系列的读者应该都清楚，生成式AI应用不是“调个接口返回文本”那么简单。从用户输入到最终输出，中间要经过输入清洗、上下文组装、模型调用、输出校验、格式化、缓存、重试、降级、日志追踪……每个环节都有不同的策略。设计模式在这里的意义，就是帮你把这串链条拆成一个个可替换的积木，而不是一团缠在一起的毛线。
2. 先拆三角组合：策略、工厂、适配器到底各自扛什么活
聊多模型路由，最核心的就是三个设计模式的组合：策略模式负责“选谁”，工厂模式负责“造谁”，适配器模式负责“包装谁”。这三者各司其职，但经常被混为一谈。我一个个拆开讲清楚它们的分工。
2.1 策略模式：把“选哪个模型”变成可插拔决策
策略模式解决的是“一组算法可以互相替换”的问题。放到生成式AI场景里，最常见的应用就是：面对多个可用模型，你到底把请求发给谁？
有人会写：
def route_request(user_input): if user_input.is_urgent(): model = "fast-model" elif user_input.is_complex(): model = "powerful-model" else: model = "cheap-model" return call_model(model, user_input)业务不复杂的时候这么写没问题，但只要路由维度一多——用户等级、成本预算、当前延迟、模型健康状态、内容类型——这段代码就变成一个巨大的分支泥潭，每次改需求都要小心翼翼地往里塞条件。
用策略模式重构，核心是定义一个路由策略接口，每种策略单独一个类：
from abc import ABC, abstractmethod class RouteStrategy(ABC): @abstractmethod def route(self, candidates, context): pass class CostFirstStrategy(RouteStrategy): def route(self, candidates, context): # 挑最便宜的可用模型 return min(candidates, key=lambda m: m.unit_cost) class LatencyFirstStrategy(RouteStrategy): def route(self, candidates, context): # 挑响应最快的可用模型 return min(candidates, key=lambda m: m.avg_latency_ms) class FallbackStrategy(RouteStrategy): def route(self, candidates, context): # 主模型失败时按优先级选择备用 for model in candidates: if model.health_status == "healthy": return model raise RuntimeError("No available model")这样，路由规则的变化被隔离在每个策略类内部，新增一种策略不用动其他代码。策略之间还能组合——先按成本过滤到候选集，再按延迟排序，最后走降级逻辑。这跟传统Java/C++里策略模式的用法完全一致，只是“算法”从排序变成了“模型选择”。
2.2 工厂模式：统一创建模型客户端，消灭散落的 new
工厂模式解决的是对象创建的问题。在生成式AI应用里，最大的坑就是每个服务自己创建模型客户端，导致配置分散、连接数失控、无法统一管理超时和重试。
不同类型的模型客户端（OpenAI兼容接口、Anthropic风格接口、自研模型网关），创建方式和初始化参数各不相同。如果在业务代码里 new，那换厂商就是一场灾难。
用简单工厂聚合创建逻辑：
class ModelClientFactory: def __init__(self): self._clients = {} def get_client(self, profile: ModelProfile): """根据配置返回共享客户端实例，避免重复创建""" if profile.name in self._clients: return self._clients[profile.name] client = self._build_client(profile) self._clients[profile.name] = client return client def _build_client(self, profile: ModelProfile): if profile.provider == "openai_compatible": return OpenAICompatibleClient(profile.base_url, profile.api_key, profile.timeout) elif profile.provider == "anthropic": return AnthropicClient(profile.api_key, profile.model_name) elif profile.provider == "internal_gateway": return InternalGatewayClient(profile.endpoint, profile.auth_token) else: raise ValueError(f"Unsupported provider: {profile.provider}")工厂带来的好处不只是创建逻辑集中，更重要的是客户端复用。一个模型实例对应一组连接池和速率限制，如果每个请求都新建，压测一上来连接直接被打爆。我见过生产环境里因为没做客户端复用，导致上游网关把整个服务限流的案例，非常惨痛。
2.3 适配器模式：不同模型差异巨大，必须套一层“翻译官”
如果说策略模式解决“选谁”，工厂模式解决“造谁”，那适配器模式解决的就是“怎么统一称呼”。不同模型的接口差异真的太大了，我列个常见的差异清单：
| 差异维度 | 示例 |
|---|---|
| 消息格式 | 有的用messages=[{role, content}]，有的用prompt字符串，有的还要分 system/user/assistant |
| 参数命名 | 有的叫max_tokens，有的叫max_tokens_to_sample |
| 返回结构 | 有的是choices[0].message.content，有的是直接返回纯文本字段 |
| 流式格式 | 有的是data: {json}，有的是data: [DONE]，有的是字节流 |
| 错误码 | 限流、超时、敏感内容，各家状态码和错误体都不一样 |
业务层不应该感知这些差异。适配器模式在这里就是做一层“翻译”，把不同模型的请求转换成业务层统一的调用格式，再把模型的响应转换成统一的结果对象：
class UnifiedLLMAdapter(ABC): @abstractmethod def chat(self, messages, **kwargs) -> LLMResponse: pass class OpenAICompatibleAdapter(UnifiedLLMAdapter): def __init__(self, client): self.client = client def chat(self, messages, **kwargs): raw = self.client.chat.completions.create( messages=messages, temperature=kwargs.get("temperature", 0.7), max_tokens=kwargs.get("max_tokens", 1024), ) return LLMResponse( text=raw.choices[0].message.content, usage=raw.usage.total_tokens, raw=raw ) class AnthropicAdapter(UnifiedLLMAdapter): def __init__(self, client): self.client = client def chat(self, messages, **kwargs): raw = self.client.completions.create( prompt=self._to_prompt(messages), max_tokens_to_sample=kwargs.get("max_tokens", 1024), ) return LLMResponse(text=raw.completion, usage=raw.usage.total_tokens, raw=raw)这样业务层拿到的永远是LLMResponse，永远调的是chat()，至于背后是哪个厂商、参数叫什么，全部隔离在适配器内部。我建议团队里所有调用模型的代码，只认这个统一接口，不允许直接碰原始 SDK。
3. 一块落地：搭一个支持热切换的多模型路由框架
前面三个模式单独拆开说不难，难的是组合起来形成一个真正能上生产的路由框架。我直接给一套可运行的骨架，代码是Python风格，但思路换成Java、Go、C#完全成立。
3.1 定义模型Profile与路由策略
第一步是定义模型的基本信息配置，也就是把“有哪些模型可用、各自什么属性”集中管理：
from dataclasses import dataclass from typing import List, Dict, Optional @dataclass class ModelProfile: name: str # 唯一标识：text-4o-mini、claude-3-haiku provider: str # openai_compatible / anthropic / internal_gateway model_name: str # 厂商侧的真实模型名 unit_cost: float # 每千token成本，用于成本优先路由 avg_latency_ms: int # 平均延迟，用于延迟优先路由 health_status: str # healthy / degraded / down capabilities: List[str] # ["text", "vision", "json_mode"]路由策略接口就是前面代码里的RouteStrategy，接收候选模型列表和调用上下文，返回选中的模型：
@dataclass class RouteContext: user_tier: str # free / premium / enterprise task_type: str # chat / summary / extraction max_budget_cents: float require_json: bool # 是否必须支持JSON模式这样路由就不再是拍脑袋写死，而是基于 profile 数据动态决策。生产环境里，这些 profile 可以存在配置中心或数据库里，模型健康状态由监控任务定期刷新，成本数据从账单服务同步。每次路由请求时，拉取最新的候选列表，策略在内存里跑一遍，非常轻量。
3.2 工厂 + 策略 + 适配器的集成实现
有了 profile 和策略，接下来就是把三者组装起来。核心是一个ModelRouter类，对外只暴露一个chat()方法：
class ModelRouter: def __init__(self, factory: ModelClientFactory, strategy: RouteStrategy): self.factory = factory self.strategy = strategy self._adapters: Dict[str, UnifiedLLMAdapter] = {} def register_adapter(self, profile: ModelProfile): client = self.factory.get_client(profile) if profile.provider == "openai_compatible": self._adapters[profile.name] = OpenAICompatibleAdapter(client) elif profile.provider == "anthropic": self._adapters[profile.name] = AnthropicAdapter(client) return self._adapters[profile.name] def chat(self, messages, candidates, context, **kwargs): # 1. 按当前策略选出目标模型 selected = self.strategy.route(candidates, context) # 2. 拿到对应的适配器 adapter = self._adapters[selected.name] # 3. 统一调用 try: return adapter.chat(messages, **kwargs) except ModelUnavailableError: # 4. 降级：从候选里剔除故障模型，换策略重试 healthy_candidates = [m for m in candidates if m.name != selected.name] fallback_strategy = FallbackStrategy() fallback_model = fallback_strategy.route(healthy_candidates, context) fallback_adapter = self._adapters[fallback_model.name] return fallback_adapter.chat(messages, **kwargs)这个框架跑起来的核心链路是：调用方给出一组候选模型和上下文，路由策略选出最优模型，工厂确保客户端存在，适配器抹平接口差异，调用完成后返回统一格式。调用方完全不知道模型是谁、策略怎么选、适配器做了多少翻译工作。
3.3 动态路由与降级切换的配置实践
代码骨架有了，实际部署时还要解决“热切换”的问题。热切换指的是：不重启服务，就能调整路由策略、增减候选模型、修改模型参数。
我的做法是引入一个RouterConfig，支持动态刷新：
class RouterConfig: def __init__(self, raw_config: dict): self.profiles: Dict[str, ModelProfile] = {} self.route_policy = raw_config.get("route_policy", "latency_first") self.candidates = raw_config.get("candidates", []) self._load_profiles(raw_config.get("models", [])) @classmethod def from_config_center(cls, config_center_client, key: str): raw = config_center_client.get(key) return cls(raw) def refresh(self): latest = self.__class__.from_config_center(...) self.profiles = latest.profiles self.candidates = latest.candidates self.route_policy = latest.route_policy配置中心一有变更，路由器拉取最新配置，下一次请求自动使用新的策略和候选集合。我建议把配置刷新的间隔控制在30秒左右，不要太频繁，避免配置中心压力过大。
线上降级的场景我举个实际例子：主模型A因上游故障开始大量超时，健康检查模块把A标记为 degraded，配置中心里路由策略自动从 cost_first 切换为 health_first，候选列表排除A。整个过程不需要发版，业务代码零改动。这就是设计模式组合带来的直接价值——把变化集中到该变化的地方，让稳定的部分真正稳定下来。
4. 链路编排：把单次生成变成可控流水线
路由框架解决的是“调用哪个模型”的问题，但真实的生成式AI应用远不止一次模型调用。从用户输入到最终返回，中间往往是一条复杂的链路：输入清洗、上下文检索、提示词组装、模型调用、输出校验、格式化、知识库写入……这时候需要另一组设计模式来管好整条链路。
4.1 模板方法模式：固定生成流程骨架
模板方法模式的思路是：父类定义算法的骨架，子类重写其中的步骤。放到生成式AI场景里，最适合的就是“标准生成流程”——大部分生成任务的骨架是固定的，变的是各个步骤的具体实现。
我定义一个抽象基类：
class GenerationPipeline(ABC): def run(self, user_input, context): cleaned = self.preprocess(user_input) # 输入清洗 prompt = self.build_prompt(cleaned, context) # 构建提示词 messages = self.prepare_messages(prompt) response = self.call_model(messages) # 调用模型 validated = self.validate_output(response) # 输出校验 return self.format_result(validated) # 统一格式化 @abstractmethod def preprocess(self, user_input): pass @abstractmethod def build_prompt(self, cleaned_input, context): pass @abstractmethod def prepare_messages(self, prompt): pass @abstractmethod def call_model(self, messages): pass @abstractmethod def validate_output(self, response): pass @abstractmethod def format_result(self, validated_response): pass然后不同的业务场景继承这个类，各自实现具体步骤：
class SummaryPipeline(GenerationPipeline): def build_prompt(self, cleaned_input, context): return f"请将以下内容总结为三条要点：\n{cleaned_input}" def validate_output(self, response): # 检查是否输出了三个要点，不足则重试 if len(response.text.split("1.")) < 3: raise InvalidOutputError("summary incomplete") return response class ExtractionPipeline(GenerationPipeline): def build_prompt(self, cleaned_input, context): return f"从以下文本中抽取JSON字段：name, date, amount\n{cleaned_input}" def validate_output(self, response): # 尝试解析JSON，失败则自动修复 return fix_json_or_retry(response.text)模板方法的好处是，新人加入团队，不需要理解整条链路，只要看懂自己负责的那几个抽象方法，就能快速接入新业务。而且流程骨架一旦定死，就不会有人在某个业务里偷偷跳过校验，质量基线是全局拉齐的。
4.2 责任链模式：让前后处理步骤按需串联
模板方法模式固定的是整个流程的“必然步骤”，但有些处理环节是可选的、可插拔的。比如输入侧可能有敏感信息过滤、恶意输入拦截、（合规场景下的）内容过滤；输出侧可能有格式校验、内容修正、敏感词替换、引用标注。这些环节不该被硬编码进流水线里，而应该可以自由组合、动态增减。
责任链模式正好解决这个问题。每个处理节点实现同一接口，节点之间串联成链，请求依次经过每个节点，任一节点可以选择继续传递或终止：
class ProcessingNode(ABC): def set_next(self, node): self._next = node return node def handle(self, data): processed = self.process(data) if self._next: return self._next.handle(processed) return processed @abstractmethod def process(self, data): pass class InputNormalizer(ProcessingNode): def process(self, data): # 去掉多余空格、统一换行符 return data.strip().replace("\r\n", "\n") class PromptInjectionFilter(ProcessingNode): def process(self, data): # 检测并移除常见的提示注入模式 data.blocked = detect_injection(data.text) return data class OutputJsonFormatter(ProcessingNode): def process(self, data): if data.require_json: data.text = ensure_valid_json(data.text) return data组装方式：
normalizer = InputNormalizer() injection_filter = PromptInjectionFilter() json_formatter = OutputJsonFormatter() normalizer.set_next(injection_filter).set_next(json_formatter) result = normalizer.handle(user_input)责任链的核心价值在于新增处理环节不需要改原有代码，只要往链里插一个新节点。我之前在一个项目里做多租户隔离，不同租户需要的输出格式不同，就是通过给每个租户配置不同的责任链来实现的，完全没动核心流程。
4.3 编排中的状态管理与输出约束
链路编排还有一个容易被忽视的问题——状态管理。电子书阅读器有阅读进度，生成式AI应用也有“对话状态”“上下文窗口状态”“重试状态”。
我建议用上下文对象贯穿整个链路，避免在多个函数参数之间传一堆零散值：
@dataclass class GenerationContext: request_id: str user_id: str session_id: str input_text: str messages: List[dict] selected_model: Optional[str] = None retry_count: int = 0 usage: Dict[str, int] = None extra: Dict[str, any] = None链路里的每个环节都读写这个上下文，方便日志追踪，也方便在任意环节拿到全链路信息。输出约束方面，我会给模型输出加一层“合约”：要求返回JSON就严格校验JSON，要求返回枚举值就校验枚举范围，不符合就自动修复或重试一次，再不行就走降级文案。这个约束逻辑放在适配器之外、业务层之内，确保它不依赖具体模型。
5. 实战避坑：从原型到生产，这五类问题最磨人
框架搭得再漂亮，上了生产环境还是会被现实毒打。我把这一年多踩过的坑总结成五类，给读者一个速查表。
| 问题类别 | 典型症状 | 核心原因 | 我的处理方案 |
|---|---|---|---|
| 模型输出格式不稳定 | 时而有JSON，时而带markdown，时而瞎编字段 | 模型是概率输出，天然不稳定 | 输出校验+自动修复+最大重试2次，再失败走兜底模板 |
| 超时与重试 | 请求卡死、上游超时、重试风暴 | 模型延迟波动大，没有统一超时控制 | 统一超时配置，重试走指数退避+随机抖动 |
| Token成本失控 | 月底账单吓人 | 没有对输入输出Token做预算控制 | 路由策略里加成本阈值，超预算自动降级到低成本模型 |
| 提示词版本混乱 | 改了提示词，线上效果时好时坏 | Prompt没有版本管理和发布流程 | Prompt存配置中心，带版本号，支持按流量灰度 |
| 日志追踪困难 | 线上报错查不到是哪个环节挂了 | 没有全链路追踪 | 每个请求生成request_id，贯穿所有链路日志 |
5.1 模型结果格式不稳定，怎么治
这个是所有生成式AI应用都要面对的“天灾”。模型不是数据库，同一个提示词跑十次可能出十种格式。我的经验是：不要相信模型的自觉，要在外层强校验强修复。
你可以在适配器层返回原始内容后，再经过一个OutputContractValidator，做三件事：
- 校验：用正则或解析器验证输出是否符合约定格式。
- 修复：如果是小问题（比如JSON多处一个逗号、多了markdown代码块标记），自动修复。
- 重试：如果修复不了，带着错误信息重新调用模型，提示词里明确告诉它上次错在哪。
注意控制重试次数，我一般最多重试2次，再多就是烧钱了。同时失败路径要有兜底，返回一个预设的降级文案，别让接口裸奔报错。
5.2 超时重试与熔断，别让上游抖动拖垮整个服务
模型服务的延迟方差极大，高峰期动辄几十秒。如果不设超时，一个慢请求会一直占着连接池；如果超时设得过短，模型本来能正常返回却频繁重试，反而造成重试风暴。
我的做法是分三层设置：
- 连接超时：3-5秒，太久说明网络或路由有问题。
- 读超时（首字节）：10-15秒，模型首字输出太慢一般就是排队了。
- 总超时：30-60秒，看任务复杂度设置。
重试策略用指数退避加随机抖动，避免所有请求在同一时间点重试，把上游打挂。如果连续N次失败，就触发熔断，直接降级到备用模型，连试都别试了。这个逻辑放在路由器的调用层，对业务方透明。
5.3 Token成本失控，必须设预算红线
生成式AI的计费逻辑跟传统API完全不一样，同一套流程，用户多聊几句、上下文长一点，成本就指数级上涨。成本控制不能靠财务月底发现，要在路由阶段就算好账。
我在RouteContext里加了max_budget_cents字段，每次路由前估算这次请求的Token消耗，如果超过预算，直接不走贵模型，改选一个低成本模型或者干脆拒绝生成。对于长上下文场景，我会在调用前做一次上下文裁剪，把不重要的历史对话丢进摘要，控制输入Token规模。
5.4 提示词版本管理，别让线上效果玄学化
很多团队把提示词当常量写在代码里，改一次发一次版，这是最低效的做法。提示词的迭代频率远高于代码发版频率，必须独立管理。
我用配置中心存提示词，每条提示词有版本号、生效环境、灰度比例。上线一个新提示词，先10%流量观察效果，没问题再全量，有问题一键回滚到旧版本。配合全链路日志，就能把“哪次效果变化是什么版本导致的”对上号。这套机制搭好之后，我再也没因为改提示词发过紧急版本。
5.5 全链路可观测性，没有追踪等于摸黑排查
生成式AI应用的链路比传统接口长得多，从入口到出口要经过清洗、检索、拼提示词、模型调用、校验、格式化好几个环节，任何一个环节出问题都可能导致最终结果变差。如果没有追踪，线上反馈“摘要质量不行”，你根本不知道是检索没召回、提示词不够好、还是模型本身的问题。
我在每个请求进来时就生成一个request_id，塞进GenerationContext，所有环节的日志都带上这个ID。模型调用的请求里也加上这个ID作为标识，万一上游报错，可以拿着这个ID去厂商侧查。日志除了记录耗时和Token用量，还会记录输出校验是否通过、是否触发重试、是否降级。这样才能从数据上判断，质量问题是偶发的还是系统性的。
6. 我的几点私人体会
这套东西从搭建到稳定运行，我最大的体会是：不要把设计模式当成“框架”，而要当成“契约”。策略模式、工厂模式、适配器模式、模板方法、责任链，本质上是定义了一套稳定的接口契约，把容易变化的因素隔离在契约之外。模型会变、提示词会变、成本策略会变，但如果契约稳定，这些变化都只是配置和实现层面的调整，不会波及整个系统。
另外一点是关于团队协作的。有了清晰的模式划分，每个人的工作边界也清晰了：A负责维护适配器，B负责调优路由策略，C负责处理输出校验。互相之间只需要遵守接口契约，不用关心对方内部实现。这对生成式AI这种快速迭代的领域太重要了——你不可能让每个人都理解所有细节，但你可以让每个人都在自己负责的层里做到极致。
最后一个小技巧：所有模式相关的代码，一定要配上充分的单元测试。策略选型要测，适配器转换要测，降级路径更要测。生成式AI没法保证输出固定，但至少你要保证路由和编排逻辑是确定性的，否则排查问题的时候根本分不清是模型的问题还是你代码的问题。把确定性的部分焊死，把不确定性的部分隔离，这是做生成式AI工程化最重要的一条原则。
