diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index 4e84d55..b2fd1ee 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -41,6 +41,9 @@ _RETRYABLE_STATUS = (429, 500, 502, 503, 504) _MAX_ATTEMPTS = 3 _TOTAL_TIMEOUT = 300 _CONNECT_TIMEOUT = 30 +# 单次 attempt 最少剩余预算(2026-09-16 统一预算 deadline 配套):剩余不足 +# 该值不再发起新 attempt(必然超时白烧),直接抛预算耗尽错误 +_MIN_ATTEMPT_SECONDS = 15 EST_TOKENS_PER_CHAR = 2 # 预估用量:中文约 1 字符≈0.5 token,保守按 2 字符 1 token @@ -426,15 +429,34 @@ async def _post_upstream(ctx, payload, require_choices=True): kind = 'chat' if require_choices else 'gen' # method 独立列(2026-09-06 用户定夺):默认 POST,profile 声明优先 method = (profile.get('method') or 'POST').upper() - timeout = int(_fnum(ctx.get('timeout')) or _TOTAL_TIMEOUT) + # 统一预算 deadline(单点治理,2026-09-16 用户裁定收敛): + # ctx['budget'] = 本次调用【总预算】(bridge._resolve_budget 经 payload. + # _timeout 恒传)。attempt 超时 = 剩余预算(合法长生成可一次用满, + # 373s 实测样本;快速失败的瞬时错误自然留出重试余量)。重试受 + # deadline 约束,总时长≈预算,客户端(预算+60s)恒覆盖上游。 + # 端点配置 timeout(百炼 120 等)不再掐断 attempt——它是供应商参数, + # 不知道调用方需求,正是 09-05/09-14/09-16 三次事故的根因;仅在 + # 直连端点且未传 _timeout 时参与兜底(与 _TOTAL_TIMEOUT 取大)。 + # 旧实现每 attempt 独享全额超时,最坏 3×预算+退避 > 客户端等待 → 客户端 + # 先断连报模糊 TimeoutError(分类器误判永久错误,2026-09-16 QC 事故链一环)。 + budget = int(_fnum(ctx.get('budget')) or 0) + if budget <= 0: + budget = max(int(_fnum(ctx.get('timeout')) or 0), _TOTAL_TIMEOUT) + deadline = time.time() + budget # 可诊断性:只记长度不记值(真 key 不落日志)。print 确保进应用日志 # (模块 logging.getLogger 的 handler 未接应用日志管道,2026-09-04 实测) - print("[inference] 组包 url=%s auth头长度=%d body字段=%s" % ( + print("[inference] 组包 url=%s auth头长度=%d body字段=%s 预算=%ds" % ( url.split('?')[0], len(headers.get('Authorization', '') or ''), - sorted(body.keys()))) + sorted(body.keys()), budget)) last = None for attempt in range(_MAX_ATTEMPTS): + remaining = deadline - time.time() + if remaining < _MIN_ATTEMPT_SECONDS: + raise ValueError('上游调用失败(总预算 %ds 耗尽,已试 %d/%d 次,%s): %s' % ( + budget, attempt, _MAX_ATTEMPTS, type(last).__name__ if last else '无', + (repr(last) if last else '')[:200])) + timeout = int(remaining) _t0 = time.time() try: async with aiohttp.ClientSession() as session: @@ -887,8 +909,14 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref='', # llm_usage.call_id 同值 → 流水可反查全部原文(trace.record_roundtrip / llm_call_trace) from appPublic.uniqueID import getID ctx['call_id'] = getID() + # 统一预算(2026-09-16 收敛):payload._timeout = 本次调用【总预算】→ + # ctx['budget'](_post_upstream 按 deadline 约束重试,总时长≈预算)。 + # ctx['timeout'] 保持供应商端点配置语义 = 单 attempt 上限,不再被覆盖 + # (旧实现覆盖后每 attempt 独享全额预算,最坏 3× 超客户端等待)。 + # 未传 _timeout(直连端点的第三方调用)→ budget 走 _TOTAL_TIMEOUT 平台 + # 缺省,不再落到端点小配置掐断长生成。 if req_timeout > 0: - ctx['timeout'] = min(req_timeout, 900) + ctx['budget'] = min(req_timeout, 900) # 异步模型:提交→轮询→取结果(用量因子走 usages 记账) model_row = ctx.get('model_row') or {} if (model_row.get('sync_mode') or 'sync') == 'async':