mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-10-13 04:01:35 +00:00
fix(embedding): keep provider-wide failures out of the per-item split
Review follow-up on the P2 raised against 4d8a3c5.
`_call_with_retry` sent every unrecognized exception to `_embed_individually`
whenever the request had more than one item. Only rate limiting and
`insufficient_quota` were excluded, so a 401, a 403 or a 5xx still split a
ten-item request into ten extra single-item requests, none of which can
succeed: credentials, permissions and server outages are properties of the
endpoint, not of any individual input.
The split path is now gated on a dedicated classifier,
`_is_provider_wide_failure()`, which recognizes 401/403 and 5xx by status
code (with the OpenAI SDK class names as a fallback for exceptions that do
not expose one). Transient server errors keep their retry policy through the
existing backoff; the remaining provider-wide failures stop the loop and
report unhealthy, like the quota branch. Input-dependent rejections such as
the per-item context-length limit are unchanged and still isolate one item
at a time.
Regression coverage: test_provider_wide_credentials_errors_are_not_split_into_per_item_requests
(401/403 -> a single request of size 10) and
test_provider_wide_server_errors_are_retried_without_splitting
(500 -> batch sizes [10, 10, 10] with sleeps [1.0, 2.0]). The existing
context-length, quota, 429 and per-item transient-retry tests are retained.
Local verification, using the repo's own hook arguments:
pylint -> Your code has been rated at 10.00/10, exit 0
black -> 2 files would be left unchanged
flake8 -> clean
Classifier discriminator check (old vs new split decision, extracted from the
production file and executed):
400 context-length -> SPLIT / SPLIT
401, 403, 500, 504 -> SPLIT / NO-SPLIT
429 and insufficient_quota keep their own earlier branches.
This commit is contained in:
parent
4d8a3c5475
commit
c9448aeb19
2 changed files with 106 additions and 0 deletions
|
|
@ -220,6 +220,17 @@ class LocalEmbeddingStore(BaseEmbeddingStore):
|
|||
self.logger.exception("Embedding request failed: quota exhausted")
|
||||
self.is_healthy = False
|
||||
return None
|
||||
if self._is_provider_wide_failure(error):
|
||||
# Credentials, permissions and server outages describe the endpoint, not any
|
||||
# single item: isolation cannot fix them, so retry the transient ones with the
|
||||
# usual backoff and give up on the rest instead of multiplying the requests.
|
||||
if self._is_transient_provider_failure(error) and attempt < self.max_retries - 1:
|
||||
delay = 2**attempt
|
||||
self.logger.warning(f"Embedding provider error; retrying in {delay:.1f}s")
|
||||
await asyncio.sleep(delay)
|
||||
continue
|
||||
self.logger.exception("Embedding request failed: provider-wide error, not splitting")
|
||||
break
|
||||
self.logger.exception("Embedding request failed")
|
||||
self.is_healthy = False
|
||||
if len(texts) > 1:
|
||||
|
|
@ -248,6 +259,28 @@ class LocalEmbeddingStore(BaseEmbeddingStore):
|
|||
results.append(None)
|
||||
return results
|
||||
|
||||
@staticmethod
|
||||
def _is_provider_wide_failure(error: Exception) -> bool:
|
||||
"""Return whether an error describes the provider rather than a single input.
|
||||
|
||||
Authentication, permission and server errors belong to the endpoint or to the
|
||||
credentials: no individual input can be blamed, so splitting the batch cannot isolate
|
||||
anything and would only multiply the rejected requests. Rate limits and quota
|
||||
exhaustion have their own branches above and never reach this classifier.
|
||||
"""
|
||||
status = getattr(error, "status_code", None)
|
||||
if isinstance(status, int) and (status in (401, 403) or 500 <= status < 600):
|
||||
return True
|
||||
return type(error).__name__ in ("AuthenticationError", "PermissionDeniedError", "InternalServerError")
|
||||
|
||||
@staticmethod
|
||||
def _is_transient_provider_failure(error: Exception) -> bool:
|
||||
"""Return whether a provider-wide failure is worth retrying with the usual backoff."""
|
||||
status = getattr(error, "status_code", None)
|
||||
if isinstance(status, int):
|
||||
return 500 <= status < 600
|
||||
return type(error).__name__ == "InternalServerError"
|
||||
|
||||
@staticmethod
|
||||
def _is_rate_limited(error: Exception) -> bool:
|
||||
"""Recognize an OpenAI-compatible 429 response without importing a provider SDK."""
|
||||
|
|
|
|||
|
|
@ -200,6 +200,38 @@ class PerItemLimitThenTransientErrorAsEmbedding:
|
|||
return [[1.0, 0.0] for _ in texts]
|
||||
|
||||
|
||||
class AuthenticationError(Exception):
|
||||
"""OpenAI-compatible 401 error used without importing the provider SDK."""
|
||||
|
||||
status_code = 401
|
||||
|
||||
|
||||
class PermissionDeniedError(Exception):
|
||||
"""OpenAI-compatible 403 error used without importing the provider SDK."""
|
||||
|
||||
status_code = 403
|
||||
|
||||
|
||||
class InternalServerError(Exception):
|
||||
"""OpenAI-compatible 500 error used without importing the provider SDK."""
|
||||
|
||||
status_code = 500
|
||||
|
||||
|
||||
class AlwaysProviderWideFailureAsEmbedding:
|
||||
"""Always fail with the same provider-wide error, recording every requested batch size."""
|
||||
|
||||
dimensions = 2
|
||||
|
||||
def __init__(self, error: Exception):
|
||||
self.error = error
|
||||
self.batch_sizes: list[int] = []
|
||||
|
||||
async def __call__(self, texts: list[str], **_kwargs):
|
||||
self.batch_sizes.append(len(texts))
|
||||
raise self.error
|
||||
|
||||
|
||||
class BadNodeEmbeddingStore(BaseEmbeddingStore):
|
||||
"""Embedding store that returns wrong-dimensional vectors."""
|
||||
|
||||
|
|
@ -953,3 +985,44 @@ def test_per_item_fallback_keeps_a_transiently_failed_vector(monkeypatch):
|
|||
assert sleeps == [1.0]
|
||||
|
||||
run(go())
|
||||
|
||||
|
||||
def test_provider_wide_credentials_errors_are_not_split_into_per_item_requests():
|
||||
"""401/403 belong to the credentials: splitting ten items only multiplies the failures."""
|
||||
|
||||
async def go():
|
||||
for error in (
|
||||
AuthenticationError("invalid api key"),
|
||||
PermissionDeniedError("no access to model"),
|
||||
):
|
||||
provider = AlwaysProviderWideFailureAsEmbedding(error)
|
||||
store = LocalEmbeddingStore(name="t_local_embedding_provider_wide", max_retries=3)
|
||||
store.as_embedding = provider
|
||||
|
||||
assert await store._call_with_retry([f"item{i}" for i in range(10)]) is None
|
||||
assert provider.batch_sizes == [10]
|
||||
assert store.is_healthy is False
|
||||
|
||||
run(go())
|
||||
|
||||
|
||||
def test_provider_wide_server_errors_are_retried_without_splitting(monkeypatch):
|
||||
"""A 500 is transient, so it is retried with the usual backoff and never split per item."""
|
||||
|
||||
async def go():
|
||||
sleeps = []
|
||||
|
||||
async def fake_sleep(delay):
|
||||
sleeps.append(delay)
|
||||
|
||||
provider = AlwaysProviderWideFailureAsEmbedding(InternalServerError("upstream unavailable"))
|
||||
store = LocalEmbeddingStore(name="t_local_embedding_provider_server", max_retries=3)
|
||||
store.as_embedding = provider
|
||||
monkeypatch.setattr(asyncio, "sleep", fake_sleep)
|
||||
|
||||
assert await store._call_with_retry([f"item{i}" for i in range(10)]) is None
|
||||
assert provider.batch_sizes == [10, 10, 10]
|
||||
assert sleeps == [1.0, 2.0]
|
||||
assert store.is_healthy is False
|
||||
|
||||
run(go())
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue