大模型业务稳定性：拆解路由重试、校验降级与复核机制，解决业务不确定性异常难题23.8

from dataclasses import dataclass
from typing import Optional, Dict

# 模型元数据配置
@dataclass
class ModelMeta:
    model_name: str
    endpoint: str
    cost_level: str  # high / medium / low
    support_task: set[str]
    error_rate: float = 0.0  # 运行时统计错误率

class ModelRouter:
    def __init__(self):
        self.model_pool: Dict[str, ModelMeta] = {
            "gpt-main": ModelMeta(
                model_name="gpt-main",
                endpoint="https://xxx/v1/chat/completions",
                cost_level="high",  # 高性能大模型，推理成本高，适合复杂推理/风控等高价值任务
                support_task={"risk_judge", "complex_reason"}
            ),
            "qwen-light": ModelMeta(
                model_name="qwen-light",
                endpoint="https://yyy/v1/chat/completions",
                cost_level="low",  # 轻量模型，推理成本低，适合抽取/改写/总结等高并发通用任务
                support_task={"extract", "rewrite", "summary"}
            )
        }
        # 故障阈值，错误率超过该值则不选中该模型
        self.error_threshold = 0.15

    def select_model(self, task_type: str) -> Optional[ModelMeta]:
        """根据任务类型+健康状态选择模型"""
        candidates = [
            m for m in self.model_pool.values()
            if task_type in m.support_task and m.error_rate < self.error_threshold
        ]
        if not candidates:
            return None
        # 业务可扩展：增加权重、成本策略，这里优先返回第一个可用候选
        return candidates[0]

# 使用示例
if __name__ == "__main__":
    router = ModelRouter()

    # 成本等级简短理由映射
    cost_reason = {
        "high": "（高性能大模型，推理成本高）",
        "medium": "（中等模型，性价比均衡）",
        "low": "（轻量模型，推理成本低）",
    }

    # 场景1：高风险判断任务 → gpt-main
    print("=== 场景1：风险判断任务(risk_judge) ===")
    selected = router.select_model(task_type="risk_judge")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑")

    # 场景2：复杂推理任务 → gpt-main
    print("\n=== 场景2：复杂推理任务(complex_reason) ===")
    selected = router.select_model(task_type="complex_reason")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑")

    # 场景3：信息提取任务 → qwen-light
    print("\n=== 场景3：信息提取任务(extract) ===")
    selected = router.select_model(task_type="extract")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑")

    # 场景4：文本改写任务 → qwen-light
    print("\n=== 场景4：文本改写任务(rewrite) ===")
    selected = router.select_model(task_type="rewrite")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑")

    # 场景5：内容总结任务 → qwen-light
    print("\n=== 场景5：内容总结任务(summary) ===")
    selected = router.select_model(task_type="summary")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑")

    # 场景6：不支持的任务类型 → 降级
    print("\n=== 场景6：不支持的翻译任务(translate) ===")
    selected = router.select_model(task_type="translate")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑")

    # 场景7：模型错误率超阈值 → 被排除
    print("\n=== 场景7：gpt-main 错误率超阈值被排除 ===")
    router.model_pool["gpt-main"].error_rate = 0.3
    selected = router.select_model(task_type="risk_judge")
    if selected:
        print(f"路由选中模型：{selected.model_name}, 成本等级：{selected.cost_level}{cost_reason[selected.cost_level]}")
    else:
        print("无可用模型，触发降级逻辑（gpt-main 因错误率超阈值被跳过，且无其他模型支持该任务）")

import time
from tenacity import retry, stop_after_attempt, wait_exponential, wait_fixed, retry_if_exception_type, before_log, after_log
import logging

# 配置日志输出
logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(message)s", datefmt="%H:%M:%S")
logger = logging.getLogger(__name__)

class ModelServiceUnavailable(Exception):
    """服务临时不可用，可以重试"""
    pass

class ModelParamInvalid(Exception):
    """参数错误，不可重试"""
    pass

# 模拟调用计数器（用于中途恢复的场景）
_call_count = 0

def call_llm_api(mock_error_type: str):
    """模拟调用大模型API"""
    global _call_count
    _call_count += 1
    logger.info(f"  [API调用] 第{_call_count}次发起请求，模拟错误类型: {mock_error_type}")

    if mock_error_type == "rate_limit":
        logger.warning(f"  [API响应] 429 请求限流 → 抛出 ModelServiceUnavailable（可重试）")
        raise ModelServiceUnavailable("429 请求限流")

    elif mock_error_type == "rate_limit_then_ok":
        # 前2次限流，第3次成功
        if _call_count <= 2:
            logger.warning(f"  [API响应] 429 请求限流 → 抛出 ModelServiceUnavailable（可重试）")
            raise ModelServiceUnavailable("429 请求限流")
        else:
            logger.info(f"  [API响应] 200 OK → 返回结果")

    elif mock_error_type == "invalid_param":
        logger.error(f"  [API响应] 400 参数非法 → 抛出 ModelParamInvalid（不可重试）")
        raise ModelParamInvalid("prompt参数非法")

    return "大模型返回结果内容"

# 只对可重试异常执行重试：指数退避 1s→2s→4s，最多3次
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=4),
    retry=retry_if_exception_type((ModelServiceUnavailable,)),
    before=before_log(logger, logging.INFO),
    after=after_log(logger, logging.INFO),
    reraise=True
)
def safe_call_llm(error_type: str):
    return call_llm_api(error_type)

if __name__ == "__main__":
    # =========================================================
    # 场景1：限流 → 重试3次均失败 → 降级
    # =========================================================
    print("=" * 60)
    print("场景1：限流异常 → 指数退避重试(1s→2s→4s)3次 → 全部失败 → 降级")
    print("=" * 60)
    _call_count = 0
    try:
        start = time.time()
        res = safe_call_llm("rate_limit")
        print(f"\n>>> 调用成功：{res}")
    except ModelServiceUnavailable:
        elapsed = time.time() - start
        print(f"\n>>> [最终结果] 重试3次全部失败（耗时{elapsed:.1f}s），进入降级流程\n")

    # =========================================================
    # 场景2：限流 → 前2次失败 → 第3次成功
    # =========================================================
    print("=" * 60)
    print("场景2：限流异常 → 前2次失败 → 第3次恢复 → 调用成功")
    print("=" * 60)
    _call_count = 0
    try:
        start = time.time()
        res = safe_call_llm("rate_limit_then_ok")
        elapsed = time.time() - start
        print(f"\n>>> [最终结果] 第3次重试成功（耗时{elapsed:.1f}s）：{res}\n")
    except Exception as e:
        print(f"\n>>> [最终结果] 调用失败：{e}\n")

    # =========================================================
    # 场景3：参数错误 → 不重试 → 立即失败
    # =========================================================
    print("=" * 60)
    print("场景3：参数非法异常(ModelParamInvalid) → 不可重试 → 立即抛错")
    print("=" * 60)
    _call_count = 0
    try:
        start = time.time()
        res = safe_call_llm("invalid_param")
        print(f"\n>>> 调用成功：{res}")
    except ModelParamInvalid:
        elapsed = time.time() - start
        print(f"\n>>> [最终结果] 参数非法，禁止重试，直接返回业务错误（耗时{elapsed:.1f}s）\n")

    # =========================================================
    # 场景4：正常调用 → 无异常 → 一次成功
    # =========================================================
    print("=" * 60)
    print("场景4：正常调用 → 无异常 → 一次返回")
    print("=" * 60)
    _call_count = 0
    try:
        start = time.time()
        res = safe_call_llm("normal")
        elapsed = time.time() - start
        print(f"\n>>> [最终结果] 一次调用成功（耗时{elapsed:.1f}s）：{res}\n")
    except Exception as e:
        print(f"\n>>> [最终结果] 调用失败：{e}\n")

    # =========================================================
    # 总结
    # =========================================================
    print("=" * 60)
    print("重试机制总结")
    print("=" * 60)
    print("策略        | 异常类型                | 重试方式       | 结果")
    print("-----------|------------------------|---------------|-------------------")
    print("指数退避    | ModelServiceUnavailable| 1s→2s→4s 3次  | 全部失败则降级")
    print("指数退避+恢复| ModelServiceUnavailable| 1s→2s→成功     | 重试期间服务恢复")
    print("不重试      | ModelParamInvalid      | 立即失败       | 参数错误直接返回")
    print("正常返回    | 无异常                  | 无需重试       | 一次调用成功")

import json
from typing import List, Dict

class LLMOutputValidator:
    def __init__(self, allow_enum: set):
        self.allow_enum = allow_enum

    def check_json_format(self, raw_text: str) -> tuple[bool, dict | None]:
        """校验JSON格式"""
        try:
            data = json.loads(raw_text)
            return True, data
        except json.JSONDecodeError:
            return False, None

    def check_business_enum(self, data: dict) -> bool:
        """业务枚举校验，分类结果必须在允许集合内"""
        label = data.get("risk_label")
        if label not in self.allow_enum:
            return False
        return True

    def detect_conflict(self, model_output_list: List[dict]) -> bool:
        """检测多模型输出是否存在结论冲突"""
        labels = {item.get("risk_label") for item in model_output_list}
        # 同时出现多种不同结论，判定冲突
        if len(labels) > 1:
            return True
        return False

if __name__ == "__main__":
    validator = LLMOutputValidator(allow_enum={"high", "medium", "low"})
    raw = '''{"risk_label":"high"}'''
    format_ok, parse_data = validator.check_json_format(raw)
    if format_ok and parse_data:
        bus_ok = validator.check_business_enum(parse_data)
        print(f"格式校验:{format_ok},业务校验:{bus_ok}")

    # 模拟多模型冲突
    outputs = [{"risk_label":"high"}, {"risk_label":"low"}]
    is_conflict = validator.detect_conflict(outputs)
    print(f"是否存在结果冲突：{is_conflict}")

from enum import Enum

# 降级等级：从轻到重，逐级兜底
class DegradeLevel(Enum):
    NORMAL = "normal"               # 正常，无需降级
    BACKUP_MODEL = "backup_model"   # 一级：主模型不可用 → 切换备用模型
    RULE_FALLBACK = "rule_fallback" # 二级：模型全不可用 → 传统规则兜底
    TEMPLATE_TIP = "template_tip"   # 三级：规则也失败 → 返回固定模板提示
    TO_MANUAL = "to_manual_review"  # 四级：最终兜底 → 转人工复核

class DegradeHandler:
    """降级处理器：逐级降级，保障业务不中断"""
    def __init__(self):
        self.template_map = {
            "tip": "系统暂时无法完成智能分析，请调整输入内容后重试。"
        }

    def rule_fallback_execute(self, input_text: str):
        """传统规则兜底示例"""
        return {"risk_label":"unknown", "reason":"AI链路异常，规则兜底标记未知"}

    def dispatch_degrade(self, degrade_level: DegradeLevel, input_context: dict):
        """根据降级等级分发到对应处理逻辑"""
        if degrade_level == DegradeLevel.BACKUP_MODEL:
            # 一级降级：主模型挂了，切到备用模型继续服务
            return {"degrade":True, "way":"切换备用模型", "context":input_context}
        elif degrade_level == DegradeLevel.RULE_FALLBACK:
            # 二级降级：模型全挂，用规则引擎兜底
            return self.rule_fallback_execute(input_context.get("text",""))
        elif degrade_level == DegradeLevel.TEMPLATE_TIP:
            # 三级降级：规则也跑不了，给用户一个友好提示
            return {"msg": self.template_map["tip"]}
        elif degrade_level == DegradeLevel.TO_MANUAL:
            # 四级降级：自动链路全断，写入人工复核队列
            return {"degrade":True, "way":"送入人工复核队列","task_id":input_context["task_id"]}
        else:
            # 正常路径，无需降级
            return {"status":"ok"}

if __name__ == "__main__":
    handler = DegradeHandler()
    context = {"task_id":"t1001","text":"待分析业务文本"}

    # ---- 逐级演示降级流程 ----
    print("=" * 50)
    print("降级流程演示：逐级兜底，保障业务可用")
    print("=" * 50)

    # 正常：无需降级
    print("\n>>> [正常] 主模型健康，直接返回")
    res = handler.dispatch_degrade(DegradeLevel.NORMAL, context)
    print(f"    结果: {res}")

    # 一级降级：主模型不可用 → 切备用模型
    print("\n>>> [一级降级] 主模型超时 → 切换备用模型")
    res = handler.dispatch_degrade(DegradeLevel.BACKUP_MODEL, context)
    print(f"    结果: {res}")

    # 二级降级：模型全不可用 → 规则兜底
    print("\n>>> [二级降级] 备用模型也失败 → 规则引擎兜底")
    res = handler.dispatch_degrade(DegradeLevel.RULE_FALLBACK, context)
    print(f"    结果: {res}")

    # 三级降级：规则失败 → 模板提示
    print("\n>>> [三级降级] 规则引擎异常 → 返回固定提示")
    res = handler.dispatch_degrade(DegradeLevel.TEMPLATE_TIP, context)
    print(f"    结果: {res}")

    # 四级降级：自动链路全断 → 转人工
    print("\n>>> [四级降级] 自动链路全断 → 转人工复核")
    res = handler.dispatch_degrade(DegradeLevel.TO_MANUAL, context)
    print(f"    结果: {res}")

    print("\n" + "=" * 50)
    print("降级链路：模型→备用→规则→模板→人工，逐级兜底不中断")

from dataclasses import dataclass
from typing import List, Optional

@dataclass
class ReviewTask:
    task_id: str
    input_content: str
    model_outputs: List[dict]
    reason: str  # 为什么进入复核：冲突/校验失败
    review_result: Optional[dict] = None
    review_operator: Optional[str] = None

class ManualReviewQueue:
    def __init__(self):
        self.queue: List[ReviewTask] = []

    def push_task(self, task: ReviewTask):
        self.queue.append(task)

    def pop_pending_task(self) -> Optional[ReviewTask]:
        if len(self.queue) == 0:
            return None
        return self.queue.pop(0)

    def submit_review_result(self, task: ReviewTask, operator: str, result: dict):
        task.review_operator = operator
        task.review_result = result
        # 此处可以把样本写入数据集，用于后续迭代优化
        print(f"任务{task.task_id}完成复核，操作员:{operator},人工结果:{result}")

if __name__ == "__main__":
    review_queue = ManualReviewQueue()
    task = ReviewTask(
        task_id="rev_0001",
        input_content="待分析业务文本",
        model_outputs=[{"risk_label":"high"},{"risk_label":"low"}],
        reason="多模型输出结论冲突"
    )
    review_queue.push_task(task)
    get_task = review_queue.pop_pending_task()
    if get_task:
        review_queue.submit_review_result(get_task, "user_a", {"risk_label":"low"})