大模型超时控制与异常重试机制：分级超时、幂等重试、退避策略与雪崩防护.161

import requests
import time

# 全局基础配置
# 单次请求最大超时时间，适配常规短文本大模型请求
BASE_SINGLE_TIMEOUT = 10
# 最大重试次数，避免无限重试
MAX_RETRY_TIMES = 3
# 基础重试间隔，无退避策略，适用于入门演示
BASIC_RETRY_SLEEP = 1

def base_llm_chat_request(prompt: str):
    """
    基础版大模型请求函数
    能力：固定超时控制 + 简单次数重试 + 异常分类捕获
    """
    # 模拟本地大模型推理服务接口
    request_url = "http://192.168.3.61:8000/v1/chat/completions"
    request_payload = {
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7
    }

    # 循环执行重试逻辑
    for retry_count in range(MAX_RETRY_TIMES):
        try:
            # 核心：强制设置请求超时时间
            response = requests.post(
                url=request_url,
                json=request_payload,
                timeout=BASE_SINGLE_TIMEOUT
            )
            # 主动抛出HTTP状态码异常
            response.raise_for_status()
            # 请求正常返回，格式化结果
            return {
                "code": 200,
                "message": "请求执行成功",
                "data": response.json(),
                "retry_used": retry_count
            }

        # 捕获请求超时专属异常
        except requests.exceptions.Timeout:
            print(f"【异常提醒】第{retry_count + 1}次请求触发全局超时，即将进入重试流程")

        # 捕获网络连接异常
        except requests.exceptions.ConnectionError:
            print(f"【异常提醒】第{retry_count + 1}次请求网络连接失败，存在链路抖动")

        # 捕获HTTP业务异常
        except requests.exceptions.HTTPError as error:
            # 区分5xx可重试异常与4xx不可重试异常
            if response.status_code >= 500:
                print(f"【异常提醒】第{retry_count + 1}次遇到服务端异常，状态码：{response.status_code}")
            else:
                return {
                    "code": response.status_code,
                    "message": f"业务参数或权限错误，不可重试：{str(error)}"
                }

        # 最后一次重试失败，直接终止流程
        if retry_count == MAX_RETRY_TIMES - 1:
            return {
                "code": 504,
                "message": "全部重试次数耗尽，大模型请求最终执行失败"
            }
        
        # 固定间隔休眠后重试
        time.sleep(BASIC_RETRY_SLEEP)

# 本地测试运行
if __name__ == "__main__":
    res = base_llm_chat_request("简述大模型超时控制的核心作用")
    print("最终请求结果：", res)

import requests
import time

def get_dynamic_timeout_config(text_length: int, is_stream: bool, user_priority: str = "normal"):
    """
    动态计算分级超时配置
    返回：总超时、流式分片超时
    """
    # 1. 根据文本长度初始化基础超时
    if text_length < 500:
        base_total = 5
    elif text_length < 2000:
        base_total = 15
    else:
        base_total = 60

    chunk_timeout = None
    # 2. 按流式/非流式调整参数
    if is_stream:
        chunk_timeout = 1.0
        base_total = min(base_total * 1.5, 120)

    # 3. 按业务优先级浮动调整
    if user_priority == "high":
        final_total = base_total * 1.2
    elif user_priority == "low":
        final_total = base_total * 0.7
    else:
        final_total = base_total

    return {
        "total_timeout": round(final_total, 1),
        "chunk_timeout": chunk_timeout
    }

def graded_timeout_llm_request(prompt: str, is_stream: bool = False, priority: str = "normal"):
    # 文本长度统计，模拟Token量级判断
    prompt_len = len(prompt)
    timeout_cfg = get_dynamic_timeout_config(prompt_len, is_stream, priority)
    total_t = timeout_cfg["total_timeout"]
    chunk_t = timeout_cfg["chunk_timeout"]

    print(f"【分级超时配置】总超时：{total_t}s，分片超时：{chunk_t}s")
    url = "http://192.168.3.6:8000/v1/chat/completions"
    payload = {
        "prompt": prompt,
        "stream": is_stream
    }

    try:
        resp = requests.post(
            url=url,
            json=payload,
            timeout=total_t,
            stream=is_stream
        )
        resp.raise_for_status()

        # 流式分片超时逻辑
        if is_stream and chunk_t:
            last_time = time.time()
            for chunk in resp.iter_lines(decode_unicode=True):
                # 检测分片间隔超时
                if time.time() - last_time > chunk_t:
                    raise TimeoutError(f"流式分片超时：连续{chunk_t}s无数据返回")
                last_time = time.time()
                if chunk:
                    print("流式片段：", chunk)

        return {
            "code": 200,
            "msg": "请求正常完成",
            "timeout_config": timeout_cfg
        }

    except requests.exceptions.Timeout:
        return {"code": 504, "msg": f"全局总超时触发，限制时长：{total_t}s"}
    except TimeoutError as e:
        return {"code": 504, "msg": f"流式分片异常：{str(e)}"}

if __name__ == "__main__":
    # 模拟长文本+流式+高优用户场景
    result = graded_timeout_llm_request(
        prompt="请详细撰写一份人工智能行业年度发展分析报告，包含技术趋势、落地场景、行业挑战",
        is_stream=True,
        priority="high"
    )
    print("最终结果：", result)

import requests
import time
import uuid
import random
from typing import Optional

# ===================== 生产级全局配置 =====================
MAX_RETRY_COUNT = 4        # 最大重试次数
BASE_BACKOFF_SEC = 1       # 基础退避秒数
MAX_BACKOFF_SEC = 16       # 最大退避上限，防止等待过久
TIMEOUT_LIMIT = 20         # 单次请求全局超时

# 可重试异常状态码白名单
RETRY_STATUS_WHITE_LIST = {502, 503, 504, 500}

class IdempotentLLMRequest:
    def __init__(self):
        # 生成全局唯一幂等请求ID
        self.request_id: str = str(uuid.uuid4())

    def _calc_backoff_time(self, retry_times: int) -> float:
        """
        计算退避等待时间：指数退避 + 随机抖动 + 最大值限制
        """
        # 指数退避
        exp_time = BASE_BACKOFF_SEC * (2 ** retry_times)
        # 限制上限
        exp_time = min(exp_time, MAX_BACKOFF_SEC)
        # 增加随机抖动 0.5 ~ 1.0倍
        jitter_ratio = random.uniform(0.5, 1.0)
        final_sleep = exp_time * jitter_ratio
        return round(final_sleep, 2)

    def request_with_retry(self, prompt: str) -> dict:
        url = "http://192.168.3.26:8000/v1/chat/completions"
        payload = {
            "prompt": prompt,
            "request_id": self.request_id,  # 幂等核心字段
            "temperature": 0.7
        }

        resp: Optional[requests.Response] = None

        for retry_idx in range(MAX_RETRY_COUNT):
            try:
                resp = requests.post(
                    url=url,
                    json=payload,
                    timeout=TIMEOUT_LIMIT
                )
                resp.raise_for_status()
                return {
                    "code": 200,
                    "success": True,
                    "request_id": self.request_id,
                    "retry_used": retry_idx,
                    "data": resp.json()
                }

            except requests.exceptions.Timeout:
                print(f"[{self.request_id}] 第{retry_idx+1}次请求超时")

            except requests.exceptions.ConnectionError:
                print(f"[{self.request_id}] 第{retry_idx+1}次网络连接异常")

            except requests.exceptions.HTTPError:
                # 非白名单状态码，直接终止重试
                if resp and resp.status_code not in RETRY_STATUS_WHITE_LIST:
                    return {
                        "code": resp.status_code,
                        "success": False,
                        "msg": "不可重试业务异常"
                    }
                print(f"[{self.request_id}] 服务端临时异常，状态码：{resp.status_code}")

            # 最后一次重试，不再等待
            if retry_idx >= MAX_RETRY_COUNT - 1:
                return {
                    "code": 504,
                    "success": False,
                    "request_id": self.request_id,
                    "msg": "重试次数耗尽，请求失败"
                }

            # 计算退避时间并休眠
            sleep_time = self._calc_backoff_time(retry_idx)
            print(f"[{self.request_id}] 等待{sleep_time}s后进行下一次重试\n")
            time.sleep(sleep_time)

if __name__ == "__main__":
    client = IdempotentLLMRequest()
    result = client.request_with_retry("分析大模型退避重试的工程价值")
    print("最终请求结果：", result)

import requests
import time
import uuid

class StreamLLMBreakpointRetry:
    def __init__(self):
        # 重试配置
        self.max_stream_retry = 2
        self.chunk_interval_timeout = 3.0
        self.task_total_timeout = 90
        # 幂等标识
        self.request_id = str(uuid.uuid4())
        # 断点缓存：保存已生成内容
        self.finished_content = ""

    def _stream_request_core(self, prompt: str, continue_text: str = None) -> dict:
        """
        核心流式请求方法
        :param prompt: 原始提问
        :param continue_text: 断点接续内容
        """
        url = "http://192.168.3.61:8000/v1/chat/completions"
        payload = {
            "prompt": prompt,
            "stream": True,
            "request_id": self.request_id,
            "continue_content": continue_text
        }

        try:
            response = requests.post(
                url=url,
                json=payload,
                stream=True,
                timeout=self.task_total_timeout
            )
            response.raise_for_status()

            last_chunk_time = time.time()
            for chunk in response.iter_lines(decode_unicode=True):
                # 分片间隔超时检测
                current_time = time.time()
                if current_time - last_chunk_time > self.chunk_interval_timeout:
                    raise TimeoutError("推理僵死，分片间隔超时")

                last_chunk_time = current_time
                if chunk and chunk.strip():
                    self.finished_content += chunk.strip()
                    print("实时输出片段：", chunk.strip())

            return {
                "success": True,
                "content": self.finished_content,
                "msg": "流式生成完成"
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "breakpoint_content": self.finished_content
            }

    def start_stream_task(self, prompt: str) -> dict:
        """流式任务入口：自动断点重试"""
        retry_count = 0
        current_continue = None

        while retry_count < self.max_stream_retry:
            print(f"\n===== 流式生成 第{retry_count + 1}次尝试 =====")
            result = self._stream_request_core(prompt, current_continue)

            if result["success"]:
                return result

            # 记录断点，准备接续重试
            current_continue = result["breakpoint_content"]
            print(f"流式任务中断，已生成内容长度：{len(current_continue)}")
            print(f"异常原因：{result['error']}")
            retry_count += 1

        # 重试耗尽，兜底返回已有内容
        return {
            "success": False,
            "content": self.finished_content,
            "msg": "流式重试已用尽，返回已生成内容"
        }

if __name__ == "__main__":
    stream_client = StreamLLMBreakpointRetry()
    final_res = stream_client.start_stream_task("写一篇500字的自然风景散文，文笔细腻优美")
    print("\n===== 最终完整内容 =====")
    print(final_res["content"])