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.logger.exception("Embedding request failed: quota exhausted")
|
||||||
self.is_healthy = False
|
self.is_healthy = False
|
||||||
return None
|
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.logger.exception("Embedding request failed")
|
||||||
self.is_healthy = False
|
self.is_healthy = False
|
||||||
if len(texts) > 1:
|
if len(texts) > 1:
|
||||||
|
|
@ -248,6 +259,28 @@ class LocalEmbeddingStore(BaseEmbeddingStore):
|
||||||
results.append(None)
|
results.append(None)
|
||||||
return results
|
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
|
@staticmethod
|
||||||
def _is_rate_limited(error: Exception) -> bool:
|
def _is_rate_limited(error: Exception) -> bool:
|
||||||
"""Recognize an OpenAI-compatible 429 response without importing a provider SDK."""
|
"""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]
|
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):
|
class BadNodeEmbeddingStore(BaseEmbeddingStore):
|
||||||
"""Embedding store that returns wrong-dimensional vectors."""
|
"""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]
|
assert sleeps == [1.0]
|
||||||
|
|
||||||
run(go())
|
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