mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-27 08:28:00 +00:00
feat(backend): bound LLM call concurrency and shed burst-rate (429) retries (#4294)
* Create a feature of Process-global LLM concurrency cap * Added configuration of llm_call of max_concurrent_calls * Classify limit_burst_rate and expose retry params via config.yaml * refactor(middleware): encapsulate LLM concurrency state in a dataclass Address PR #4294 review feedback (github-code-quality bot): the bare module-level globals _GLOBAL_CONCURRENCY_LOOP / _GLOBAL_CONCURRENCY_LIMIT were flagged as unused - a false positive, since both are read on the recreate condition, but the `global`-declaration pattern tripped the analyzer. Replace the three globals + `global` declaration with a single _ConcurrencyState dataclass singleton mutated in place. Behavior is unchanged (lazy recreate when the running loop or configured limit changes); the state is now co-located and no longer relies on bare globals. dataclasses is already an established harness convention. Co-Authored-By: Claude <noreply@anthropic.com> * fix(middleware): make LLM concurrency limiter process-wide + jitter burst retry Addresses PR #4294 review (fancyboi999, CHANGES_REQUESTED) - two P1 issues. P1 #1: the asyncio.Semaphore limiter was loop-bound, so it recreated per event loop and the cap was NOT process-wide: lead-agent calls (main loop) and subagent calls (the isolated persistent loop in subagents/executor.py) each got their own semaphore, and the sync graph path (wrap_model_call) bypassed the cap entirely. Recreating on loop/limit change also abandoned permits held by the prior instance. Replace it with a _ProcessWideLimiter built on threading primitives (not loop-bound): one limiter shared across every event loop and both sync/async wrappers. The cap is mutable via set_limit (never recreates, so in-flight permits are never abandoned); permits release in finally and async waiters unregister on cancellation, so cancellation never leaks capacity. Wire it into wrap_model_call (sync) too - previously a direct handler() call. P1 #2: the first (and only) burst-rate retry was deterministic at 5000ms. prev_delay_ms was seeded from the 1000ms normal base, so for burst the window collapsed to randint(5000, max(5000, 1000*3)) = randint(5000, 5000) - a fleet that failed together realigned on the same 5s tick. Seed the first retry from the reason-specific base (prev_delay_ms=None on loop init) so the burst window is [burst_base, cap] = [5000, 8000], non-degenerate. Retry-After is still honored verbatim. Tests: rename semaphore tests -> limiter; add an autouse fixture resetting the process singleton; add regressions the reviewer asked for - cross-loop (lead + isolated-loop subagent), two concurrent sync calls, limit-change while a permit is held (same instance, permit preserved), cancellation no-leak, and burst first-retry non-degeneracy with default config (real and seeded RNG) plus a concurrent de-synchronization case. Verified the burst guard goes red on the old logic ({5000}) and green on the new. Co-Authored-By: Claude <noreply@anthropic.com> * Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> * Potential fix for pull request finding 'Statement has no effect' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> * fix(middleware): lossless limiter handoff, generation-aware cap, burst-rate CB gate Address the open P1/P2 review findings on #4294: - P1 #1 (cancellation handoff): reserve the permit for a specific waiter at dequeue time (grant-at-dequeue, _AsyncWaiter.granted) so a waiter cancelled in the post-dequeue / pre-reacquire window hands its reservation to the next waiter (_handoff_granted_permit_locked) instead of stranding it. No cancellation window remains. - P1 #2 (hot-reload generation): move cap updates out of the per-attempt path; give the limiter one generation-aware owner (set_limit_if_newer with a monotonic instance seq proxy for config freshness) so a stale in-flight run cannot rewrite a freshly-lowered cap. max_concurrent_calls is now genuinely hot-reloadable, resolving the reload-boundary inconsistency by option (b) - no STARTUP_ONLY_FIELDS change (retry params truly hot-reload). - P2 (circuit breaker): gate _record_failure on reason != "burst_rate" so burst-rate (limit_burst_rate) exhaustion - a transient slope-throttle, not "provider down" - does not trip the CB and fast-fail ALL calls. - P3: clamp the jitter window to the cap before drawing (uniform spread instead of piling at the cap); document the per-process / GATEWAY_WORKERS cap semantics in config + the field description. Tests: add the reviewer-requested regressions (cancel-after-dequeue handoff; stale-instance-doesn't-overwrite-lowered-cap across sync + isolated-loop async; burst_rate-exhaustion-doesn't-trip-CB sync + async). Each is red on the prior buggy logic and green on the fix. _build_middleware now routes llm_call knobs through AppConfig so __init__ applies the cap. 71 middleware tests pass; 212 across the blast radius (1 pre-existing skip). Co-Authored-By: Claude <noreply@anthropic.com> * fix(middleware): startup-only LLM concurrency cap; report effective retry budget Addresses review feedback on #4294 (fancyboi999 CHANGES_REQUESTED on acfc7617): P1 - the generation guard measured construction order, not config freshness, so a stale AppConfig(cap=3) constructed after a fresher AppConfig(cap=1) could restore the higher cap; and on a downscale 3->1 release() handed excess permits to queued waiters, keeping in_flight pegged at the old cap. Replace the pseudo-generation path with a startup-only cap: the first middleware __init__ resolves and freezes the cap; later instances (newer or older config) are no-ops. No runtime cap mutation means no downscale race and no freshness/construction-order race. Per-call gate is now `limiter is None` only, so a reloaded instance with max_concurrent_calls=0 cannot silently drop the frozen cap. Removes _owner_seq / set_limit_if_newer / _grant_to_queued_locked / _next_instance_seq; file 946 -> 926 lines. P2 - burst-rate calls are capped at 2 attempts but the retry log line, the llm_retry stream event max_attempts, and the user-facing message still used self.retry_max_attempts (3), so the frontend showed 1/3 then stopped after attempt 2. Thread the effective max_attempts (_max_attempts_for) through the logger, _emit_retry_event, and _build_retry_message. Also: document max_concurrent_calls as startup-only in the config field description and config.example.yaml (prose only - the startup-only: prefix is top-level AppConfig-field granularity and would mislabel the otherwise hot-reloadable llm_call section / break the reload_boundary drift test). Rewrite the cap-mutation tests for startup-only semantics; add P2 retry-budget event tests (sync+async, teeth-verified red on the bug); fix bot nits (empty except blocks -> gather(return_exceptions=True); bare await statements -> assigned+asserted). Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
This commit is contained in:
Willem Jiang
GitHub
parent
e66f455d51
commit
ac5fd46281
@ -4,8 +4,10 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
import threading
|
||||
import time
|
||||
from collections import deque
|
||||
from collections.abc import Awaitable, Callable
|
||||
from email.utils import parsedate_to_datetime
|
||||
from typing import Any, override
|
||||
@ -62,6 +64,18 @@ _AUTH_PATTERNS = (
|
||||
"未授权",
|
||||
)
|
||||
|
||||
# Provider burst-rate (``limit_burst_rate``) signals. This is a *rate-of-change*
|
||||
# limit, not a quota limit: the provider throttles when request RPM ramps up too
|
||||
# steeply (e.g. the 08:30 morning peak going 0 -> full throttle in seconds).
|
||||
# Matched against both the error message and the error ``code``/``type``.
|
||||
_BURST_PATTERNS = (
|
||||
"limit_burst_rate",
|
||||
"rate increased too quickly",
|
||||
"burst rate",
|
||||
"请求速率增长过快",
|
||||
"突发速率",
|
||||
)
|
||||
|
||||
# Per-exception retry budget overrides.
|
||||
#
|
||||
# Some transient errors are retriable in principle but expensive to retry at
|
||||
@ -82,6 +96,18 @@ _RETRY_BUDGET_OVERRIDES: dict[str, int] = {
|
||||
"StreamChunkTimeoutError": 2,
|
||||
}
|
||||
|
||||
# Per-reason retry budget overrides, applied in addition to the per-exception
|
||||
# overrides above; the tightest bound wins (so neither loosens the other) and
|
||||
# the user-configured ``retry_max_attempts`` still caps everything.
|
||||
#
|
||||
# A burst-rate (``limit_burst_rate``) 429 gets a tight budget on purpose:
|
||||
# retrying into the burst adds demand to the very request-rate slope being
|
||||
# throttled, so we keep at most one retry (with a longer backoff) and then shed
|
||||
# load rather than hammering the provider. Keys are ``_classify_error`` reasons.
|
||||
_REASON_RETRY_BUDGETS: dict[str, int] = {
|
||||
"burst_rate": 2,
|
||||
}
|
||||
|
||||
# Exception class names that indicate the upstream stream-chunk watchdog
|
||||
# fired because the model stalled mid-flight. These deserve a more specific
|
||||
# user-facing message than the generic "temporarily unavailable" copy,
|
||||
@ -98,12 +124,261 @@ _STREAM_DROP_EXCEPTIONS: frozenset[str] = frozenset(
|
||||
)
|
||||
|
||||
|
||||
# Process-global LLM call concurrency cap. ONE limiter is shared across every
|
||||
# ``LLMErrorHandlingMiddleware`` instance and every call path: the lead agent
|
||||
# (main event loop), subagents (the isolated persistent loop in
|
||||
# subagents/executor.py), ``asyncio.run`` tests, and the sync graph path. That
|
||||
# matters because a provider burst-rate (``limit_burst_rate``) limit fires on
|
||||
# the *slope* of the request rate, so the cap must bound aggregate in-flight
|
||||
# calls process-wide - a per-loop cap (which is what asyncio.Semaphore would
|
||||
# give) is defeated the moment subagent fan-out runs on a second loop.
|
||||
#
|
||||
# Correctness invariants the design below preserves:
|
||||
# * Lossless waiter handoff: a permit handed to a waiter is *reserved* for
|
||||
# that waiter at dequeue time (``granted=True``). If the waiter is
|
||||
# cancelled before it wakes, the reserved permit is re-handed to the next
|
||||
# waiter (or freed) - so a cancellation in the post-dequeue/pre-reacquire
|
||||
# window never strands the next waiter with capacity idle.
|
||||
# * Startup-only cap: the cap is resolved ONCE, at the first middleware
|
||||
# construction (``_apply_configured_cap``), and frozen thereafter. Later
|
||||
# ``__init__`` calls never touch the cap - whether they hold a newer or an
|
||||
# older ``AppConfig`` snapshot. This removes the pseudo-generation path
|
||||
# entirely: with no cap mutation at runtime there is no downscale that
|
||||
# could hand excess permits to queued waiters (keeping ``in_flight`` pegged
|
||||
# at the old cap), and no construction-order race where a stale config
|
||||
# constructed after a fresher one could restore a higher cap. Per-attempt
|
||||
# callers only acquire/release. Changing the cap requires a gateway
|
||||
# restart (see ``LlmCallConfig.max_concurrent_calls``).
|
||||
|
||||
|
||||
class _AsyncWaiter:
|
||||
"""A parked async caller awaiting a transferred permit.
|
||||
|
||||
``granted`` is flipped to ``True`` (under the limiter lock) at the exact
|
||||
moment a permit is reserved for this waiter - by ``release`` handing off a
|
||||
returning permit, or by another cancelling waiter handing off its reserved
|
||||
permit. The reservation is atomic with the dequeue, so the invariant
|
||||
``granted is True <=> not in _async_waiters`` always holds: once granted,
|
||||
the permit is already counted in ``_in_flight`` and the waiter need only
|
||||
wake and return. A cancelled waiter therefore knows from ``granted``
|
||||
whether it owes a handoff (granted) or is merely unregistering (not yet
|
||||
granted).
|
||||
"""
|
||||
|
||||
__slots__ = ("loop", "event", "granted")
|
||||
|
||||
def __init__(self, loop: asyncio.AbstractEventLoop, event: asyncio.Event) -> None:
|
||||
self.loop = loop
|
||||
self.event = event
|
||||
self.granted = False
|
||||
|
||||
|
||||
class _ProcessWideLimiter:
|
||||
"""In-flight call limiter shared across event loops and sync/async wrappers.
|
||||
|
||||
``asyncio.Semaphore`` binds to the first event loop that uses it and raises
|
||||
if acquired from another, so it cannot cap lead-agent and subagent calls
|
||||
together (they run on different loops), nor the sync graph path. This
|
||||
limiter is built on ``threading`` primitives (not loop-bound): every call
|
||||
path shares one in-flight counter and one cap.
|
||||
|
||||
The cap is **immutable**: it is set once at construction (by
|
||||
``_apply_configured_cap`` on the first middleware ``__init__``) and never
|
||||
mutated afterwards. Because the cap never changes at runtime there is no
|
||||
downscale race (a lowered cap could otherwise keep admitting queued
|
||||
waiters until ``in_flight`` drains) and no config-freshness race (a stale
|
||||
snapshot constructed later could otherwise restore a higher cap). Per-
|
||||
attempt callers (``acquire_sync``/``acquire_async``/``release``) never
|
||||
touch the cap. Permits are released in a ``finally`` and an async waiter
|
||||
that is cancelled after its permit was reserved hands the reservation to
|
||||
the next waiter, so capacity never leaks and a cancellation never strands
|
||||
a later waiter.
|
||||
"""
|
||||
|
||||
def __init__(self, limit: int) -> None:
|
||||
self._lock = threading.Lock()
|
||||
self._cond = threading.Condition(self._lock)
|
||||
self._in_flight = 0
|
||||
self._limit = max(0, limit)
|
||||
# FIFO of async callers waiting on capacity. Each waiter lives on its
|
||||
# caller's loop; release/handoff wakes one across loops via
|
||||
# call_soon_threadsafe so the wakeup runs on the right loop.
|
||||
self._async_waiters: deque[_AsyncWaiter] = deque()
|
||||
|
||||
@property
|
||||
def limit(self) -> int:
|
||||
return self._limit
|
||||
|
||||
@property
|
||||
def in_flight(self) -> int:
|
||||
return self._in_flight
|
||||
|
||||
def acquire_sync(self) -> None:
|
||||
"""Block the calling thread until a permit is available, then take one."""
|
||||
with self._cond:
|
||||
while not self._try_acquire_locked():
|
||||
self._cond.wait()
|
||||
|
||||
def release(self) -> None:
|
||||
"""Return one permit, handing it to a waiter if one is queued.
|
||||
|
||||
If an async waiter is queued, the returning permit *transfers* to it
|
||||
(ownership moves; ``_in_flight`` is unchanged) and its event is set so
|
||||
it wakes already owning a permit. Otherwise the permit returns to the
|
||||
free pool (``_in_flight -= 1``) and one sync waiter is notified to grab
|
||||
it on its next ``_try_acquire_locked`` re-check.
|
||||
"""
|
||||
with self._cond:
|
||||
if self._async_waiters:
|
||||
waiter = self._async_waiters.popleft()
|
||||
waiter.granted = True
|
||||
if not self._wake_locked(waiter):
|
||||
# Owner loop closed: the transferred permit is stranded;
|
||||
# hand it to the next waiter or free it.
|
||||
self._handoff_granted_permit_locked()
|
||||
return
|
||||
if self._in_flight > 0:
|
||||
self._in_flight -= 1
|
||||
self._cond.notify()
|
||||
|
||||
async def acquire_async(self) -> None:
|
||||
"""Acquire a permit without blocking the event loop.
|
||||
|
||||
Free capacity -> take one immediately. Otherwise park on an
|
||||
``asyncio.Event``; ``release`` / a cap-raise transfers a permit to us
|
||||
(``granted=True``) and sets the event. On cancellation, if a permit was
|
||||
already reserved for us, hand it to the next waiter (or free it) so the
|
||||
reservation is never lost; if we were still queued (not yet granted),
|
||||
just unregister - no permit was reserved for us, so there is nothing to
|
||||
release.
|
||||
"""
|
||||
loop = asyncio.get_running_loop()
|
||||
while True:
|
||||
waiter = _AsyncWaiter(loop=loop, event=asyncio.Event())
|
||||
with self._cond:
|
||||
if self._try_acquire_locked():
|
||||
return
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug(
|
||||
"LLM call parking on process-wide limiter (in_flight=%d, limit=%d, queued=%d)",
|
||||
self._in_flight,
|
||||
self._limit,
|
||||
len(self._async_waiters) + 1,
|
||||
)
|
||||
self._async_waiters.append(waiter)
|
||||
try:
|
||||
await waiter.event.wait()
|
||||
except asyncio.CancelledError:
|
||||
with self._cond:
|
||||
if waiter.granted:
|
||||
# A permit was reserved for us but we're cancelling
|
||||
# before waking. Pass the reservation to the next
|
||||
# waiter (or free it) so it is not stranded.
|
||||
self._handoff_granted_permit_locked()
|
||||
else:
|
||||
# Still queued, never granted (granted is set only when
|
||||
# dequeued, under the lock): just unregister.
|
||||
self._async_waiters.remove(waiter)
|
||||
raise
|
||||
return # woken => granted => we own a permit (already in _in_flight)
|
||||
|
||||
def _try_acquire_locked(self) -> bool:
|
||||
if self._in_flight < self._limit:
|
||||
self._in_flight += 1
|
||||
return True
|
||||
return False
|
||||
|
||||
def _handoff_granted_permit_locked(self) -> None:
|
||||
"""Transfer an already-reserved permit to the next queued waiter, or free it.
|
||||
|
||||
Used when a waiter that had a permit reserved cancels before waking, or
|
||||
when a reservation target's loop is dead. The permit is already counted
|
||||
in ``_in_flight``; transferring keeps it counted (ownership moves to the
|
||||
next waiter), freeing returns it to the pool. Either way ``_in_flight``
|
||||
stays correct and the reservation is never lost.
|
||||
"""
|
||||
while self._async_waiters:
|
||||
waiter = self._async_waiters.popleft()
|
||||
waiter.granted = True
|
||||
if self._wake_locked(waiter):
|
||||
return # ownership transferred; _in_flight unchanged
|
||||
# dead loop; try the next waiter
|
||||
# No async waiter to take it: free the permit and wake a sync waiter.
|
||||
if self._in_flight > 0:
|
||||
self._in_flight -= 1
|
||||
self._cond.notify()
|
||||
|
||||
def _wake_locked(self, waiter: _AsyncWaiter) -> bool:
|
||||
"""Schedule ``event.set`` on the waiter's loop. False if the loop is dead."""
|
||||
try:
|
||||
waiter.loop.call_soon_threadsafe(waiter.event.set)
|
||||
return True
|
||||
except RuntimeError:
|
||||
return False # owner loop closed: the wakeup cannot land
|
||||
|
||||
|
||||
_LIMITER_LOCK = threading.Lock()
|
||||
_PROCESS_LIMITER: _ProcessWideLimiter | None = None
|
||||
|
||||
# Whether the process-wide cap has been resolved yet. The cap is startup-only:
|
||||
# the first ``LLMErrorHandlingMiddleware`` ``__init__`` resolves it (creating a
|
||||
# limiter for a positive cap, or leaving it ``None`` for a disabled cap) and
|
||||
# every subsequent ``__init__`` is a no-op - regardless of whether its
|
||||
# ``AppConfig`` snapshot is newer or older than the first. This is the single
|
||||
# owner of the cap; per-attempt callers only acquire/release.
|
||||
_CAP_RESOLVED: bool = False
|
||||
|
||||
|
||||
def _get_process_limiter() -> _ProcessWideLimiter | None:
|
||||
"""Return the process-wide LLM-call limiter, or ``None`` when the cap is
|
||||
disabled (or before the first middleware construction resolves it).
|
||||
|
||||
Per-attempt callers use this to acquire/release only - it never changes the
|
||||
cap. ``limiter is None`` is the sole gate for "cap disabled": a per-call
|
||||
short-circuit on the instance's configured value would let a later
|
||||
(reloaded) instance with ``max_concurrent_calls=0`` silently drop the cap
|
||||
mid-process, which is exactly the hot-reload churn the startup-only design
|
||||
removes.
|
||||
"""
|
||||
return _PROCESS_LIMITER
|
||||
|
||||
|
||||
def _apply_configured_cap(limit: int) -> None:
|
||||
"""Resolve the process-wide cap from the first middleware ``__init__``.
|
||||
|
||||
Startup-only: the very first call wins and freezes the cap. A positive
|
||||
``limit`` creates the limiter at that cap; ``limit <= 0`` resolves the cap
|
||||
as disabled (limiter stays ``None``, callers short-circuit on
|
||||
``limiter is None``). Every later call - whether it carries a newer or an
|
||||
older ``AppConfig`` snapshot, and whether it would raise or lower the cap -
|
||||
is ignored, so the cap can never be mutated at runtime. Changing it requires
|
||||
a gateway restart.
|
||||
"""
|
||||
global _PROCESS_LIMITER, _CAP_RESOLVED
|
||||
if _CAP_RESOLVED:
|
||||
return # cap already frozen at first construction; this instance is a no-op
|
||||
with _LIMITER_LOCK:
|
||||
if _CAP_RESOLVED:
|
||||
return
|
||||
_CAP_RESOLVED = True
|
||||
if limit > 0:
|
||||
_PROCESS_LIMITER = _ProcessWideLimiter(limit)
|
||||
|
||||
|
||||
class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
"""Retry transient LLM errors and surface graceful assistant messages."""
|
||||
|
||||
retry_max_attempts: int = 3
|
||||
retry_base_delay_ms: int = 1000
|
||||
retry_cap_delay_ms: int = 8000
|
||||
# Longer backoff base used only for burst-rate (limit_burst_rate) 429s, so
|
||||
# the single burst retry lands after the throttle window subsides.
|
||||
burst_retry_base_delay_ms: int = 5000
|
||||
# Process-wide cap on concurrently in-flight LLM calls. 0 disables the cap
|
||||
# (default) so existing deployments see no behavior change; set to a
|
||||
# positive int to bound aggregate concurrency and smooth provider
|
||||
# burst-rate (limit_burst_rate) spikes. See _get_process_limiter.
|
||||
max_concurrent_llm_calls: int = 0
|
||||
|
||||
def __init__(self, *, app_config: AppConfig, **kwargs: Any) -> None:
|
||||
super().__init__(**kwargs)
|
||||
@ -111,6 +386,23 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
self.circuit_failure_threshold = app_config.circuit_breaker.failure_threshold
|
||||
self.circuit_recovery_timeout_sec = app_config.circuit_breaker.recovery_timeout_sec
|
||||
|
||||
# Retry / backoff / concurrency knobs are all configured via the
|
||||
# ``llm_call`` section of config.yaml; they override the class defaults
|
||||
# above so operators can tune them without code changes.
|
||||
llm_call = app_config.llm_call
|
||||
self.retry_max_attempts = llm_call.retry_max_attempts
|
||||
self.retry_base_delay_ms = llm_call.retry_base_delay_ms
|
||||
self.retry_cap_delay_ms = llm_call.retry_cap_delay_ms
|
||||
self.burst_retry_base_delay_ms = llm_call.burst_retry_base_delay_ms
|
||||
self.max_concurrent_llm_calls = llm_call.max_concurrent_calls
|
||||
|
||||
# Resolve the process-wide cap (startup-only: the first ``__init__`` in
|
||||
# the process wins and freezes it; later instances - newer or older
|
||||
# config - are no-ops). Per-attempt callers only acquire/release, so the
|
||||
# cap can never be mutated at runtime and there is no downscale or
|
||||
# config-freshness race to admit waiters above the live cap.
|
||||
_apply_configured_cap(self.max_concurrent_llm_calls)
|
||||
|
||||
# Circuit Breaker state
|
||||
self._circuit_lock = threading.Lock()
|
||||
self._circuit_failure_count = 0
|
||||
@ -118,17 +410,24 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
self._circuit_state = "closed"
|
||||
self._circuit_probe_in_flight = False
|
||||
|
||||
def _max_attempts_for(self, exc: BaseException) -> int:
|
||||
def _max_attempts_for(self, exc: BaseException, reason: str = "transient") -> int:
|
||||
"""Return the effective max attempt count for this exception.
|
||||
|
||||
Falls back to `self.retry_max_attempts` unless the exception class name
|
||||
appears in the per-exception override table.
|
||||
The user-configured ``retry_max_attempts`` is the ceiling; per-exception
|
||||
(``_RETRY_BUDGET_OVERRIDES``, keyed by class name) and per-reason
|
||||
(``_REASON_RETRY_BUDGETS``, keyed by ``_classify_error`` reason)
|
||||
overrides can only *tighten* it. The tightest bound wins, so a burst-rate
|
||||
429 never gets more attempts than its dedicated budget even if the
|
||||
operator raised the global cap.
|
||||
"""
|
||||
override = _RETRY_BUDGET_OVERRIDES.get(type(exc).__name__)
|
||||
if override is None:
|
||||
return self.retry_max_attempts
|
||||
|
||||
return min(override, self.retry_max_attempts)
|
||||
candidates = [self.retry_max_attempts]
|
||||
class_override = _RETRY_BUDGET_OVERRIDES.get(type(exc).__name__)
|
||||
if class_override is not None:
|
||||
candidates.append(class_override)
|
||||
reason_override = _REASON_RETRY_BUDGETS.get(reason)
|
||||
if reason_override is not None:
|
||||
candidates.append(reason_override)
|
||||
return min(candidates)
|
||||
|
||||
def _check_circuit(self) -> bool:
|
||||
"""Returns True if circuit is OPEN (fast fail), False otherwise."""
|
||||
@ -203,6 +502,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
return False, "quota"
|
||||
if _matches_any(lowered, _AUTH_PATTERNS):
|
||||
return False, "auth"
|
||||
# Burst-rate (limit_burst_rate) 429 is retriable but needs its own
|
||||
# policy: a tight retry budget and a longer backoff base (see
|
||||
# _REASON_RETRY_BUDGETS / _build_retry_delay_ms). Detected before the
|
||||
# generic 429->transient mapping so it isn't lumped in with ordinary
|
||||
# transient errors.
|
||||
if _matches_any(lowered, _BURST_PATTERNS) or _matches_any(str(error_code).lower(), _BURST_PATTERNS):
|
||||
return True, "burst_rate"
|
||||
|
||||
exc_name = exc.__class__.__name__
|
||||
if exc_name in {
|
||||
@ -232,17 +538,121 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
|
||||
return False, "generic"
|
||||
|
||||
def _build_retry_delay_ms(self, attempt: int, exc: BaseException) -> int:
|
||||
def _bounded_model_call_sync(
|
||||
self,
|
||||
request: ModelRequest,
|
||||
handler: Callable[[ModelRequest], ModelResponse],
|
||||
) -> ModelResponse:
|
||||
"""Run one sync model attempt under the process-global concurrency cap.
|
||||
|
||||
The limiter wraps a *single* attempt only (not the retry loop), so
|
||||
backoff sleeps release the slot for other callers. ``limiter is None``
|
||||
(cap disabled at startup) is a direct passthrough; a non-``None``
|
||||
limiter is always consulted - the cap is frozen at the first
|
||||
``__init__``, so a later instance whose ``max_concurrent_llm_calls`` is
|
||||
0 cannot silently drop it. Permits release on any exit (return or
|
||||
raise) via ``finally`` so a raised handler never leaks a slot.
|
||||
"""
|
||||
limiter = _get_process_limiter()
|
||||
if limiter is None:
|
||||
return handler(request)
|
||||
limiter.acquire_sync()
|
||||
try:
|
||||
return handler(request)
|
||||
finally:
|
||||
limiter.release()
|
||||
|
||||
async def _bounded_model_call(
|
||||
self,
|
||||
request: ModelRequest,
|
||||
handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
|
||||
) -> ModelResponse:
|
||||
"""Run one async model attempt under the process-global concurrency cap.
|
||||
|
||||
The limiter wraps a *single* attempt only (not the retry loop), so
|
||||
backoff sleeps release the slot for other callers - we bound in-flight
|
||||
requests, not waiting ones. ``limiter is None`` (cap disabled at
|
||||
startup) is a direct passthrough; a non-``None`` limiter is always
|
||||
consulted (cap frozen at first ``__init__``). Permits release on any
|
||||
exit (return, raise, or cancellation) via ``finally``;
|
||||
``acquire_async`` separately cleans up if cancelled while waiting, so
|
||||
capacity never leaks.
|
||||
"""
|
||||
limiter = _get_process_limiter()
|
||||
if limiter is None:
|
||||
return await handler(request)
|
||||
await limiter.acquire_async()
|
||||
try:
|
||||
return await handler(request)
|
||||
finally:
|
||||
limiter.release()
|
||||
|
||||
def _build_retry_delay_ms(self, prev_delay_ms: int | None, exc: BaseException, reason: str = "transient") -> int:
|
||||
"""Compute the next retry delay (ms) using decorrelated jitter.
|
||||
|
||||
An explicit ``Retry-After`` from the provider is honored as-is (no
|
||||
jitter) - the server told us exactly when to come back, and for a
|
||||
burst-rate 429 this is strongly preferred over any computed delay.
|
||||
Otherwise AWS-style "decorrelated jitter" is applied:
|
||||
``delay = random(base, min(cap, max(base, seed * 3)))`` where ``seed``
|
||||
is the previous delay, or the reason-specific base on the first retry
|
||||
(``prev_delay_ms is None``). The window is clamped to the cap *before*
|
||||
drawing (not after) so the distribution stays uniform up to the cap
|
||||
rather than piling up at it. ``reason="burst_rate"`` swaps in
|
||||
``burst_retry_base_delay_ms`` (longer than the normal base) so the
|
||||
single burst retry lands after the throttle window subsides.
|
||||
|
||||
Seeding the first retry from the *reason-specific* base (not always the
|
||||
normal base) is what keeps the first-and-only burst retry
|
||||
non-degenerate: with the normal base (1000ms) the burst window would
|
||||
collapse to ``randint(5000, max(5000, 1000*3)) = randint(5000, 5000)``
|
||||
and every concurrent burst failure would realign on the same 5s tick.
|
||||
Seeding from 5000ms gives ``randint(5000, min(8000, 15000)) =
|
||||
randint(5000, 8000)`` with defaults, so a fleet that failed together
|
||||
spreads out across the whole window.
|
||||
|
||||
Deterministic exponential backoff (``base * 2^(attempt-1)``) makes
|
||||
every concurrent retryer realign on the same backoff ticks; when a
|
||||
whole fleet fails at once (e.g. a provider burst-rate limit at the
|
||||
morning peak) that synchronized retry storm re-triggers the very limit
|
||||
we are backing off from. Decorrelated jitter spreads those retries
|
||||
across a random window so they don't re-peak in lockstep.
|
||||
"""
|
||||
retry_after = _extract_retry_after_ms(exc)
|
||||
if retry_after is not None:
|
||||
return retry_after
|
||||
backoff = self.retry_base_delay_ms * (2 ** max(0, attempt - 1))
|
||||
return min(backoff, self.retry_cap_delay_ms)
|
||||
base = self.burst_retry_base_delay_ms if reason == "burst_rate" else self.retry_base_delay_ms
|
||||
cap = self.retry_cap_delay_ms
|
||||
seed = base if prev_delay_ms is None else prev_delay_ms
|
||||
# Clamp the window to the cap *before* drawing so the jitter spreads
|
||||
# uniformly across [base, min(cap, seed*3)] instead of concentrating at
|
||||
# the cap: with defaults seed*3 (=15000) >> cap (=8000), drawing
|
||||
# randint(base, seed*3) then min(delay, cap) would put ~70% of draws at
|
||||
# exactly cap, re-clustering a fleet that the jitter is meant to spread.
|
||||
high = min(cap, max(base, seed * 3))
|
||||
if high < base:
|
||||
return cap # base exceeds cap (misconfiguration): the cap wins
|
||||
return random.randint(base, high)
|
||||
|
||||
def _build_retry_message(self, attempt: int, wait_ms: int, reason: str) -> str:
|
||||
def _build_retry_message(
|
||||
self,
|
||||
attempt: int,
|
||||
wait_ms: int,
|
||||
reason: str,
|
||||
*,
|
||||
max_attempts: int,
|
||||
) -> str:
|
||||
seconds = max(1, round(wait_ms / 1000))
|
||||
reason_text = "provider is busy" if reason == "busy" else "provider request failed temporarily"
|
||||
return f"LLM request retry {attempt}/{self.retry_max_attempts}: {reason_text}. Retrying in {seconds}s."
|
||||
reason_text = {
|
||||
"busy": "provider is busy",
|
||||
"burst_rate": "provider is throttling request burst rate",
|
||||
}.get(reason, "provider request failed temporarily")
|
||||
# ``max_attempts`` is the *effective* budget for this call (from
|
||||
# ``_max_attempts_for``), not the configured ceiling: a burst-rate call
|
||||
# is capped at 2 attempts, so its message must read ``1/2`` not ``1/3``
|
||||
# even when ``retry_max_attempts`` is the default 3 - otherwise the UI
|
||||
# promises a retry that will never happen.
|
||||
return f"LLM request retry {attempt}/{max_attempts}: {reason_text}. Retrying in {seconds}s."
|
||||
|
||||
def _build_circuit_breaker_message(self) -> str:
|
||||
return "The configured LLM provider is currently unavailable due to continuous failures. Circuit breaker is engaged to protect the system. Please wait a moment before trying again."
|
||||
@ -271,6 +681,8 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
return "The configured LLM provider rejected the request because the account is out of quota, billing is unavailable, or usage is restricted. Please fix the provider account and try again."
|
||||
if reason == "auth":
|
||||
return "The configured LLM provider rejected the request because authentication or access is invalid. Please check the provider credentials and try again."
|
||||
if reason == "burst_rate":
|
||||
return "The configured LLM provider is temporarily throttling requests because the request rate increased too quickly (burst-rate limit). Please wait a moment and try again."
|
||||
if reason in {"busy", "transient"}:
|
||||
# Stream-drop failures (chunk-gap timeout, peer-closed connection,
|
||||
# raw read error) almost always point at a single oversized
|
||||
@ -297,7 +709,14 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
detail=_extract_error_detail(exc),
|
||||
)
|
||||
|
||||
def _emit_retry_event(self, attempt: int, wait_ms: int, reason: str) -> None:
|
||||
def _emit_retry_event(
|
||||
self,
|
||||
attempt: int,
|
||||
wait_ms: int,
|
||||
reason: str,
|
||||
*,
|
||||
max_attempts: int,
|
||||
) -> None:
|
||||
try:
|
||||
from langgraph.config import get_stream_writer
|
||||
|
||||
@ -306,10 +725,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
{
|
||||
"type": "llm_retry",
|
||||
"attempt": attempt,
|
||||
"max_attempts": self.retry_max_attempts,
|
||||
# Effective budget for this call (burst-rate == 2), not the
|
||||
# configured ceiling - the frontend renders this and the
|
||||
# ``message`` below, so both must describe the loop that runs.
|
||||
"max_attempts": max_attempts,
|
||||
"wait_ms": wait_ms,
|
||||
"reason": reason,
|
||||
"message": self._build_retry_message(attempt, wait_ms, reason),
|
||||
"message": self._build_retry_message(attempt, wait_ms, reason, max_attempts=max_attempts),
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
@ -330,9 +752,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
)
|
||||
|
||||
attempt = 1
|
||||
prev_delay_ms: int | None = None
|
||||
while True:
|
||||
try:
|
||||
response = handler(request)
|
||||
response = self._bounded_model_call_sync(request, handler)
|
||||
self._record_success()
|
||||
return response
|
||||
except GraphBubbleUp:
|
||||
@ -341,17 +764,18 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
raise
|
||||
except Exception as exc:
|
||||
retriable, reason = self._classify_error(exc)
|
||||
max_attempts = self._max_attempts_for(exc)
|
||||
max_attempts = self._max_attempts_for(exc, reason)
|
||||
if retriable and attempt < max_attempts:
|
||||
wait_ms = self._build_retry_delay_ms(attempt, exc)
|
||||
wait_ms = self._build_retry_delay_ms(prev_delay_ms, exc, reason)
|
||||
prev_delay_ms = wait_ms
|
||||
logger.warning(
|
||||
"Transient LLM error on attempt %d/%d; retrying in %dms: %s",
|
||||
attempt,
|
||||
self.retry_max_attempts,
|
||||
max_attempts,
|
||||
wait_ms,
|
||||
_extract_error_detail(exc),
|
||||
)
|
||||
self._emit_retry_event(attempt, wait_ms, reason)
|
||||
self._emit_retry_event(attempt, wait_ms, reason, max_attempts=max_attempts)
|
||||
time.sleep(wait_ms / 1000)
|
||||
attempt += 1
|
||||
continue
|
||||
@ -361,10 +785,14 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
_extract_error_detail(exc),
|
||||
exc_info=exc,
|
||||
)
|
||||
if retriable:
|
||||
if retriable and reason != "burst_rate":
|
||||
self._record_failure()
|
||||
else:
|
||||
# Non-retriable: release the probe without recording a failure.
|
||||
# Non-retriable, OR burst_rate (a transient provider
|
||||
# slope-throttle, not "provider down"): release the half-open
|
||||
# probe without recording a failure so the circuit doesn't
|
||||
# trip and fast-fail ALL calls for the recovery window - the
|
||||
# exact self-inflicted outage #4290 is trying to prevent.
|
||||
self._release_half_open_probe()
|
||||
return self._build_user_fallback_message(exc, reason)
|
||||
|
||||
@ -383,9 +811,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
)
|
||||
|
||||
attempt = 1
|
||||
prev_delay_ms: int | None = None
|
||||
while True:
|
||||
try:
|
||||
response = await handler(request)
|
||||
response = await self._bounded_model_call(request, handler)
|
||||
self._record_success()
|
||||
return response
|
||||
except GraphBubbleUp:
|
||||
@ -394,17 +823,18 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
raise
|
||||
except Exception as exc:
|
||||
retriable, reason = self._classify_error(exc)
|
||||
max_attempts = self._max_attempts_for(exc)
|
||||
max_attempts = self._max_attempts_for(exc, reason)
|
||||
if retriable and attempt < max_attempts:
|
||||
wait_ms = self._build_retry_delay_ms(attempt, exc)
|
||||
wait_ms = self._build_retry_delay_ms(prev_delay_ms, exc, reason)
|
||||
prev_delay_ms = wait_ms
|
||||
logger.warning(
|
||||
"Transient LLM error on attempt %d/%d; retrying in %dms: %s",
|
||||
attempt,
|
||||
self.retry_max_attempts,
|
||||
max_attempts,
|
||||
wait_ms,
|
||||
_extract_error_detail(exc),
|
||||
)
|
||||
self._emit_retry_event(attempt, wait_ms, reason)
|
||||
self._emit_retry_event(attempt, wait_ms, reason, max_attempts=max_attempts)
|
||||
await asyncio.sleep(wait_ms / 1000)
|
||||
attempt += 1
|
||||
continue
|
||||
@ -414,10 +844,14 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
_extract_error_detail(exc),
|
||||
exc_info=exc,
|
||||
)
|
||||
if retriable:
|
||||
if retriable and reason != "burst_rate":
|
||||
self._record_failure()
|
||||
else:
|
||||
# Non-retriable: release the probe without recording a failure.
|
||||
# Non-retriable, OR burst_rate (a transient provider
|
||||
# slope-throttle, not "provider down"): release the half-open
|
||||
# probe without recording a failure so the circuit doesn't
|
||||
# trip and fast-fail ALL calls for the recovery window - the
|
||||
# exact self-inflicted outage #4290 is trying to prevent.
|
||||
self._release_half_open_probe()
|
||||
return self._build_user_fallback_message(exc, reason)
|
||||
|
||||
|
||||
@ -65,6 +65,63 @@ class CircuitBreakerConfig(BaseModel):
|
||||
recovery_timeout_sec: int = Field(default=60, description="Time in seconds before attempting to recover the circuit")
|
||||
|
||||
|
||||
class LlmCallConfig(BaseModel):
|
||||
"""Configuration for LLM call execution (concurrency / rate shaping).
|
||||
|
||||
Distinct from :class:`CircuitBreakerConfig` (which handles a *failing*
|
||||
provider) and from :class:`ModelConfig` (which describes model endpoints):
|
||||
these knobs shape how many LLM calls run at once and how the retry/backoff
|
||||
loop behaves. Capping concurrency caps the *slope* of the request rate,
|
||||
which is what a provider burst-rate (``limit_burst_rate``) limit fires on.
|
||||
"""
|
||||
|
||||
max_concurrent_calls: int = Field(
|
||||
default=0,
|
||||
ge=0,
|
||||
description=(
|
||||
"Process-wide cap on concurrently in-flight LLM calls. 0 disables "
|
||||
"the cap (default, preserving existing behavior). Set to a positive "
|
||||
"int to smooth provider burst-rate (limit_burst_rate) spikes by "
|
||||
"bounding the request-rate slope at the morning peak. Per-process, "
|
||||
"not per-cluster: with GATEWAY_WORKERS > 1 the aggregate cap is "
|
||||
"effectively max_concurrent_calls * GATEWAY_WORKERS (and a "
|
||||
"multi-node rollout multiplies it further), so size the per-process "
|
||||
"value accordingly and pair it with an nginx limit_req at the ingress "
|
||||
"for a true cluster-wide slope cap. Startup-only: the cap is captured "
|
||||
"at the first LLM run and frozen for the process lifetime, so editing "
|
||||
"it in config.yaml takes effect only after a gateway restart (the "
|
||||
"other llm_call.* knobs remain hot-reloadable). Freezing avoids the "
|
||||
"downscale/config-freshness races a runtime-mutable cap would "
|
||||
"introduce on a process-wide, cross-loop limiter."
|
||||
),
|
||||
)
|
||||
retry_max_attempts: int = Field(
|
||||
default=3,
|
||||
ge=1,
|
||||
description="Max LLM call attempts (1 = no retry) for retriable transient errors.",
|
||||
)
|
||||
retry_base_delay_ms: int = Field(
|
||||
default=1000,
|
||||
ge=0,
|
||||
description="Base (ms) for the decorrelated-jitter retry backoff; seeds the first retry delay.",
|
||||
)
|
||||
retry_cap_delay_ms: int = Field(
|
||||
default=8000,
|
||||
ge=0,
|
||||
description="Hard cap (ms) on any single retry backoff delay.",
|
||||
)
|
||||
burst_retry_base_delay_ms: int = Field(
|
||||
default=5000,
|
||||
ge=0,
|
||||
description=(
|
||||
"Base (ms) for the backoff when the provider returns a burst-rate "
|
||||
"(limit_burst_rate) 429. Higher than retry_base_delay_ms so the "
|
||||
"single burst retry lands after the throttle window subsides. "
|
||||
"Ignored when the provider sends Retry-After (honored verbatim)."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class LoggingEnhanceConfig(BaseModel):
|
||||
"""Request trace logging enhancement settings."""
|
||||
|
||||
@ -175,6 +232,7 @@ class AppConfig(BaseModel):
|
||||
input_polish: InputPolishConfig = Field(default_factory=InputPolishConfig, description="Pre-send input polishing configuration.")
|
||||
suggestions: SuggestionsConfig = Field(default_factory=SuggestionsConfig, description="Follow-up suggestions configuration.")
|
||||
circuit_breaker: CircuitBreakerConfig = Field(default_factory=CircuitBreakerConfig, description="LLM circuit breaker configuration")
|
||||
llm_call: LlmCallConfig = Field(default_factory=LlmCallConfig, description="LLM call execution configuration (concurrency / rate shaping)")
|
||||
channel_connections: ChannelConnectionsConfig = Field(
|
||||
default_factory=ChannelConnectionsConfig,
|
||||
description=format_field_description(
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -1985,6 +1985,49 @@ authorization:
|
||||
# # Time in seconds before attempting to recover (default: 60)
|
||||
# recovery_timeout_sec: 60
|
||||
|
||||
# ============================================================================
|
||||
# LLM Call Concurrency Configuration
|
||||
# ============================================================================
|
||||
# Cap the number of concurrently in-flight LLM calls process-wide. A provider
|
||||
# burst-rate (limit_burst_rate) error fires on the *slope* of the request rate,
|
||||
# not on a static quota - so the morning peak (e.g. 08:30) ramping from ~0 to
|
||||
# full throttle in seconds gets rejected even when total RPM is within budget.
|
||||
# Capping concurrency caps that slope. Retries alone make it worse (they add
|
||||
# demand inside the very burst being throttled); pair this cap with the
|
||||
# decorrelated-jitter backoff already built into the LLM error-handling
|
||||
# middleware, and ideally an nginx `limit_req` at the ingress.
|
||||
#
|
||||
# 0 disables the cap (default) - existing deployments see no behavior change.
|
||||
#
|
||||
# Per-process, not per-cluster: the cap bounds in-flight LLM calls within ONE
|
||||
# gateway process. With GATEWAY_WORKERS > 1 the aggregate cap across the
|
||||
# deployment is effectively `max_concurrent_calls * GATEWAY_WORKERS`, and a
|
||||
# multi-node rollout multiplies it further - size the per-process value with
|
||||
# that in mind (and pair it with an nginx `limit_req` at the ingress for a
|
||||
# true cluster-wide slope cap).
|
||||
#
|
||||
# Startup-only: the cap is captured at the first LLM run and frozen for the
|
||||
# process lifetime. Editing `max_concurrent_calls` here takes effect only after
|
||||
# a gateway restart; the other `llm_call.*` knobs below remain hot-reloadable.
|
||||
# Freezing the cap avoids the downscale and config-freshness races that a
|
||||
# runtime-mutable, process-wide/cross-loop limiter would otherwise hit (a
|
||||
# lowered cap could keep admitting queued waiters, or a stale config snapshot
|
||||
# constructed after a fresher one could restore a higher cap).
|
||||
|
||||
# llm_call:
|
||||
# # Max concurrently in-flight LLM calls across the whole process (default: 0 = disabled)
|
||||
# max_concurrent_calls: 0
|
||||
# # Max LLM call attempts for retriable transient errors, 1 = no retry (default: 3)
|
||||
# retry_max_attempts: 3
|
||||
# # Base delay (ms) for the decorrelated-jitter retry backoff (default: 1000)
|
||||
# retry_base_delay_ms: 1000
|
||||
# # Hard cap (ms) on any single retry backoff delay (default: 8000)
|
||||
# retry_cap_delay_ms: 8000
|
||||
# # Backoff base (ms) used ONLY for burst-rate (limit_burst_rate) 429s - higher
|
||||
# # than retry_base_delay_ms so the single burst retry lands after the throttle
|
||||
# # window subsides. Ignored when the provider sends Retry-After (default: 5000)
|
||||
# burst_retry_base_delay_ms: 5000
|
||||
|
||||
# ============================================================================
|
||||
# SSO / OIDC Authentication (optional)
|
||||
# ============================================================================
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user