这系列写到第十二篇,后台问法已经从“设计模式在AI开发里还有用吗”变成了“你到底在讲哪几个模式、代码长什么样”。说实话,我很喜欢这种变化,说明读者不再停留在概念围观,而是真的在把LLM应用当工程来做了。生成式AI应用的代码,难度不在“能跑”,而在“能维护、能测试、能在模型和供应商之间平滑切换”。这一篇不讲虚的,直接拆六个我实际在项目里反复打磨过的模式,全部脱胎于经典23种设计模式,但做了面向生成式AI场景的改造。核心关注三类:创建型怎么管理提示词装配、结构型怎么构建稳定的调用骨架、行为型怎么应对模型决策的不确定性。无论你是正在写提示词工程的初学开发者,还是已经做过几个Agent应用的资深工程师,这篇都能直接抄作业。
1. 内容整体设计与思路拆解
1.1 为什么生成式AI应用需要设计模式
很多人有个误区,认为LLM应用是“提示词艺术”,只要prompt写得好就行,代码随便堆堆也能出效果。这个判断在小Demo阶段成立,一旦进入真实业务——比如多轮客服工单、内容审核流水线、知识库问答系统——代码复杂度会指数级上升。
我见过太多项目死在同一个坑里:一个main函数从读取用户输入到调用模型再到解析结果,全塞在一坨里。上下文拼接用f-string裸拼,模型名硬编码在十几个地方,换模型供应商要改的代码点多到需要全局搜索。这种代码不是不能跑,是一周后你自己都改不动。
设计模式解决的根本问题就是“对变化的隔离”。经典模式把创建对象、组合结构、协调行为中的变化点抽出来,让系统的稳定部分和变动部分解耦。LLM应用里变化点特别多:模型版本在变、提示词模板在调、输出的JSON结构在改、同一个功能可能要切换不同供应商。这些变化如果没有被设计模式框住,就会散落在业务代码里,改一行牵一发动全身。
我这套思路参考了经典23种设计模式,但不是为用模式而用模式。筛选标准很朴素:第一,近半年真实项目里高频使用;第二,踩过坑、付出过代价;第三,效果能在代码评审时明显讲出来。按这个标准,最终留下六个模式,分布在三个大类里。
1.2 从经典23种模式到LLM场景的筛选逻辑
传统设计模式的三分法依然有效:创建型管“对象怎么来”,结构型管“类和对象怎么组合”,行为型管“交互和职责怎么分配”。映射到生成式AI应用,对应关系非常精确。
创建型模式里我重点用Builder,也就是上下文构建器。LLM应用的“对象”就是送进模型的Prompt,它由系统提示、用户输入、知识库片段、历史对话、工具描述等多段内容装配而成。用new一个字一个字拼,恰恰是创建型模式最反对的做法。
结构型模式里重点用Decorator和Facade。模型输出往往是“裸响应”,需要经过过滤、校验、格式化、缓存才能进入业务层。把这些横切关注点挂在主调用的外层,就是装饰器的典型用法。Facade则用来屏蔽底层调用的复杂性——真实场景里LLM调用域包含重试、超时、限流、成本记录,这些不能暴露给业务方。
行为型模式里重点用Strategy和Adapter。同一个语义任务可以用不同模型、不同温度参数、不同提示策略来完成,这是策略模式的标准舞台。Adapter则用来弥合不同模型供应商之间的接口差异。除此之外还有个隐藏角色——状态机,它在多轮对话里负责会话状态的迁移,后面的综合示例里会展示。
这套筛选逻辑有一个显著收益:代码评审时,我们讨论的不再是“这段为什么这么写”,而是“这是哪种模式、它的边界在哪里、变化点在哪个方向上”。讨论效率提升非常明显。
2. 核心实操:创建型与结构型模式
2.1 上下文构建器:把Prompt装配工程化
先看反面案例。这是我在一个真实项目的旧代码里随手摘的:
def generate_summary(user_input, user_name, chat_history, kb_chunks): prompt = f""" 你现在是一个专业的客服助手。 用户姓名是{user_name}。 以下是用户输入:{user_input} 以下是历史对话:{chat_history} 以下是知识库内容:{kb_chunks} 请根据以上内容生成工单摘要。 """ response = call_llm(prompt) return response这段代码几个硬伤:knowledge base可以为空,但模板里仍然会出现“以下是知识库内容:”这种空标签,模型容易被干扰;chat_history和kb_chunks的长度没有上限,Token预算完全失控;系统提示词改一个标点,业务调用方全部受影响。
Builder模式改造后,上下文装配集中在一个类里:
class ContextBuilder: def __init__(self, token_budget: int = 3000): self._parts = [] self._budget = token_budget def add_system_prompt(self, text: str) -> "ContextBuilder": self._parts.append(("system", text)) return self def add_user_input(self, text: str, max_len: int = 500) -> "ContextBuilder": trimmed = text[:max_len] self._parts.append(("user", f"用户输入:{trimmed}")) return self def add_knowledge(self, chunks: list[str], max_chars: int = 800) -> "ContextBuilder": if not chunks: return self merged = self._smart_merge(chunks, max_chars) self._parts.append(("context", f"参考知识:\n{merged}")) return self def _smart_merge(self, chunks: list[str], limit: int) -> str: # 按相关性分数降序,累计限制长度,超出部分主动丢弃并记录截断标记 result = [] used = 0 for chunk in chunks: # 假设已按相关性排序 if used + len(chunk) > limit: result.append("...[已截断]") break result.append(chunk) used += len(chunk) return "\n".join(result) def build(self) -> list[dict]: return [ {"role": "system" if t == "system" else "user", "content": c} for t, c in self._parts ]所有部分都是条件添加、可叠加、可设置独立预算,最后统一输出成对话消息结构。这样业务代码只需要按业务规则调用add_xxx,不需要关心底层的字符串装配细节。
这里有个很重要的设计决策:为什么不用f-string模板直接拼,而是用Builder逐段添加?因为真实场景中每个“上下文片段”是否出现,取决于业务状态。知识库只有在召回到相关内容时才需要出现;历史对话在第几轮决定截断策略;工具调用说明只在特定Agent节点才拼入。这些“条件装配”用模板语言写起来会很啰嗦,但从Builder的方法链中读出来非常清晰。
2.2 装饰器管道:让模型输出在管线里净化
模型返回的原始字符串,直接塞进业务逻辑,十有八九出事。输出里可能夹着思考过程中的碎语、JSON格式时好时坏、个别词触犯内容安全规则、同样的请求重复消耗成本。这些问题的最佳解决位置,不是业务层,而是调用外层的一个装饰器管道。
我用Python来实现一个精简版:
def output_pipeline(*decorators): def deco(func): def wrapper(*args, **kwargs): result = func(*args, **kwargs) for fn in reversed(decorators): result = fn(result) return result return wrapper return deco def validate_json(raw: str) -> dict: # 抽取第一个 { 到最后一个 } 之间的内容,避免前后杂质 start = raw.find("{") end = raw.rfind("}") if start == -1 or end == -1: raise ValueError("输出中没有找到JSON结构") return json.loads(raw[start:end + 1]) def check_sensitive(raw: str) -> str: # 命中敏感词表则按规则替换,不放行 for word, repl in SENSITIVE_MAP.items(): raw = raw.replace(word, repl) return raw def semantic_cache(raw: str) -> str: # 嵌入相似度缓存,命中则直接返回缓存结果 key = embed(raw) if key in cache: return cache[key] return raw装饰器管道解决了四个典型问题:第一个是输出格式不稳定,用validate_json把“模型最爱在JSON前后加废话”的毛病解决掉;第二个是合规风险,用check_sensitive保证业务红线不破;第三个是成本浪费,用semantic_cache让重复请求不再二次计费;第四个是质量监控,管道里还可以挂日志装饰器,记录每次调用的输入输出哈希、耗时、模型版本。
使用装饰器后,业务代码长这样:
class TicketService: @output_pipeline(check_sensitive, validate_json, semantic_cache) def classify(self, context: list[dict]) -> dict: response = self.llm.chat(context) return response业务方法只关心“我调用模型拿输出”,至于输出之前被净化过什么、缓存过什么、校验过什么,完全不用关心。更重要的是,任何新需求,比如“增加一个字数超限截断”,只需要在管道里加一个装饰器函数,不需要改动现有业务方法。
3. 核心实操:行为型与架构型模式
3.1 策略模式:模型路由与参数调度的开关
实际开发里,一个功能不太可能永远使用同一个模型和同一套参数。我做过一个工单分类系统,简单工单用快速廉价模型足以应付,复杂工单必须换高精度模型,而且每次换模型,系统提示词的措辞、temperature参数甚至解析方式都不一样。
策略模式把“决策”和“实现”拆开。先定义一个通用策略接口:
class LLMStrategy(ABC): @abstractmethod def complete(self, context: list[dict]) -> str: pass @property @abstractmethod def cost_indicator(self) -> float: pass class FastModelStrategy(LLMStrategy): def __init__(self): self.model = "fast-model-v1" self.temperature = 0.3 def complete(self, context): return call_llm(self.model, context, temperature=self.temperature) class PreciseModelStrategy(LLMStrategy): def __init__(self): self.model = "precise-model-v2" self.temperature = 0.1 def complete(self, context): # 高精度模型需要额外注入推理引导 guided = context + [{"role": "system", "content": "请分步骤思考,再输出JSON"}] return call_llm(self.model, guided, temperature=self.temperature)然后需要一个选择器,通常结合业务规则和模型能力做路由:
class StrategyRouter: def __init__(self): self._registry = { "faq": FastModelStrategy(), "complex_ticket": PreciseModelStrategy(), "escalated_review": PreciseModelStrategy(), } def select(self, ticket_type: str) -> LLMStrategy: # 越权词命中时强制走精密模型 if any(word in ticket_type for word in ESCALATION_WORDS): return self._registry["escalated_review"] return self._registry.get(ticket_type, self._registry["faq"])这样做的好处是,模型选型从业务代码中彻底剥离。业务层只需传递一个“场景标签”,选哪个模型、用什么参数,全部封闭在策略族内部。后来换了一次底层模型,只改了一个类的构造函数,测试没动一行。
3.2 门面模式:把混乱的模型调用收敛成一个接口
没有门面之前,业务层要处理的东西大概是这样的:构造HTTP请求、设置超时、处理429限流重试、记录成本、捕获异常、统一日志格式。这些事在六七个业务方法里各写一遍之后,你一定分不清到底有哪些超时设置不一致。
门面模式在这里的意义,不是“封装网络调用”这么简单,而是提供一个业务侧真正愿意使用的稳定接口。业务方只认两个方法:chat和chat_batch。
class LLMFacade: def __init__(self, strategy: LLMStrategy, config: dict): self._strategy = strategy self._timeout = config["timeout_seconds"] self._max_retries = config["max_retries"] self._cost_tracker = CostTracker() def chat(self, context: list[dict]) -> str: for attempt in range(self._max_retries): try: result = self._strategy.complete(context) self._cost_tracker.record(self._strategy.cost_indicator) return result except RateLimitError: sleep(backoff(attempt)) except ConnectionError: continue raise LLMServiceUnavailable("多次重试后仍然失败") def chat_batch(self, contexts: list[list[dict]]) -> list[str]: return [self.chat(c) for c in contexts]门面把重试策略、超时管理、成本跟踪等横切关注点全部内聚。业务模块从此不用知道底层用的是HTTP还是本地推理,也不需要在业务层捕异常。还有一个好处是测试容易了——单元测试时只需要mock门面,所有业务逻辑和模型无关。
3.3 适配器模式:多供应商切换的兼容层
大多数团队不会只绑定一家模型供应商。开源的、闭源的、不同版本的模型,可能在同一个系统里共存。每家的请求格式不同、返回结构不同,如果直接在业务里写两种格式的判断逻辑,代码就是灾难。
Adapter模式在LLM场景中的作用,是把不同供应商的API统一成内部标准响应结构:
class ProviderAdapter(ABC): @abstractmethod def to_internal(self, raw_response) -> InternalResponse: pass class OpenAIImpl(ProviderAdapter): def to_internal(self, raw_response): return InternalResponse( text=raw_response["choices"][0]["message"]["content"], tokens=raw_response["usage"]["total_tokens"], model=raw_response["model"], raw=raw_response, ) class LocalModelImpl(ProviderAdapter): def to_internal(self, raw_response): # 本地模型返回格式可能完全是另一套 return InternalResponse( text=raw_response["output"], tokens=raw_response["meta"]["tokens"], model=raw_response["meta"]["model_name"], raw=raw_response, )一旦所有响应都被包成InternalResponse,上层代码使用的字段就是完全统一的。后来我们在不告知业务层的情况下替换了某一个供应商的底层实现,适配器层改完,业务层零改动。
4. 综合实战:一个客服工单分类服务的完整落地
4.1 需求描述与类职责划分
为了验证上面六个模式能共存,我搭了一个实际的客服工单分类服务。需求长这样:系统接收用户提交的工单文本,需要输出工单类别、紧急程度、对应的处理建议摘要。工单分为简单咨询、技术故障、投诉、商务合作四类;紧急程度分高中低三档。整套流程要求:优先控制成本,简单问题走快速模型;紧急投诉必须走高精度模型且需要复核标记;所有输出必须是合法JSON。
按职责划分成六个类,每个类对应一个清晰角色:
| 类名 | 对应模式 | 职责边界 |
|---|---|---|
| ContextBuilder | Builder | 装配系统提示、用户工单、历史备注、规则片段 |
| LLMStrategy接口 | Strategy | 定义模型交互的共同调用契约 |
| FastModelStrategy / PreciseModelStrategy | Strategy实现 | 根据工单类型选择不同模型与参数 |
| LLMFacade | Facade | 统一入口,内置重试、超时、成本统计 |
| 输出管道装饰器 | Decorator | 过滤、JSON校验、缓存处理 |
| ProviderAdapter | Adapter | 屏蔽不同模型供应商的响应格式差异 |
类之间的依赖方向只有一个:门面依赖策略,策略依赖适配器,业务主流程只依赖门面和上下文构建器。
4.2 核心代码与运行流程
主流程的核心代码大约是这个样子:
class TicketClassifier: def __init__(self, router: StrategyRouter, facade: LLMFacade, budget: int): self._router = router self._facade = facade self._budget = budget def classify(self, ticket: dict) -> dict: # 第一步:路由决定用哪个策略 strategy = self._router.select(ticket["type_hint"]) # 第二步:门面切换策略 self._facade.set_strategy(strategy) # 第三步:Builder装配上下文 context = ( ContextBuilder(self._budget) .add_system_prompt(SYSTEM_PROMPT_TEMPLATE) .add_user_input(ticket["text"]) .add_knowledge(ticket.get("kb_chunks", [])) .build() ) # 第四步:调用门面,装饰器自动处理输出 result = self._facade.chat(context) # 第五步:解析为业务结构 return self._validate_result(result)运行顺序中有一个容易被忽略的关键点:策略路由必须在构建上下文之前完成。原因是不同策略对上下文的组织方式有不同要求,高精度模式可能需要额外的“分步思考”引导,快速模式可能需要更精简的指令,所以Builder也需要一个策略感知的参数。我把这个参数加到了Builder的构造函数中,用模式标识来决定是否追加推理引导片段。
另一个细节是装饰器的应用位置。装饰器挂在门面的chat方法上,而不是业务类的classify方法上。这样任何通过门面出去的调用都天然经过过滤和校验,而不是某个业务模块漏掉了校验。
4.3 实测效果与关键参数选择
用真实生产数据回放一遍,效果比想象中稳。主要参数如下:上下文预算是2500 Token,其中系统提示词约400,用户工单最多截断到500字,知识库片段最多融合800字,历史对话则根据轮数动态分配剩余预算。
成本上节约非常明显。简单咨询类工单约占总量的60%,全程走了快速模型,单次调用成本是精密模型的四分之一。而紧急投诉类工单占比不到10%,虽然走了高精度模型,但因为数量少,总体成本依然可控。整体算下来,相比“所有工单统一用高精度模型”的方案,成本下降约45%。
延迟上也有取舍。快速模型的P95延迟约2.1秒,精密模型约5.8秒。为了不让用户等待时间过长,我把精密模型的使用场景进一步细分:紧急程度为高的工单才允许触发精密模型,中等紧急度一律走快速模型加二次复核。复核逻辑复用同一个服务但变更策略参数,只是多一次迭代调用。这套参数组合稳定运行了三周,分类准确率稳定在92%以上,格式错误率为压测以来最低。
5. 常见问题与排查技巧实录
5.1 高频故障速查表
在落地这段代码的过程中,我积累了一张问题排查表,分享出来可以直接参考:
| 问题症状 | 大概率原因 | 排查与解法 |
|---|---|---|
| 上下文被模型忽略,输出偏离业务要求 | Prompt过长,关键指令被海量知识片段淹没 | 在Builder中增加“关键指令前置”规则;把系统提示始终放在所有上下文的头部且不后移 |
| JSON解析偶尔报错,且多发于复杂工单 | 模型在JSON前后输出解释性文字,或JSON内部字段顺序不稳定 | 装饰器内先定位首尾大括号再解析;嵌套引号导致转义错误时可改用非贪婪正则解析 |
| 同一个判断逻辑换模型后结果变化大 | 回归测试缺位,只验证了原模型行为 | 建立黄金测试集,每条包含输入、期望输出字段、必须遵守的边界条件,切换策略前必须全量回归 |
| 延迟抖动严重,个别请求耗时暴增 | 知识库片段未按相关性截断,全部塞给模型 | 在Builder的smart_merge中限长,并记录截断原因到日志 |
| 单次对话累计成本高 | 缓存粒度太粗,只缓存最终输出 | 改成在装饰器管道外层做语义缓存,命中后还能返回完整的业务结构 |
| 部署后发现业务代码依赖了供应商专有字段 | 适配器层没有彻底屏蔽内部字段 | 规定业务代码只能读取InternalResponse,禁止访问raw属性;代码评审时专项检查 |
5.2 独家避坑经验
第一个深刻的教训是关于缓存键的。装饰器管道里做语义缓存,为了简单第一版直接用文本哈希,结果发现用户输入有微小差异时缓存完全失效。后来改成嵌入向量相似度缓存,缓存命中率从13%提升到41%。但必须设置相似度阈值,不能盲目命中,我用的阈值是0.86,低于它宁可重新推理也不返回旧答案。
第二个教训是关于超时重试的归属。第一版把重试逻辑写在策略类里,结果两个策略各有不同的重试次数和退避算法,排查超时问题时需要分头看。后来统一收敛到门面层,策略层只负责“怎么调模型”,门面层统一负责“调用失败怎么办”。这个改动后,超时相关的线上问题处理速度提升明显。
第三个经验是关于上下文预算的分散式管理。如果Builder只有一个总预算,加一个字段就要重算一遍,很麻烦。我给每个add方法设了独立的默认上限,再在build时汇总并给出超预算警告。这样代码里每次添加内容,都自带本段预算上限,不至于某些片段野蛮生长挤占其他片段的Token。
第四个经验是关于装饰器顺序。管道装饰器顺序不是随便排的——我先做敏感词替换,再做JSON校验,最后做缓存。原因是敏感词替换会修改文本内容,如果先做了缓存,可能缓存中存的就是未过滤的脏文本。JSON校验必须在格式清洗之后,否则无法定位真正的格式错误。缓存放到管道末尾,让过滤器最晚介入,以便缓存内容就是最终清洗后的结果。管道顺序混乱,一度让我在日志里看到了泄漏的敏感词,调整顺序之后问题消失。
最后再分享一个我最近在琢磨的扩展方向:把上述模式应用到多个Agent协作的场景。每个Agent节点不再各自为政,而是共享同一个门面和同一个上下文构建器,只是通过策略模式注入不同的System Prompt和能力标识。这样Agent编排层就可以成为标准的业务层,复用这张模式网带来的稳定性和可控性。这些模式本身的代码量不大,但带来的长期收益是巨大的。如果你也在搭类似的服务,我的建议是:先把需要变化的地方列出来,再去找对应的模式,而不是反过来为了用模式而用模式。