★ FEATURED ARTICLE 1. 从“调用确定函数”到“编排概率模型”生成式AI对设计模式的根本冲击先说一段我自己的经历。去年我接手一个客服工单分类项目按照过去写业务系统的经验上来就用经典的策略模式加工厂模式把代码搭得整整齐齐。各个模块各司其职接口定义得清清楚楚。结果一联调问题全出在最核心的那个环节——模型调用。同一个Prompt有时候能正确分类有时候会把“退款”判成“投诉”而且不同供应商的返回格式五花八门有的给JSON有的在JSON外面包一层Markdown代码块有的干脆把一大段解释性文本直接丢回来。我那些设计模式在模型输出的不确定性面前几乎帮不上忙。后来我意识到一个关键问题传统设计模式的隐含前提是确定性。调用一个函数给定同样的输入返回结果是确定的所以可以把功能拆成稳定的小模块用接口去约束彼此协作。但生成式AI不一样你面对的是一个概率系统同样的上下文可能得出不同的结果性能有波动成本有波动连模型能力本身也在快速迭代。这种情况下设计模式并没有失效而是它的“含义”变了——从“封装确定性的实现”变成了“封装不确定性之外的稳定性”。换句话说生成式AI时代的架构设计核心目标变成了三件事把模型变化隔离掉、把业务稳定性沉淀下来、把不可控的输出纳入可控的流程。这也是我在这个系列里反复强调的设计模式的本质是“解决反复出现的结构性问题”而生成式AI引入的恰恰是一批全新的结构性问题比如上下文怎么管理、多模型怎么切换、失败怎么重试、输出怎么校验、成本怎么控制。这些问题用旧模式硬套会很难受但把它们拆开看其实每一个背后都能找到对应的模式思想。这篇文章是系列第十篇。到了这个位置我不想再单纯罗列“AI设计模式清单”而是想把前九篇的内容整合起来站在实战角度做一次系统性复盘哪些经典设计模式被AI重构了、哪些真正属于AI时代的新模式、以及一套可以落到代码里的完整案例。适合正在做RAG、智能客服、内容生成工具、AI Agent的同学参考也适合那些刚看完设计模式理论、想知道“这东西在AI项目里到底怎么用”的读者。2. 经典设计模式在AI场景下的“重新发明”旧模式没死只是换了副面孔。下面这几个是AI项目里我用得最多也最值得重新理解的经典模式。2.1 工厂模式从“对象生产”到“模型供应商管理”传统工厂模式负责创建对象调用方不需要关心具体实现类是哪个只需要一个统一接口。在AI场景下这个思想直接对应到模型供应商管理——你的代码可能同时接入OpenAI、国产大模型、本地部署的模型每个供应商的API地址、鉴权方式、模型名称、限流策略都不一样。如果这些细节散落在业务代码里换模型就是一场灾难。工厂模式在这里的最佳实践是做一个“模型客户端工厂”class LLMClientFactory: _registry {} classmethod def register(cls, provider: str, builder): cls._registry[provider] builder classmethod def create(cls, provider: str, model_name: str, **kwargs): builder cls._registry.get(provider) if not builder: raise ValueError(fUnknown provider: {provider}) return builder(model_namemodel_name, **kwargs) # 使用方式 client LLMClientFactory.create(openai, gpt-4o, timeout30)这样业务代码只依赖create()方法底层换供应商、加新模型都不需要改业务逻辑。值得注意的是注册过程最好在配置加载阶段完成而不是运行时再动态判断这样启动阶段就能暴露配置错误。我实测下来的一个优化是工厂不只是创建客户端还可以顺带把模型能力元信息带上。比如哪些模型支持JSON输出、上下文窗口多大、每千token单价多少。这些信息在后面的路由和成本控制里非常有用属于“把参数对象升格为能力描述对象”的变体。2.2 策略模式从算法切换到Prompt策略族传统策略模式用来解决“一组算法可以互相替换”的问题比如排序策略、加密策略。在生成式AI场景里最自然的映射是Prompt策略族——针对同一个问题可以有零样本、少样本、思维链、反思式等多种策略它们的成本不同、效果不同、时延不同。很多团队的Prompt散落在各个调用点改一版提示词要全局搜索替换。用策略模式重新组织之后结构会清爽很多class ExtractionStrategy(ABC): abstractmethod def build_messages(self, text: str) - list[dict]: ... class ZeroShotExtraction(ExtractionStrategy): def build_messages(self, text): return [ {role: system, content: 从用户反馈中提取退款原因。}, {role: user, content: text} ] class FewShotExtraction(ExtractionStrategy): examples [...] # 预先筛选好的示例 def build_messages(self, text): return [ {role: system, content: 你是工单分类助手参考示例输出。}, *self.examples, {role: user, content: text} ]这里的关键不只是“把Prompt封装起来”而是给策略加上效果元数据。我在生产环境里的做法是在策略对象上记录准确的评估分数和平均响应时间这样路由层可以基于历史表现动态选择策略。这在传统策略模式里不太会发生但在AI场景里策略本身的质量是一个需要持续监控的变量。2.3 观察者模式从事件通知到流式输出处理观察者模式在传统GUI里很常见一个按钮被点击多个监听器响应。在AI应用里它的新形态是流式输出的事件分发——大模型生成内容是一个token一个token返回的不同的订阅者需要实时处理不同的片段。比如一个AI写作助手前端要在流式返回时渲染Markdown后台需要在生成过程中检测敏感词日志系统要记录每个token的时间戳用于延迟分析。这时候如果所有逻辑都写在同一个流式回调里代码会迅速腐化。用观察者模式处理class TokenStream: def __init__(self): self._subscribers [] def subscribe(self, handler): self._subscribers.append(handler) async def emit(self, token: str): for handler in self._subscribers: await handler(token)看上去很简单但实际生产要考虑几个关键点订阅者之间的异常隔离、处理器的执行顺序、以及背压处理。我在项目里遇到过一个问题渲染处理器跑得慢拖慢了整条流式链路后来改成“每个订阅者独立消费一个异步队列”才解决。这相当于给观察者模式加了一层缓冲代价是内存占用上升但换来了主链路的稳定。2.4 门面模式统一入口掩盖多模型混乱门面模式的意义是给复杂子系统提供一个简单入口。在AI应用里这个思想几乎是刚需——因为一个有规模的AI项目内部往往同时存在多个模型服务、多套Prompt流程、多个后处理模块调用方如果直接面对这些东西协作成本极高。我在工单系统里做的一个门面是“CompletionService”class CompletionService: def __init__(self, router, fallback_handler, validator): self._router router self._fallback_handler fallback_handler self._validator validator async def complete(self, request: CompletionRequest) - CompletionResult: provider self._router.route(request) try: raw await provider.complete(request) except ProviderError: raw await self._fallback_handler.fallback(request) validated self._validator.validate(raw) return CompletionResult(validatedvalidated, rawraw, providerprovider.name())调用方只需要调用complete()完全不感知内部有路由、有重试、有校验。这个门面的价值在于它把“模型调用的复杂度”从业务逻辑里剥离出来了。业务方不需要关心“这个请求应该发给哪个模型”“失败了怎么办”“返回格式对不对”只需要说“我要完成这个任务”。这里有个设计教训门面不要做成“万能门面”。我见过有人把Prompt管理、上下文存储、成本统计全部塞进门面里结果这个门面变成了上帝类。合理的边界是——门面只负责编排不负责具体实现。上下文管理应该由独立的模块承担成本统计应该走日志管道门面只是协调它们。2.5 适配器模式把不同LLM的接口差异磨平适配器模式在AI项目里的角色比在任何传统系统里都重要。原因很简单不同LLM供应商的接口风格差异太大了。OpenAI用/chat/completionsAnthropic用/messages本地部署的模型可能是完全自定义的协议。更麻烦的是同一个模型在不同时间返回的结构也可能不一致。适配器要做的事情有两层第一层是协议适配把不同供应商的请求参数和返回结果转换成统一结构第二层是异常语义适配——有的供应商超时抛这个异常有的抛另一个有的限流返回429有的用429但不是限流这些差异必须统一成内部错误类型。class OpenAIAdapter: def __init__(self, client): self._client client async def complete(self, request: InternalRequest) - InternalResponse: try: resp await self._client.chat.completions.create( modelrequest.model, messagesrequest.messages, temperaturerequest.temperature, ) return InternalResponse( contentresp.choices[0].message.content, usageTokenUsage(resp.usage.prompt_tokens, resp.usage.completion_tokens) ) except RateLimitError: raise ProviderRateLimit(provideropenai) except TimeoutError: raise ProviderTimeout(provideropenai, retryableTrue)做适配器的时候我建议连token统计格式都统一而不是各用各的。后续做成本核算时你会发现“有的供应商返回的token数包含缓存命中、有的不算缓存”这种坑如果没有统一结构成本数据会非常混乱。3. AI应用真正需要的“新设计模式”除了经典模式的变形AI应用里还涌现出了一批传统设计模式没有覆盖的新结构。这些结构不是理论推演出来的而是从大量实际项目里反复沉淀出来的“事实标准”。3.1 上下文注入器Context InjectorRAG是AI应用最常见的架构而RAG的核心问题就是“怎么把检索到的内容塞进Prompt里”。没经验的团队直接在代码里做字符串拼接结果上下文越拼越乱token数超限、关键信息被截断、注入内容格式不一致各种问题层出不穷。上下文注入器的职责是统一决定上下文怎么组装、按什么顺序排列、哪些该保留哪些该丢弃。class ContextInjector: def __init__(self, max_tokens: int, formatter: ContextFormatter): self._max_tokens max_tokens self._formatter formatter def inject(self, system_prompt: str, user_question: str, retrieved: list[Document]) - list[dict]: ranked self._formatter.rank_documents(retrieved) budget self._max_tokens - estimate_tokens(system_prompt) - estimate_tokens(user_question) selected self._truncate_to_budget(ranked, budget) return [...]这个模式最大的价值是“上下文预算管理”。生成式AI的关键约束之一就是token窗口上下文注入器把“内容重要程度排序”和“token预算分配”这两个关注点封装起来业务代码就不需要再关心细节。我在项目中往往会把元数据注入也放进来做比如当前时间、用户画像、历史交互摘要这些信息对模型回答质量影响很大但如果没有统一注入器每个业务方各加一段很快上下文就会爆炸。3.2 意图路由器Intent Router当系统接入多个模型时必然出现“什么请求该走什么模型”的问题。我见过最朴素的方案是if-else写死这样的问题是没有弹性、不可灰度、不能基于实时状态做调整。意图路由器的核心逻辑是把请求分类映射到不同的处理链路。分类的依据可以是显式的意图标签也可以是嵌入向量的相似度甚至是模型自身给出的判断。class IntentRouter: def __init__(self, rules: list[RouteRule], default_route: str): self._rules rules self._default_route default_route def route(self, request: CompletionRequest) - Route: for rule in self._rules: if rule.matches(request): return rule.target_route return self._default_route我做路由时一般会引入“模型能力矩阵”而不是简单映射。比如简单分类任务走小模型代码生成走代码专用模型复杂推理走强推理模型敏感内容走带审核的模型。路由规则必须支持运行时更新因为模型能力和成本策略变化太快。有条件的团队可以直接把路由规则做成配置中心里的动态条件这样灰度切换模型就不需要发版。3.3 自我修正循环Self-Correction Loop传统软件里代码出错是bug修一次就好。但生成式AI输出质量不佳是常态甚至同一个Prompt这次成功下次失败。所以AI应用需要一种“输出质量验证自动修正”的循环结构我称之为自我修正循环。这个模式的骨架是生成 → 校验 → 不通过则修正 → 再生成 → 达到最大轮次后降级。核心在于校验器设计——校验器不能只做格式检查还要做语义层面的检查。async def self_correcting_generate(generator, validator, max_attempts3): last_output None for attempt in range(max_attempts): output await generator.generate() result await validator.validate(output) if result.is_valid: return output, result last_output output # 将校验失败原因反馈给生成器做修正 return last_output, ValidationResult(is_validFalse, errors[max_attempts_exceeded])这里容易踩的坑是修正轮数不能太多。每轮修正都意味着额外的token成本和延迟实测下来很多场景2到3轮就够了超过3轮成功率提升就非常有限成本却线性上升。另一个值得重视的点是校验器本身也可能出错比如漏判了关键信息所以修正反馈里最好带上面向生成器的“为什么失败”的具体说明而不是简单的一句“输出不符合要求”。3.4 可观测性代理Observability Proxy传统系统有日志、有监控。但AI应用的特殊之处在于同样的输入输出质量问题往往无法从代码逻辑上完全解释必须依赖对Prompt、模型版本、参数配置、运行成本的细致观测。我见过太多AI项目排障靠“重新跑一次试试”这种排障方式的效率太低了。可观测性代理做的事情是在模型调用链路上透明地记录完整上下文——请求和响应的完整内容、token消耗、时延、模型版本、当时的Prompt策略、命中哪些缓存、有无重试。这样当线上出现质量问题时可以回到过去复现和分析。class ObservabilityProxy: def __init__(self, inner_service, recorder): self._inner inner_service self._recorder recorder async def complete(self, request): start time.perf_counter() try: response await self._inner.complete(request) self._recorder.record_success(request, response, latencytime.perf_counter() - start) return response except Exception as e: self._recorder.record_failure(request, errore, latencytime.perf_counter() - start) raise这个模式看起来简单难在“记录什么、怎么存储、怎么检索”。我的经验是至少要支持按request_id追踪一条完整链路从用户请求进入系统开始到最终响应返回中间经过了多少次模型调用、每次调用用了哪个模型、Prompt是什么、输出是什么全部串成一条trace。这样出了问题不需要用户复述直接看trace就能定位。4. 完整案例多模型客服工单处理系统说了一堆模式把它们串起来才是真正的功夫。我用一个实际做过的多模型客服工单处理系统为例子把前面所有模式落一遍。4.1 需求与约束系统的功能是接收用户提交的工单文本自动完成意图分类、紧急度判断、回复建议生成。约束条件很现实预算敏感不能什么请求都用大模型需要有一个分级路由策略。部分工单涉及用户隐私必须走本地部署模型不能走外部API。模型会频繁升级代码不能跟着模型版本频繁改。线上偶发供应商超时需要有降级方案。运营方需要看到每个工单的完整处理链路用了哪个模型、为什么用这个模型、耗时多久。4.2 整体结构整个系统的结构分为四层第一层是接入层通过门面模式的CompletionService统一接收内部各业务方的处理请求。业务方完全不感知下游模型细节。第二层是路由决策层核心是意图路由器。路由器会读取工单内容先做一个轻量级分类再根据分类结果和敏感标记决定走哪个模型链路。比如简单工单走本地小模型复杂投诉走外部强模型敏感信息走本地专用模型。第三层是模型适配层每个供应商都有对应的适配器统一转换成内部响应格式。这一层只负责“把模型差异抹平”。第四层是增强处理层包括上下文注入器把历史工单、用户信息注入到Prompt里、自我修正循环生成结果校验不通过时自动修正、可观测性代理全链路记录。4.3 关键代码片段路由决策是系统最核心的部分我贴一段精简代码class TicketRouter: def __init__(self, classifier, factories, sensitive_detector, recorder): self._classifier classifier self._factories factories self._sensitive_detector sensitive_detector self._recorder recorder async def route(self, ticket: Ticket) - Route: # 1. 先做意图分类走便宜的本地模型 intent await self._classifier.classify(ticket.text) # 2. 敏感信息检测 is_sensitive self._sensitive_detector.contains(ticket.text) # 3. 路由决策 if is_sensitive: return Route(providerlocal-secure, modellocal-7b, prioritynormal) elif intent complex_complaint: return Route(provideropenai, modelgpt-4o, priorityhigh) elif intent simple_query: return Route(providerlocal, modellocal-3b, prioritylow) else: return Route(providerfallback, modelmedium, prioritynormal)这段代码的精髓在于分类动作本身就走的是小模型因为它只需要判断意图的粗粒度方向不需要生成内容用便宜模型就够。很多团队一上来就把工单文本发给大模型做“分类生成建议”三合一结果成本立刻失控。模型调用主链路class CompletionService: def __init__(self, router, adapter_factory, validator_factory, recorder): self._router router self._adapter_factory adapter_factory self._validator_factory validator_factory self._recorder recorder async def handle_ticket(self, ticket: Ticket) - TicketResult: route await self._router.route(ticket) # 构建上下文 docs await retrieve_docs(ticket.id) messages ContextInjector(...).inject( system_prompt你是客服助手回答格式为JSON。, user_questionticket.text, retrieveddocs ) adapter self._adapter_factory.create(route.provider) # 带自我修正的生成 generator PartialGenerator(adapter, messages) validator self._validator_factory.create(route.task_type) output, result await self_correcting_generate(generator, validator, max_attempts2) # 记录全链路信息 self._recorder.record_ticket_trace(ticket.id, route, messages, output, result) return TicketResult(...)这段代码体现的就是门面、路由、适配器、注入器、自我修正循环的协作。每一层只做自己的事出了问题能很快定位到具体层。4.4 实测效果与成本表现这个系统上线后我统计了两周数据几个数字比较有参考价值工单类型路由目标平均延迟单条成本准确率简单查询本地3B模型0.8秒极低92%普通工单中等模型2.1秒中87%复杂投诉GPT-4o4.5秒高95%敏感内容本地7B模型1.5秒低90%混合路由策略和纯大模型方案相比整体成本大约下降了60%延迟下降了45%而最关键工单的准确率没有明显下降。不是所有请求都需要最强模型需求分级本身就能省一大半钱。这是意图路由最实在的收益。当时线上也遇到过意外情况外部模型有一次持续了二十分钟的故障网关层超时大量抛出。因为这个系统每个环节都有超时控制和降级策略复杂投诉自动降到本地模型处理虽然回答质量略降但整个系统没有宕机。这就是门面加路由加降级配合起来的效果。5. 我在实战中踩过的坑与边界模式提供了思路但真正落地时有一堆细节问题。这一章写几个我亲测吃亏后总结出来的教训。5.1 过度设计是真实存在的陷阱有一个项目我一开始就把所有模式全部套上去工厂、门面、代理、观察者、策略……结果代码量翻了三倍团队同事接手时一脸茫然。后来我意识到设计模式不是装饰品是为了解决真实存在的复杂度。如果系统只接一个模型、调用点只有两处门面模式和适配器模式带来的收益就非常有限。我的判断标准是当“变更的代价”明显升高时才值得引入模式。比如第一次需要在A模型和B模型之间做切换时再做工厂和适配器当第三个调用方出现时再做门面当Prompt策略开始出现差异化效果时再把策略模式引进来。提前设计的成本往往比事后再重构要高。5.2 没有模拟器测试就是空中楼阁AI应用的测试是老大难问题。每次调用真实模型结果都是概率性的不可能用传统断言做回归测试。我的做法是构建一个模型模拟器在测试环境里用预先录制好的响应来替代真实模型调用这样核心业务逻辑可以稳定测试CI才能跑得起来。class MockAdapter: def __init__(self, response_map): self._response_map response_map async def complete(self, request): key request.messages[-1][content] return self._response_map.get(key, InternalResponse(contentmock, usageTokenUsage(1, 1)))模拟器解决了“逻辑正确性”的回归问题但真实模型的质量验证还是需要单独的评测集。我的做法是维护一个三五十条真实样本的评测集每次模型或Prompt有变更就批量跑一遍人工抽查质量分。这个成本不高但能大幅减少线上翻车概率。5.3 模式会失效的时刻有几个场景我明确建议放弃复杂模式第一个是原型验证阶段。这个阶段代码就是用来快速验证“这个功能到底可不可行”的先直接用最简单的方式调模型跑通了再谈结构化。我见过不少团队连MVP都没跑通就在纠结适配器接口怎么定义这不是工程素养是资源浪费。第二个是单模型单链路的工具类脚本。比如一个离线数据清洗脚本只调一个模型一次不需要路由、门面、观察者这些结构函数式写法反而清晰。第三个是团队没有足够熟悉模式的人维护时。设计模式是“结构是一种沟通”但如果团队里没人看得懂这套结构反而成了维护负担。这个判断虽然听起来不够“技术”但在实际项目里非常重要。5.4 给后来者的三个操作建议如果让我给正在建设AI应用的团队提三个建议我会这么说第一先把可观测性做起来再谈优化。没有全链路trace你根本不知道质量下降是模型换代导致的、还是上下文排序变化导致的、还是路由规则改错了。可观测性代理的搭建成本并不高但它决定了你能不能从“靠运气”变成“靠数据”。第二把模型变更当发布事件管理而不是代码变更。模型的Prompt策略、版本升级都要走评审和灰度流程。我见过太多线上问题最后定位到“昨天顺手改了一个system prompt”。给Prompt配置一个版本号和全链路trace是AI项目稳定性的基石。第三用成本指标驱动架构演进。不要只看准确率要把单次调用成本和整体预算一起纳入设计考量。当你把成本指标挂到路由决策层你会发现许多架构上的优化方向会自己浮现出来。坦白说生成式AI的出现确实让一部分经典设计模式显得“老派”但它真正改变的不是“要不要用模式”而是“模式要解决的问题变了”。从函数调用的确定性世界走进概率模型的非确定性世界所幸软件工程沉淀下来的抽象分层思想没有过时只是需要换一批具体的实现方式。如果你也在做AI应用不妨对照上面的模式重新审视一下自己的代码——哪些复杂度被隐藏了哪些复杂度被放大了这比纠结“用没用对设计模式”要更有价值。
阅读完成 · 觉得有帮助？
