diff --git a/gateway/scripts/ocg_router.py b/gateway/scripts/ocg_router.py index 48b1c94..fec9646 100644 --- a/gateway/scripts/ocg_router.py +++ b/gateway/scripts/ocg_router.py @@ -20,7 +20,7 @@ OpenAI 兼容代理,背后池化 6 个 OpenCode Go API key。 └─ dashboard (:5803) ◄── /api/status """ -import os, sys, json, time, logging, threading +import os, sys, json, time, logging, threading, socket from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler from urllib.request import Request, urlopen, HTTPError from urllib.error import URLError @@ -51,6 +51,9 @@ MAX_RETRIES = 6 # 单次上游请求保底超时(秒):上游处理大请求(1MB+)可能要 60-90s+, # 故设宽超时兜底(防 TCP 挂死无限等),正常靠错误分类快速换 key,不依赖此超时 UPSTREAM_REQUEST_TIMEOUT = 180 +# 首字节超时(秒):上游连接建立后,30s 内必须返回第一个字节,否则判定该 key 挂起,换下一个 key。 +# 这是"区分上游慢 vs 挂"的关键——有数据流=在正常生成,不该中断;无响应=key 有问题,快速切换。 +FIRST_BYTE_TIMEOUT = 30 # ── 日志 ──────────────────────────────────────────────────── LOG_DIR = _GATEWAY_DIR / "logs" @@ -633,7 +636,12 @@ class RouterHandler(BaseHTTPRequestHandler): self._error_response(502, f"All OCG keys exhausted: {last_error}") def _forward_request(self, api_key, body, is_stream): - """转发单次请求到指定 key。返回 {ok, status, error}。""" + """转发单次请求到指定 key。返回 {ok, status, error}。 + + stream=True:流式透传——边收上游边发客户端,客户端尽早拿到首 token。 + stream=False:完整读取 + 空输出检测(需要完整 JSON)。 + 两者都做首字节超时(FIRST_BYTE_TIMEOUT):连接建立后 30s 内无响应 = key 挂起,换 key。 + """ try: req = Request( f"{OPENCODE_BASE}/chat/completions", @@ -651,10 +659,43 @@ class RouterHandler(BaseHTTPRequestHandler): err_body = resp.read().decode("utf-8", errors="replace")[:500] return {"ok": False, "status": status, "error": f"HTTP {status}: {err_body}"} - # 透传响应 + 空输出检测(2026-08-14 老莫:空输出自动换 key) - resp_body = resp.read() - if not is_stream: - # 非流式:解析 JSON 检测空输出 + # 首字节超时:底层 socket 设 FIRST_BYTE_TIMEOUT, + # 30s 内读不到第一个字节 → 判定 key 挂起(socket.timeout 会被捕获并换 key) + try: + sock = resp.fp.raw._sock + sock.settimeout(FIRST_BYTE_TIMEOUT) + except Exception: + pass # 拿不到 socket 就算了,靠总超时兜底 + + if is_stream: + # 流式透传:边收边发,客户端尽早拿到首 token + # 一旦开始有数据流(上游在生成),后续读取不再受首字节超时限制 + try: + self.send_response(status) + for k, v in resp.headers.items(): + if k.lower() in ("content-type", "access-control-allow-origin"): + self.send_header(k, v) + self.send_header("Access-Control-Allow-Origin", "*") + self.end_headers() + while True: + chunk = resp.read(8192) + if not chunk: + break + try: + sock.settimeout(None) # 已有数据流,解除首字节超时 + except Exception: + pass + self.wfile.write(chunk) + self.wfile.flush() + self.wfile.flush() + return {"ok": True} + except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): + # 客户端中途断开:数据已发一部分,不算 key 问题 + log.info("client disconnected during streaming via %s", api_key[:8]) + return {"ok": True} # 响应已开始,视为成功(客户端自行处理断流) + else: + # 非流式:完整读取 + 空输出检测 + resp_body = resp.read() try: import json as _json d = _json.loads(resp_body.decode("utf-8", errors="replace")) @@ -668,15 +709,14 @@ class RouterHandler(BaseHTTPRequestHandler): has_reasoning = bool(reasoning and str(reasoning).strip()) if (not content or not str(content).strip()) and not has_reasoning: if finish_reason == "length": - # max_tokens 截断,content 可能为空但属于正常响应 log.warning("content empty but finish_reason=length (truncated), passing through") else: log.warning("empty content detected via key, retrying next key") return {"ok": False, "status": 0, "error": "empty_content"} except Exception: pass # 解析失败不阻断(正常透传) - self._send_response(status, dict(resp.headers), resp_body) - return {"ok": True} + self._send_response(status, dict(resp.headers), resp_body) + return {"ok": True} except HTTPError as e: err_body = e.read().decode("utf-8", errors="replace")[:500] @@ -684,6 +724,9 @@ class RouterHandler(BaseHTTPRequestHandler): except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError) as e: # 客户端断开:不是 key 问题,标记特殊错误码让调用方不重试不冷却 return {"ok": False, "status": -1, "error": f"client_disconnected: {e}"} + except (TimeoutError, socket.timeout) as e: + # 首字节超时或总超时:key 挂起,换下一个 key 重试(不冷却) + return {"ok": False, "status": 0, "error": f"upstream_timeout: {e}"} except URLError as e: return {"ok": False, "status": 0, "error": f"URLError: {e.reason}"} except Exception as e: