多模型时代如果还在用 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, keylambda m: m.unit_cost) class LatencyFirstStrategy(RouteStrategy): def route(self, candidates, context): # 挑响应最快的可用模型 return min(candidates, keylambda 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(fUnsupported 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( messagesmessages, temperaturekwargs.get(temperature, 0.7), max_tokenskwargs.get(max_tokens, 1024), ) return LLMResponse( textraw.choices[0].message.content, usageraw.usage.total_tokens, rawraw ) class AnthropicAdapter(UnifiedLLMAdapter): def __init__(self, client): self.client client def chat(self, messages, **kwargs): raw self.client.completions.create( promptself._to_prompt(messages), max_tokens_to_samplekwargs.get(max_tokens, 1024), ) return LLMResponse(textraw.completion, usageraw.usage.total_tokens, rawraw)这样业务层拿到的永远是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工程化最重要的一条原则。
