Skip to content

Silence repeated rate limit and connection errors in OpenAI utils #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hemanth-asirvatham merged 1 commit into from
Feb 7, 2026
Merged

Silence repeated rate limit and connection errors in OpenAI utils #22

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 73 additions & 14 deletions src/gabriel/utils/openai_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2421,6 +2421,9 @@ async def get_all_embeddings(
get_embedding_kwargs.setdefault("base_url", base_url)
error_logs: Dict[str, List[str]] = defaultdict(list)
queue: asyncio.Queue[Tuple[str, str, int]] = asyncio.Queue()
first_timeout_logged = False
first_rate_limit_logged = False
first_connection_logged = False
for item in items:
queue.put_nowait((item[1], item[0], max_retries))

Expand Down Expand Up @@ -2474,6 +2477,42 @@ def maybe_adjust_concurrency() -> None:
successes_since_adjust = 0
rate_limit_errors_since_adjust = 0

def _log_embedding_timeout_once(message: str) -> None:
nonlocal first_timeout_logged
if not first_timeout_logged:
logger.warning(
"Encountered first timeout error. Future timeout errors will be silenced."
)
first_timeout_logged = True
else:
logger.debug("Timeout error: %s", message)

def _log_embedding_rate_limit_once(detail: Optional[str] = None) -> None:
nonlocal first_rate_limit_logged
if not first_rate_limit_logged:
logger.warning(
"Encountered first rate limit error. Future rate limit errors will be silenced."
)
first_rate_limit_logged = True
else:
if detail:
logger.debug("Rate limit error: %s", detail)
else:
logger.debug("Rate limit error encountered.")

def _log_embedding_connection_once(detail: Optional[str] = None) -> None:
nonlocal first_connection_logged
if not first_connection_logged:
logger.warning(
"Encountered first connection error. Future connection errors will be silenced."
)
first_connection_logged = True
else:
if detail:
logger.debug("Connection error: %s", detail)
else:
logger.debug("Connection error encountered.")

async def worker() -> None:
nonlocal processed, cooldown_until, active_workers, concurrency_cap
nonlocal rate_limit_errors_since_adjust, successes_since_adjust
Expand Down Expand Up @@ -2540,7 +2579,7 @@ async def worker() -> None:
else:
error_message = f"API call timed out after {elapsed:.2f} s"
error_logs[ident].append(error_message)
logger.warning(f"Timeout error for {ident}: {error_message}")
_log_embedding_timeout_once(error_message)
if attempts_left - 1 > 0:
backoff = random.uniform(1, 2) * (2 ** (max_retries - attempts_left))
await asyncio.sleep(backoff)
Expand All @@ -2554,7 +2593,7 @@ async def worker() -> None:
pickle.dump(embeddings, f)
except RateLimitError as e:
error_logs[ident].append(str(e))
logger.warning(f"Rate limit error for {ident}: {e}")
_log_embedding_rate_limit_once(str(e))
cooldown_until = time.time() + global_cooldown
rate_limit_errors_since_adjust += 1
successes_since_adjust = 0
Expand All @@ -2572,7 +2611,7 @@ async def worker() -> None:
pickle.dump(embeddings, f)
except APIConnectionError as e:
error_logs[ident].append(str(e))
logger.warning(f"Connection error for {ident}: {e}")
_log_embedding_connection_once(str(e))
if attempts_left - 1 > 0:
backoff = random.uniform(1, 2) * (2 ** (max_retries - attempts_left))
await asyncio.sleep(backoff)
Expand Down Expand Up @@ -4255,6 +4294,8 @@ def _effective_parallel_ceiling() -> int:
timeout_errors_since_last_status = 0
connection_errors_since_last_status = 0
first_timeout_logged = False
first_rate_limit_logged = False
first_connection_logged = False
current_tokens_per_call = float(estimated_tokens_per_call)
last_rate_limit_concurrency_change = 0.0
restart_requested = False
Expand Down Expand Up @@ -4341,6 +4382,32 @@ def _log_timeout_once(message: str, note: str, *, dedup_key: Hashable) -> None:
else:
logger.debug("Timeout error: %s", message)

def _log_rate_limit_once(detail: Optional[str] = None) -> None:
nonlocal first_rate_limit_logged
if not first_rate_limit_logged:
logger.warning(
"Encountered first rate limit error. Future rate limit errors will be silenced and tracked in periodic updates."
)
first_rate_limit_logged = True
else:
if detail:
logger.debug("Rate limit error: %s", detail)
else:
logger.debug("Rate limit error encountered.")

def _log_connection_once(detail: Optional[str] = None) -> None:
nonlocal first_connection_logged
if not first_connection_logged:
logger.warning(
"Encountered first connection error. Future connection errors will be silenced and tracked in periodic updates."
)
first_connection_logged = True
else:
if detail:
logger.debug("Connection error: %s", detail)
else:
logger.debug("Connection error encountered.")

def _record_timeout_event(now: Optional[float] = None) -> None:
ts = now if now is not None else time.time()
timeout_error_times.append(ts)
Expand Down Expand Up @@ -5317,10 +5384,7 @@ async def _maybe_retry(backoff: float) -> bool:
connection_errors_since_last_status += 1
connection_error_times.append(time.time())
maybe_adjust_for_connection_errors()
_emit_first_error(
f"{base_message}. This can indicate network instability or bandwidth limitations.",
dedup_key=("connection-error", error_detail or None),
)
_log_connection_once(base_message)
error_key: Hashable = (
"non-timeout-error",
type(e).__name__,
Expand Down Expand Up @@ -5409,8 +5473,7 @@ async def _maybe_retry(backoff: float) -> bool:
status.time_of_last_rate_limit_error = time.time()
cooldown_until = status.time_of_last_rate_limit_error + global_cooldown
error_text = str(e)
logger.warning(f"Rate limit error for {ident}: {e}")
_emit_first_error(f"Rate limit error encountered: {error_text}")
_log_rate_limit_once(error_text)
error_logs[ident].append(error_text)
rate_limit_error_times.append(time.time())
rate_limit_errors_since_adjust += 1
Expand Down Expand Up @@ -5496,11 +5559,7 @@ async def _maybe_retry(backoff: float) -> bool:
inflight.pop(ident, None)
status.num_api_errors += 1
status.num_connection_errors += 1
_emit_first_error(
f"Connection error encountered: {e}. This can indicate network instability or bandwidth limitations.",
dedup_key=("connection-error", str(e)),
)
logger.debug(f"Connection error for {ident}: {e}")
_log_connection_once(str(e))
error_logs[ident].append(str(e))
connection_error_times.append(time.time())
connection_errors_since_adjust += 1
Expand Down
Loading