Skip to content

fix(llms): don't stack the sdk's retries on rate-limit replays #8333

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

Open
Simin Yuan (simin-yuan) wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(llms): don't stack the sdk's retries on rate-limit replays #8333

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0996788
fix(llms): don't stack the SDK's retries on rate-limit replays
simin-yuan Oct 20, 2026
5156ac0
fix(llms): disable sdk retries on replays of non-streaming chat
simin-yuan Oct 20, 2026
b581eb2
fix(llms): drop the sdk's retries on replays of structured-output calls
simin-yuan Nov 1, 2026
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
73 changes: 49 additions & 24 deletions lib/crewai/src/crewai/llms/providers/openai/completion.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
from crewai.llms.hooks.base import BaseInterceptor
from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport
from crewai.llms.providers.utils.common import safe_tool_conversion
from crewai.llms.retry import in_rate_limit_retry_replay
from crewai.utilities.agent_utils import is_context_length_exceeded
from crewai.utilities.exceptions.context_window_exceeding_exception import (
LLMContextLengthExceededError,
Expand Down Expand Up @@ -471,6 +472,18 @@ def _get_sync_client(self) -> Any:
self._client = self._build_sync_client()
return self._client

def _client_for_attempt(self, client: Any) -> Any:
"""Return the client to use for the current attempt.

The SDK's own retries belong to the first attempt only. When CrewAI's
rate-limit retry replays the call, the replay would stack its three
tries on top of the three already spent, so the replay runs with SDK
retries disabled: one request per replay instead of three.
"""
if in_rate_limit_retry_replay():
return client.with_options(max_retries=0)
return client

def _get_async_client(self) -> Any:
if self._async_client is None:
self._async_client = self._build_async_client()
Expand Down Expand Up @@ -1184,7 +1197,9 @@ def _handle_responses(
) -> str | ResponsesAPIResult | Any:
"""Handle non-streaming Responses API call."""
try:
response: Response = self._get_sync_client().responses.create(**params)
response: Response = self._client_for_attempt(
self._get_sync_client()
).responses.create(**params)

if self.auto_chain and response.id:
self._last_response_id = response.id
Expand Down Expand Up @@ -1329,9 +1344,9 @@ async def _ahandle_responses(
) -> str | ResponsesAPIResult | Any:
"""Handle async non-streaming Responses API call."""
try:
response: Response = await self._get_async_client().responses.create(
**params
)
response: Response = await self._client_for_attempt(
self._get_async_client()
).responses.create(**params)

if self.auto_chain and response.id:
self._last_response_id = response.id
Expand Down Expand Up @@ -1476,7 +1491,9 @@ def _handle_streaming_responses(
final_response: Response | None = None
usage: dict[str, Any] | None = None

stream = self._get_sync_client().responses.create(**params)
stream = self._client_for_attempt(self._get_sync_client()).responses.create(
**params
)
response_id_stream = None

for event in stream:
Expand Down Expand Up @@ -1613,7 +1630,9 @@ async def _ahandle_streaming_responses(
final_response: Response | None = None
usage: dict[str, Any] | None = None

stream = await self._get_async_client().responses.create(**params)
stream = await self._client_for_attempt(
self._get_async_client()
).responses.create(**params)
response_id_stream = None

async for event in stream:
Expand Down Expand Up @@ -2225,7 +2244,9 @@ def _handle_completion(
parse_params = {
k: v for k, v in params.items() if k != "response_format"
}
raw_parsed = self._get_sync_client().beta.chat.completions.with_raw_response.parse(
raw_parsed = self._client_for_attempt(
self._get_sync_client()
).beta.chat.completions.with_raw_response.parse(
**parse_params,
response_format=response_model,
)
Expand Down Expand Up @@ -2260,11 +2281,9 @@ def _handle_completion(
)
return parsed_object

raw_response = (
Comment thread
cursor[bot] marked this conversation as resolved.
Resolved
self._get_sync_client().chat.completions.with_raw_response.create(
**params
)
)
raw_response = self._client_for_attempt(
self._get_sync_client()
).chat.completions.with_raw_response.create(**params)
_raise_for_upstream_error(
raw_response.text,
model=self.model,
Expand Down Expand Up @@ -2519,7 +2538,9 @@ def _handle_streaming_completion(
}

stream: ChatCompletionStream[BaseModel]
with self._get_sync_client().beta.chat.completions.stream(
with self._client_for_attempt(
self._get_sync_client()
).beta.chat.completions.stream(
**parse_params, response_format=response_model
) as stream:
for chunk in stream:
Expand Down Expand Up @@ -2560,9 +2581,9 @@ def _handle_streaming_completion(
logging.error("Failed to get parsed result from stream")
return ""

completion_stream: Stream[ChatCompletionChunk] = (
self._get_sync_client().chat.completions.create(**params)
)
completion_stream: Stream[ChatCompletionChunk] = self._client_for_attempt(
self._get_sync_client()
).chat.completions.create(**params)

usage_data: dict[str, Any] | None = None
stream_finish_reason: str | None = None
Expand Down Expand Up @@ -2667,7 +2688,9 @@ async def _ahandle_completion(
parse_params = {
k: v for k, v in params.items() if k != "response_format"
}
raw_parsed = await self._get_async_client().beta.chat.completions.with_raw_response.parse(
raw_parsed = await self._client_for_attempt(
self._get_async_client()
).beta.chat.completions.with_raw_response.parse(
**parse_params,
response_format=response_model,
)
Expand Down Expand Up @@ -2702,9 +2725,9 @@ async def _ahandle_completion(
)
return parsed_object

raw_response = await self._get_async_client().chat.completions.with_raw_response.create(
**params
)
raw_response = await self._client_for_attempt(
self._get_async_client()
).chat.completions.with_raw_response.create(**params)
_raise_for_upstream_error(
raw_response.text,
model=self.model,
Expand Down Expand Up @@ -2849,7 +2872,9 @@ async def _ahandle_streaming_completion(
if response_model:
completion_stream: AsyncIterator[
ChatCompletionChunk
] = await self._get_async_client().chat.completions.create(**params)
] = await self._client_for_attempt(
self._get_async_client()
).chat.completions.create(**params)

accumulated_content = ""
usage_data: dict[str, Any] | None = None
Expand Down Expand Up @@ -2914,9 +2939,9 @@ async def _ahandle_streaming_completion(
)
return accumulated_content

stream: AsyncIterator[
ChatCompletionChunk
] = await self._get_async_client().chat.completions.create(**params)
stream: AsyncIterator[ChatCompletionChunk] = await self._client_for_attempt(
self._get_async_client()
).chat.completions.create(**params)

usage_data = None
stream_finish_reason: str | None = None
Expand Down
24 changes: 22 additions & 2 deletions lib/crewai/src/crewai/llms/retry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,18 @@
_active_llm_rate_limit_retry: contextvars.ContextVar[bool] = contextvars.ContextVar(
"_active_llm_rate_limit_retry", default=False
)
_llm_retry_replay: contextvars.ContextVar[bool] = contextvars.ContextVar(
"_llm_retry_replay", default=False
)


def in_rate_limit_retry_replay() -> bool:
"""Whether the attempt running right now replays an already-failed call.

Providers that carry retries of their own use this to keep them for the
first attempt and drop them on replays, so the two layers stop multiplying.
"""
return _llm_retry_replay.get()


class _ThrottlingErrorClassifier:
Expand Down Expand Up @@ -141,7 +153,11 @@ def attempt() -> tuple[bool, _T | None, Exception | None]:
token = _active_llm_rate_limit_retry.set(True)
try:
for attempt_number in range(1, _LLM_RATE_LIMIT_MAX_ATTEMPTS + 1):
succeeded, result, error = attempt()
replay_token = _llm_retry_replay.set(attempt_number > 1)
try:
succeeded, result, error = attempt()
finally:
_llm_retry_replay.reset(replay_token)
if succeeded:
return cast(_T, result)
if error is None:
Expand Down Expand Up @@ -183,7 +199,11 @@ async def attempt() -> tuple[bool, _T | None, Exception | None]:
token = _active_llm_rate_limit_retry.set(True)
try:
for attempt_number in range(1, _LLM_RATE_LIMIT_MAX_ATTEMPTS + 1):
succeeded, result, error = await attempt()
replay_token = _llm_retry_replay.set(attempt_number > 1)
try:
succeeded, result, error = await attempt()
finally:
_llm_retry_replay.reset(replay_token)
if succeeded:
return cast(_T, result)
if error is None:
Expand Down
Loading
Loading