refactor(models): 统一模型错误分类与重试边界

bot/vikingbot/compile/ops/common.py

@@ -11,7 +11,7 @@
 from typing import TYPE_CHECKING
 
 from openviking.core.namespace import relative_uri_path
-from openviking.utils.model_retry import ERROR_CLASS_INPUT_TOO_LARGE, classify_api_error
+from openviking.models.model_error import ModelErrorKind, model_error_kind
 from vikingbot.compile import file_ops
 from vikingbot.compile.hashing import content_hash, digest
 from vikingbot.compile.pipeline_io import MAX_JOB_ATTEMPTS
@@ -347,7 +347,7 @@ async def job(runtime: Pipeline, name, inputs, work):
             except OSError as exc:
                 if (
                     attempt == MAX_JOB_ATTEMPTS - 1
-                    or classify_api_error(exc) == ERROR_CLASS_INPUT_TOO_LARGE
+                    or model_error_kind(exc) is ModelErrorKind.INPUT_TOO_LARGE
                 ):
                     raise
                 runtime.metrics["job_retries"] += 1

bot/vikingbot/compile/ops/reduce.py

@@ -11,7 +11,7 @@
 from typing import TYPE_CHECKING
 
 from openviking.core.namespace import relative_uri_path
-from openviking.utils.model_retry import ERROR_CLASS_INPUT_TOO_LARGE, classify_api_error
+from openviking.models.model_error import ModelErrorKind, model_error_kind
 from vikingbot.compile import file_ops
 from vikingbot.compile.hashing import content_hash, digest
 from vikingbot.compile.ops import common
@@ -317,7 +317,7 @@ def validate_files(response):
             break
         except ModelCallError as exc:
             if (
-                classify_api_error(exc) != ERROR_CLASS_INPUT_TOO_LARGE
+                model_error_kind(exc) is not ModelErrorKind.INPUT_TOO_LARGE
                 or depth == 3
                 or stage == "map"
             ):

bot/vikingbot/compile/pipeline_io.py

@@ -16,6 +16,7 @@
 from loguru import logger
 from pydantic import BaseModel, ValidationError
 
+from openviking.models.model_error import ModelError
 from openviking.utils.token_estimation import estimate_text_tokens
 from openviking_cli.exceptions import NotFoundError
 from vikingbot.compile.hashing import digest
@@ -318,6 +319,12 @@ async def call(self, stage, messages, tools, *, max_tokens):
                     self.usage[name] = self.usage.get(name, 0) + value
                     self.metrics[f"{stage}_{name}"] += value
             if response.finish_reason == "error":
+                error_kind = getattr(response, "error_kind", None)
+                if error_kind is not None:
+                    raise ModelError(
+                        error_kind,
+                        response.content or "Model transport failure; see provider diagnostics",
+                    )
                 raise OSError(
                     response.content or "Model transport failure; see provider diagnostics"
                 )

bot/vikingbot/providers/base.py

@@ -7,6 +7,8 @@
 
 import json_repair
 
+from openviking.models.model_error import ModelErrorKind
+
 
 @dataclass
 class ToolCallRequest:
@@ -27,6 +29,7 @@ class LLMResponse:
     finish_reason: str = "stop"
     usage: dict[str, int] = field(default_factory=dict)
     reasoning_content: str | None = None  # Kimi, DeepSeek-R1 etc.
+    error_kind: ModelErrorKind | None = None
 
     @property
     def has_tool_calls(self) -> bool:

bot/vikingbot/providers/vlm_adapter.py

@@ -15,6 +15,7 @@
 
 from loguru import logger
 
+from openviking.models.model_error import model_error_kind
 from openviking.models.vlm.backends.volcengine_vlm import build_volcengine_request_headers
 from openviking.utils.model_retry import is_retryable_rate_limit_error, rate_limit_retry_delay
 from openviking.utils.multimodal import redact_image_data_urls
@@ -336,6 +337,7 @@ async def chat(
             return LLMResponse(
                 content=f"Error calling LLM in VLM Adapter: {str(e)}",
                 finish_reason="error",
+                error_kind=model_error_kind(e),
             )
 
     async def chat_stream(
@@ -389,6 +391,7 @@ async def chat_stream(
                 response=LLMResponse(
                     content=f"Error calling LLM in VLM Adapter stream: {str(exc)}",
                     finish_reason="error",
+                    error_kind=model_error_kind(exc),
                 ),
             )
 

openviking/config/embedding.py

@@ -17,12 +17,6 @@
 from openviking.models.vlm.token_usage import TokenUsageTracker
 from openviking.service.task_tracker_concurrency import KeyedAsyncLockPool
 from openviking.utils.circuit_breaker import CircuitBreaker, CircuitBreakerOpen
-from openviking.utils.model_retry import (
-    ERROR_CLASS_AUTH,
-    ERROR_CLASS_INPUT_TOO_LARGE,
-    ERROR_CLASS_PERMANENT,
-    classify_api_error,
-)
 from openviking_cli.utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -260,16 +254,7 @@ async def _execute(self, resource, content, is_query):
                         f"got {len(result.dense_vector)}"
                     )
         except Exception as exc:
-            # The breaker is shared by every call on this account, so only
-            # provider-health failures may trip it. Request-level rejections
-            # (400 / input too large) are specific to one input, and auth
-            # errors are surfaced per call instead.
-            if classify_api_error(exc) not in {
-                ERROR_CLASS_AUTH,
-                ERROR_CLASS_INPUT_TOO_LARGE,
-                ERROR_CLASS_PERMANENT,
-            }:
-                resource.breaker.record_failure(exc)
+            resource.breaker.record_failure(exc)
             raise
         resource.breaker.record_success()
         return result

openviking/models/embedder/base.py

@@ -12,16 +12,14 @@
 from threading import Lock
 from typing import Any, AsyncIterator, Awaitable, Callable, Dict, List, Optional, TypeVar, Union
 
+from openviking.models.model_error import model_error_kind
 from openviking.telemetry import get_current_telemetry
 from openviking.utils.embedding_input import (
     resolve_embedding_max_input_tokens,
     truncate_embedding_input,
 )
 from openviking.utils.exceptions import AllCredentialsFailedError
-from openviking.utils.model_retry import (
-    OrderedCredentialSwitcher,
-    classify_api_error,
-)
+from openviking.utils.model_retry import OrderedCredentialSwitcher
 from openviking_cli.utils import get_logger
 
 T = TypeVar("T")
@@ -856,16 +854,18 @@ def _embed_with_failover(self, method_name: str, *args, **kwargs) -> EmbedResult
                 self._switcher.commit_success(idx)
                 return result
             except Exception as exc:
-                error_class = classify_api_error(exc)
-                aggregated_errors.append((credential_id, error_class, exc, idx))
+                kind = model_error_kind(exc)
 
-                if self._switcher.is_fail_fast(error_class):
+                if self._switcher.is_fail_fast(kind):
                     # Request-level failure: re-raise the original exception;
                     # trying other credentials is useless.
                     raise
+                aggregated_errors.append((credential_id, kind, exc, idx))
 
                 logger.warning(
-                    f"Credential {credential_id} failed with {error_class}, trying next credential"
+                    "Credential %s failed with %s, trying next credential",
+                    credential_id,
+                    kind.value,
                 )
 
         raise AllCredentialsFailedError(aggregated_errors)
@@ -901,16 +901,18 @@ async def _embed_with_failover_async(self, method_name: str, *args, **kwargs) ->
                 self._switcher.commit_success(idx)
                 return result
             except Exception as exc:
-                error_class = classify_api_error(exc)
-                aggregated_errors.append((credential_id, error_class, exc, idx))
+                kind = model_error_kind(exc)
 
-                if self._switcher.is_fail_fast(error_class):
+                if self._switcher.is_fail_fast(kind):
                     # Request-level failure: re-raise the original exception;
                     # trying other credentials is useless.
                     raise
+                aggregated_errors.append((credential_id, kind, exc, idx))
 
                 logger.warning(
-                    f"Credential {credential_id} failed with {error_class}, trying next credential"
+                    "Credential %s failed with %s, trying next credential",
+                    credential_id,
+                    kind.value,
                 )
 
         raise AllCredentialsFailedError(aggregated_errors)

openviking/models/embedder/cohere_embedders.py

@@ -12,6 +12,7 @@
 import httpx
 
 from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult, truncate_and_normalize
+from openviking.models.model_error import model_error_boundary
 from openviking.utils.async_client_cache import LoopScopedAsyncClientCache
 from openviking_cli.utils import get_logger
 
@@ -132,6 +133,7 @@ def _normalize_vector(self, vector: List[float]) -> List[float]:
             return truncate_and_normalize(vector, self._dimension)
         return vector
 
+    @model_error_boundary
     def embed(self, text: str, is_query: bool = False) -> EmbedResult:
         input_type = "search_query" if is_query else "search_document"
         try:
@@ -157,6 +159,7 @@ def embed(self, text: str, is_query: bool = False) -> EmbedResult:
         except Exception as e:
             raise RuntimeError(f"Cohere embedding failed: {e}") from e
 
+    @model_error_boundary
     async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult:
         input_type = "search_query" if is_query else "search_document"
 

openviking/models/embedder/dashscope_embedders.py

@@ -17,6 +17,7 @@
     EmbedResult,
     truncate_and_normalize,
 )
+from openviking.models.model_error import model_error_boundary
 from openviking.telemetry import get_current_telemetry
 from openviking.utils.async_client_cache import LoopScopedAsyncClientCache
 from openviking_cli.utils.logger import default_logger as logger
@@ -225,6 +226,7 @@ def _get_async_httpx_client(self) -> httpx.AsyncClient:
     # embed
     # ------------------------------------------------------------------
 
+    @model_error_boundary
     def embed(self, text: EmbeddingInput, is_query: bool = False) -> EmbedResult:
         if isinstance(text, list):
             contents = self._to_dashscope_contents(text)
@@ -256,6 +258,7 @@ def _call() -> EmbedResult:
     # embed_async
     # ------------------------------------------------------------------
 
+    @model_error_boundary
     async def embed_async(self, text: EmbeddingInput, is_query: bool = False) -> EmbedResult:
         if isinstance(text, list):
             contents = self._to_dashscope_contents(text)
@@ -291,6 +294,7 @@ async def _call() -> EmbedResult:
     # embed_content — multimodal content (text + image URLs)
     # ------------------------------------------------------------------
 
+    @model_error_boundary
     def embed_content(
         self, contents: List[Dict[str, str]], *, enable_fusion: Optional[bool] = None
     ) -> EmbedResult:
@@ -326,6 +330,7 @@ def _call() -> EmbedResult:
         except Exception as e:
             raise RuntimeError(f"DashScope content embedding failed: {e}") from e
 
+    @model_error_boundary
     async def embed_content_async(
         self, contents: List[Dict[str, str]], *, enable_fusion: Optional[bool] = None
     ) -> EmbedResult:

openviking/models/embedder/gemini_embedders.py

@@ -20,6 +20,7 @@
     EmbedResult,
     truncate_and_normalize,
 )
+from openviking.models.model_error import model_error_boundary
 from openviking_cli.utils import get_logger
 
 logger = get_logger(__name__)
@@ -184,6 +185,7 @@ def __repr__(self) -> str:
             f"task_type={self.task_type!r})"
         )
 
+    @model_error_boundary
     def embed(
         self,
         text: str,
@@ -225,6 +227,7 @@ def _call() -> EmbedResult:
         except (APIError, ClientError) as e:
             _raise_api_error(e, self.model_name)
 
+    @model_error_boundary
     async def embed_async(
         self,
         text: str,

openviking/models/embedder/jina_embedders.py

@@ -10,6 +10,7 @@
     DenseEmbedderBase,
     EmbedResult,
 )
+from openviking.models.model_error import model_error_boundary
 from openviking.utils.async_client_cache import LoopScopedAsyncClientCache
 from openviking_cli.utils import get_logger
 
@@ -183,6 +184,7 @@ def _raise_task_error(self, error: openai.APIError) -> None:
                 f"to a valid task type for this model. API details: {error.message}"
             ) from error
 
+    @model_error_boundary
     def embed(self, text: str, is_query: bool = False) -> EmbedResult:
         """Perform dense embedding on text
 
@@ -224,6 +226,7 @@ def _call() -> EmbedResult:
         except Exception as e:
             raise RuntimeError(f"Embedding failed: {str(e)}") from e
 
+    @model_error_boundary
     async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult:
         client = self._get_async_client()
 

openviking/models/embedder/litellm_embedders.py

@@ -12,6 +12,7 @@
 import litellm
 
 from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult
+from openviking.models.model_error import model_error_boundary
 from openviking.telemetry import get_current_telemetry
 from openviking_cli.utils import get_logger
 
@@ -166,6 +167,7 @@ def _usage_value(key: str, default: int = 0) -> int:
             completion_tokens=output_tokens,
         )
 
+    @model_error_boundary
     def embed(self, text: str, is_query: bool = False) -> EmbedResult:
         """Perform dense embedding on text via litellm.
 
@@ -199,6 +201,7 @@ def _call() -> EmbedResult:
         except Exception as e:
             raise RuntimeError(f"LiteLLM embedding failed: {e}") from e
 
+    @model_error_boundary
     async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult:
         async def _call() -> EmbedResult:
             kwargs = self._build_kwargs(is_query=is_query)

openviking/models/embedder/minimax_embedders.py

@@ -8,6 +8,7 @@
 import requests
 
 from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult
+from openviking.models.model_error import ModelError, ModelErrorKind, model_error_boundary
 from openviking.utils.async_client_cache import LoopScopedAsyncClientCache
 from openviking_cli.utils.logger import default_logger as logger
 
@@ -112,14 +113,21 @@ def _call_api(self, texts: List[str], is_query: bool = False) -> List[List[float
             # Check for business error code
             base_resp = data.get("base_resp", {})
             if base_resp.get("status_code") != 0:
-                raise RuntimeError(f"MiniMax API error: {base_resp.get('status_msg')}")
+                raise ModelError(
+                    ModelErrorKind.UNKNOWN,
+                    f"MiniMax API error: {base_resp.get('status_msg')}",
+                    provider="minimax",
+                    code=str(base_resp.get("status_code")),
+                )
 
             vectors = data.get("vectors", [])
             if not vectors:
                 raise RuntimeError("MiniMax API returned empty vectors")
 
             return vectors
 
+        except ModelError:
+            raise
         except requests.exceptions.RequestException as e:
             raise RuntimeError(f"MiniMax network error: {str(e)}") from e
         except Exception as e:
@@ -169,18 +177,26 @@ async def _call_api_async(self, texts: List[str], is_query: bool = False) -> Lis
 
             base_resp = data.get("base_resp", {})
             if base_resp.get("status_code") != 0:
-                raise RuntimeError(f"MiniMax API error: {base_resp.get('status_msg')}")
+                raise ModelError(
+                    ModelErrorKind.UNKNOWN,
+                    f"MiniMax API error: {base_resp.get('status_msg')}",
+                    provider="minimax",
+                    code=str(base_resp.get("status_code")),
+                )
 
             vectors = data.get("vectors", [])
             if not vectors:
                 raise RuntimeError("MiniMax API returned empty vectors")
 
             return vectors
+        except ModelError:
+            raise
         except httpx.HTTPError as e:
             raise RuntimeError(f"MiniMax network error: {str(e)}") from e
         except Exception as e:
             raise RuntimeError(f"MiniMax embedding failed: {str(e)}") from e
 
+    @model_error_boundary
     def embed(self, text: str, is_query: bool = False) -> EmbedResult:
         """Perform dense embedding on text"""
         vectors = self._run_once(
@@ -199,6 +215,7 @@ def embed(self, text: str, is_query: bool = False) -> EmbedResult:
         )
         return result
 
+    @model_error_boundary
     async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult:
         async def _call() -> EmbedResult:
             vectors = await self._call_api_async([text], is_query=is_query)