穩定 Ollama embedding GCP 失敗熔斷

services/ollama_service.py

@@ -59,6 +59,8 @@ EMBED_TIMEOUT = int(os.getenv('OLLAMA_EMBED_TIMEOUT', os.getenv('EMBEDDING_TIMEO
 EMBED_MAX_TIMEOUT = int(os.getenv('OLLAMA_EMBED_MAX_TIMEOUT', '15'))
 EMBED_KEEP_ALIVE = os.getenv('OLLAMA_EMBED_KEEP_ALIVE', '1m')
 EMBED_MAX_CHARS = int(os.getenv('OLLAMA_EMBED_MAX_CHARS', '4000'))
+EMBED_GCP_FAILURE_COOLDOWN_SEC = int(os.getenv('OLLAMA_EMBED_GCP_FAILURE_COOLDOWN_SEC', '60'))
+EMBED_GCP_FAILURE_NOTICE_SEC = int(os.getenv('OLLAMA_EMBED_GCP_FAILURE_NOTICE_SEC', '30'))
 FALLBACK_111_KEEP_ALIVE = os.getenv('OLLAMA_111_KEEP_ALIVE', '5m')
 FALLBACK_111_MAX_TIMEOUT = int(os.getenv('OLLAMA_111_MAX_TIMEOUT', '20'))
 FALLBACK_111_NUM_CTX = int(os.getenv('OLLAMA_111_NUM_CTX', '4096'))
@@ -87,6 +89,7 @@ _RESOLVE_TTL = 120  # 主機健康狀態快取 120 秒
 _unhealthy_marks: dict = {}  # host_url -> ts；30s 內被標記就跳過
 _UNHEALTHY_TTL = 30  # 主機被標 unhealthy 後 30 秒內跳過 resolve
 _fallback_111_circuit_cache: dict = {'blocked': False, 'reason': '', 'ts': 0}
+_embedding_gcp_failure_circuit: dict = {'blocked_until': 0.0, 'notice_ts': 0.0, 'tried': ()}
 
 
 def mark_unhealthy(host: str) -> None:
@@ -132,6 +135,43 @@ def _clear_resolved_host_cache() -> None:
     _resolved_host_cache['ts'] = 0
 
 
+def _embedding_gcp_circuit_active() -> bool:
+    """背景 embedding 不落 111；GCP 全掛時短暫熔斷，避免每筆任務重打兩台。"""
+    if EMBED_GCP_FAILURE_COOLDOWN_SEC <= 0:
+        return False
+    import time
+    now = time.time()
+    blocked_until = float(_embedding_gcp_failure_circuit.get('blocked_until') or 0)
+    if now >= blocked_until:
+        return False
+
+    notice_ts = float(_embedding_gcp_failure_circuit.get('notice_ts') or 0)
+    if now - notice_ts >= EMBED_GCP_FAILURE_NOTICE_SEC:
+        logger.warning(
+            "[Embed] GCP embedding circuit open for %.1fs; tried=%s",
+            blocked_until - now,
+            list(_embedding_gcp_failure_circuit.get('tried') or ()),
+        )
+        _embedding_gcp_failure_circuit['notice_ts'] = now
+    return True
+
+
+def _open_embedding_gcp_circuit(attempted_hosts: List[str]) -> None:
+    if EMBED_GCP_FAILURE_COOLDOWN_SEC <= 0 or not attempted_hosts:
+        return
+    import time
+    now = time.time()
+    _embedding_gcp_failure_circuit.update({
+        'blocked_until': now + EMBED_GCP_FAILURE_COOLDOWN_SEC,
+        'notice_ts': now,
+        'tried': tuple(attempted_hosts),
+    })
+
+
+def _reset_embedding_gcp_circuit() -> None:
+    _embedding_gcp_failure_circuit.update({'blocked_until': 0.0, 'notice_ts': 0.0, 'tried': ()})
+
+
 def _fallback_111_block_reason(host: str) -> Tuple[bool, str]:
     """Return whether 111 fallback should be skipped for this request.
 
@@ -982,6 +1022,8 @@ class OllamaService:
         clean_text = (text or "").strip()
         if not clean_text:
             return []
+        if not host and not allow_111_fallback and _embedding_gcp_circuit_active():
+            return []
         if len(clean_text) > EMBED_MAX_CHARS:
             logger.info(
                 "[Embed] input clipped from %s to %s chars for model=%s",
@@ -1037,7 +1079,10 @@ class OllamaService:
             if blocked_111:
                 logger.warning("[Embed] skip 111 fallback explicit host: %s", block_reason)
                 return []
-            return _embed_one(host.rstrip("/"))
+            vec = _embed_one(host.rstrip("/"))
+            if vec:
+                _reset_embedding_gcp_circuit()
+            return vec
 
         # HOTFIX 三主機 retry 鏈（與 generate() 同模式）
         attempted_hosts: List[str] = []
@@ -1086,9 +1131,12 @@ class OllamaService:
 
             vec = _embed_one(target_host)
             if vec:
+                _reset_embedding_gcp_circuit()
                 return vec
             logger.info(f"[Embed] retry #{attempt+1}/{max_attempts} — {target_host} failed, mark_unhealthy + 取新主機")
 
+        if not allow_111_fallback:
+            _open_embedding_gcp_circuit(attempted_hosts)
         logger.error(f"[Embed] all {len(attempted_hosts)} hosts failed; tried={attempted_hosts}")
         return []