6.3 救场与应变：鲁棒性与复杂场景

6.3 救场与应变：鲁棒性与复杂场景

本节摘要：线上双智能体系统的三类高发事故——漂移跑偏、死循环、预算超支——各自有自动处置剧本：熔断、检测、守卫。本节给出三个剧本的最小实现，再讨论复杂场景的降级策略：群戏不行退对手戏，对手戏不行退人工。6.2 节备了接口，本节备应急预案。

事故清单先行

先承认一个事实：双智能体系统上线后，出事的概率不取决于模型多好，取决于编排的闸门有多硬。全册出现过的所有失效模式里，能在生产环境造成真实损失的集中在三个：漂移跑偏（产出偏离任务，白烧调用）、死循环（两侧互礼循环，账单失控）、预算超支（单场戏吃掉整批预算）。三类的共同点：都不需要换模型，都需要硬闸门。本节逐个给最小处置剧本，全部可以叠加在 3.2 节的演出循环上。

剧本一：漂移熔断

漂移的在线处置与离线巡检不同——离线可以事后复盘，在线必须当场熔断：检测到漂移特征立即暂停对话，触发定向修正。实现要点是把 2.1 节的巡检正则搬进循环体，命中后不终止演出，而是插入一条系统级修正消息：

import re

DRIFT_PAT = re.compile(r"(非常好|太专业|好问题|需要我帮你|那我替你|顺便也)")

def drift_fuse(text: str, drift_count: dict) -> bool:
    """在线漂移熔断：命中返回 True，并由调用方插入修正消息。"""
    if DRIFT_PAT.search(text):
        drift_count["n"] += 1
        return drift_count["n"] >= 2      # 连续两次命中才熔断，单次可能是误报
    return False

drift_count = {"n": 0}
sample = "这个问题提得非常好，我们顺便也讨论一下风控吧"
if drift_fuse(sample, drift_count):
    print("触发修正：请回到总任务的当前幕次，不评价对方，不加新话题")
​

触发修正：请回到总任务的当前幕次，不评价对方，不加新话题
​

两个设计决定说明一下。连续两次才熔断：单次命中常是误报（正常文本偶尔带触发词），双次命中率大大降低，代价是慢半轮——可接受。熔断不等于停演：修正消息让戏回到轨道，只有修正无效才升级到终止。熔断次数本身要记账，一场戏熔断两次以上就该标记为病场，进 6.1 看板的漂移率分子。

剧本二：死循环检测

死循环与漂移不同：它不是立场病，是重复病——两侧交换着说高度相似的话，形式合规、内容归零。检测手段是内容指纹：对每轮文本取哈希，滑窗内重复率超阈值即判死循环：

import hashlib
import collections

def loop_detector(window: int = 3, threshold: float = 0.6):
    """滑窗重复检测：最近 window 轮内相似内容占比超阈值即报警。"""
    seen = collections.deque(maxlen=window)
    def check(text: str) -> bool:
        # 归一化后取哈希：长度相近且词面相近的轮次会撞车
        key = hashlib.md5("".join(text.split())[:200].encode()).hexdigest()
        dup = sum(1 for k in seen if k == key)
        seen.append(key)
        return dup / window >= threshold
    return check

checker = loop_detector()
rounds = ["第5轮方案：实现均值函数如下", "第5轮方案：实现均值函数如下",
          "第5轮方案：实现均值函数如下"]
for t in rounds:
    if checker(t):
        print("检测到死循环：终止本场，标记病场")
        break
​

检测到死循环：终止本场，标记病场
​

哈希指纹只抓"一字不差"的重复，误放走换皮循环——工程上升级为相邻轮的编辑距离或嵌入相似度。但注意处置优先级：简易版先上线，精确版按需升级。死循环的损失是线性的（每轮固定成本），简易版抓大头已经止损大半。

剧本三：预算守卫

预算守卫在 3.3 节已经给了闭包实现，这里补它的分层用法：单场预算之上，还要有批次预算与日预算，三层守卫层层兜底。因为事故往往不是单场超支，而是某批任务卡集体偏大：

def layered_guard(show_budget: int, batch_budget: int, day_budget: int):
    """三层预算守卫：单场 批次 日累计。"""
    state = {"batch": 0, "day": 0}
    def charge(tokens: int) -> str:
        state["batch"] += tokens
        state["day"] += tokens
        if state["day"] > day_budget:
            return "日预算触顶：全产线停机，人工介入"
        if state["batch"] > batch_budget:
            return "批次触顶：本批停止，剩余任务顺延"
        return "ok"
    def reset_batch():
        state["batch"] = 0
    return charge, reset_batch

charge, reset_batch = layered_guard(show_budget=20000,
                                    batch_budget=150000, day_budget=400000)
print(charge(60000))      # 第一场
print(charge(60000))      # 第二场
​

ok
ok
​

三层守卫的处置动作不同：单场触顶只停本场，批次触顶停批顺延，日触顶全停等人。触顶动作越重，阈值要越保守——日预算的阈值应设在"就算全部烧光也不影响业务"的水平。

复杂场景的降级阶梯

救场手册的最后一页写降级策略。线上复杂场景出问题时，按阶梯退，不要硬扛：

FALLBACK_LADDER = [
    {"level": 1, "setup": "多幕对手戏加评论家", "trigger": "默认配置"},
    {"level": 2, "setup": "单幕对手戏去掉加戏", "trigger": "幕内反复退回或熔断频繁"},
    {"level": 3, "setup": "单智能体加模板提示", "trigger": "对手戏仍不稳定"},
    {"level": 4, "setup": "人工处理并入队列", "trigger": "价值高且自动化失败"},
]
for step in FALLBACK_LADDER:
    print(f"L{step['level']} {step['setup']} —— 触发：{step['trigger']}")
​

L1 多幕对手戏加评论家 —— 触发：默认配置
L2 单幕对手戏去掉加戏 —— 触发：幕内反复退回或熔断频繁
L3 单智能体加模板提示 —— 触发：对手戏仍不稳定
L4 人工处理并入队列 —— 触发：价值高且自动化失败
​

降级阶梯的设计哲学是 4.4 节扩编判据的镜像：扩编要克制，降级要果断。很多团队的通病是在 L1 硬扛——任务明显不适合却反复加戏救场，结果每次救场成本都比 L3 的一次调用高一个量级。判断该不该降级的信号很具体：同一任务连续三场进病场，立即降一级，不做第二轮回拉扯。

应急手册备齐。全册最后一站，走出工作台：看看剧场外的生态，同好、答案与贡献都在哪里。