Prompt 熔断策略：当模型持续输出异常时自动降级

一、个性化深度引言

凌晨 1:30，智能客服的满意度评分从 4.2 断崖式跌到 1.8。我打开对话日志，发现模型在给所有用户输出同样的一句话："请稍后再试，我们的服务暂时不可用。"——这是模型自己编造的。上游的 Prompt 微调脚本意外地将 System Prompt 末尾的max_tokens截断了一半，导致模型在生成不完整回复后陷入循环。

这个 bug 上线了 40 分钟才被发现。期间有 3800 个用户会话被影响。如果有 Prompt 输出的实时异常检测和自动熔断机制，能在第 1 分钟就把这个版本拦下来。

二、个性化原理剖析

Prompt 熔断（Circuit Breaker）借鉴了微服务架构中的断路器模式。核心思路是：实时监控模型输出的质量指标，当指标持续异常时，自动切换到安全的兜底策略。

具体状态流转逻辑如下：系统初始处于正常运行（CLOSED）状态，模型正常输出并经过质量检测；当异常率超过阈值（如 10%/30s）时，状态切换至熔断开启（OPEN），此时直接返回静态兜底回复。经过冷却时间（如 60s）后，系统进入半开探测（HALF_OPEN）状态进行探测请求，若连续 5 次正常则恢复 CLOSED，否则重新回到 OPEN。

熔断触发条件需要多维检测。单一指标容易误报或漏报。见证奇迹的时刻在于这四个指标的组合效应——单一指标触发时误报率约 15%，但四指标 OR 逻辑可将漏报率降到接近 0%，代价是误报率升至约 8%。对于线上服务，这个 trade-off 是值得的。

我们定义了四个熔断指标：

• 重复率异常：最近 100 条回复中，完全相同的内容超过 30%
• 长度异常：平均回复长度低于正常值的 40%
• 拒绝率异常：包含"无法""不能""抱歉"等拒绝词的回复超过 50%
• 脱靶率：回复内容与意图分类完全不匹配的比例超过 20%

四个指标采用"OR"逻辑——任一触发即熔断。这虽然增加了误熔断概率，但线上场景宁可误判也不漏判。

三、个性化代码实践

import time import threading from collections import deque from enum import Enum from dataclasses import dataclass, field class CircuitState(Enum): CLOSED = "closed" # 正常 OPEN = "open" # 熔断 HALF_OPEN = "half_open" # 半开（探测恢复） @dataclass class OutputMetrics: """模型输出质量指标窗口""" window: deque = field(default_factory=lambda: deque(maxlen=100)) def add(self, text: str, intent: str): self.window.append({"text": text, "intent": intent, "ts": time.time()}) def anomaly_score(self) -> dict: """计算异常分数""" if len(self.window) < 50: return {"level": "normal"} texts = [w["text"] for w in self.window] # 设计原因：检测重复率——相同回复超过30%说明模型卡死 from collections import Counter text_counts = Counter(texts) max_dup_ratio = max(text_counts.values()) / len(texts) # 设计原因：检测长度异常——平均长度低于正常40%是截断或循环 # 80字符是客服回复的最短合理长度 avg_len = np.mean([len(t) for t in texts]) # 设计原因：检测拒绝率——高频拒绝词说明模型在推诿 reject_words = {"无法", "不能", "抱歉", "无法提供", "不支持"} reject_count = sum( 1 for t in texts if any(w in t for w in reject_words) ) reject_ratio = reject_count / len(texts) score = 0 triggers = [] if max_dup_ratio > 0.3: score += 40 triggers.append(f"重复率={max_dup_ratio:.1%}") if avg_len < 32: # 正常40% = 80*0.4 score += 30 triggers.append(f"平均长度={avg_len:.0f}") if reject_ratio > 0.5: score += 30 triggers.append(f"拒绝率={reject_ratio:.1%}") return { "score": score, "level": "normal" if score < 60 else "critical", "triggers": triggers, } class PromptCircuitBreaker: """Prompt 输出熔断器""" def __init__(self, llm_service, fallback_templates): self.llm = llm_service self.fallback = fallback_templates # 兜底回复模板 self.state = CircuitState.CLOSED self.metrics = OutputMetrics() self.lock = threading.Lock() # 设计原因：30s 检查窗口，平衡检测灵敏度与误报率 self.check_interval = 30 self.failure_threshold = 60 # 异常分数阈值 self.cooldown_period = 60 # 熔断冷却时间（秒） self.half_open_probe = 5 # 半开探测请求数 self.last_failure_time = 0 self.probe_count = 0 self.probe_success = 0 # 设计原因：后台线程定时检查指标 self.monitor = threading.Thread(target=self._monitor_loop, daemon=True) self.monitor.start() def generate(self, prompt: str, intent: str) -> str: """带熔断保护的生成接口""" with self.lock: if self.state == CircuitState.OPEN: # 设计原因：熔断状态下直接返回兜底模板 # 用意图分类选择最匹配的模板 return self.fallback.get_template(intent) if self.state == CircuitState.HALF_OPEN: # 设计原因：半开状态只放行探测请求 if self.probe_count >= self.half_open_probe: return self.fallback.get_template(intent) self.probe_count += 1 # 正常/半开探测：调用真实模型 try: response = self.llm.generate(prompt) self.metrics.add(response, intent) if self.state == CircuitState.HALF_OPEN: self.probe_success += 1 return response except Exception: # 设计原因：任何异常都记录，但不立即熔断 # 单次异常可能是网络抖动 self.metrics.add("[ERROR]", intent) return self.fallback.get_template(intent) def _monitor_loop(self): """监控循环：检测异常并触发熔断""" while True: time.sleep(self.check_interval) anomaly = self.metrics.anomaly_score() with self.lock: if self.state == CircuitState.CLOSED: if anomaly["score"] >= self.failure_threshold: # 设计原因：触发熔断，记录时间用于冷却 self.state = CircuitState.OPEN self.last_failure_time = time.time() self._alert(f"Prompt熔断触发: {anomaly['triggers']}") elif self.state == CircuitState.OPEN: # 设计原因：冷却时间到后进入半开状态探测 if time.time() - self.last_failure_time > self.cooldown_period: self.state = CircuitState.HALF_OPEN self.probe_count = 0 self.probe_success = 0 elif self.state == CircuitState.HALF_OPEN: # 设计原因：探测请求全部成功则恢复 if self.probe_count >= self.half_open_probe: if self.probe_success == self.half_open_probe: self.state = CircuitState.CLOSED self._alert("Prompt熔断恢复") else: # 探测失败，重新熔断 self.state = CircuitState.OPEN self.last_failure_time = time.time()

四、个性化边界权衡

误熔断的代价：四个指标的OR逻辑虽然不漏判，但可能误熔断。例如某个用户群体确实都在问同一个问题，导致重复率升高。解决方案是为企业级客户设置白名单，白名单内不触发重复率检测。

兜底模板的质量：熔断后用户看到的是预定义的静态回复而非模型生成。为了让兜底模板尽可能个性化，我们按意图分类准备了50+个兜底模板，覆盖所有高频场景。低频场景统一使用善意引导模板（"您的问题已记录，人工客服稍后联系您"）。

检测窗口大小的权衡：100条窗口在QPS=100时代表1秒的数据，在QPS=1时代表100秒。不同QPS下同样的窗口大小含义不同。更精确的做法是用时间窗口（如30秒）而非数量窗口，但实现复杂度更高。

级联熔断的风险：如果Prompt熔断 + GPU过载 + 数据库超时同时发生，系统可能完全不可用。需要全局熔断协调器，避免多个熔断器同时打开导致完全降级。

五、总结

Prompt熔断策略通过四维异常指标（重复率、长度、拒绝率、脱靶率）实时监控模型输出质量，异常时自动切换到兜底模板。三个状态（CLOSED/OPEN/HALF_OPEN）遵循标准断路器模式，60秒冷却窗口在可用性和安全性之间取平衡。兜底模板需按意图分类预设，误熔断需企业白名单缓解。