mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(llm-error): retry provider errors reported inside a 200 stream (#6897)
* fix(llm-error): retry provider errors reported inside a 200 stream An OpenAI-compatible gateway that answers 200 early, to keep a slow request alive with SSE keepalives, reports a later provider failure (a 504, say) as an SSE error event. The openai SDK raises that as the base APIError, with no status code and no response, so _classify_error found no status, no transient class name and no busy text, and called it generic: the run failed on its first attempt. An error with no status now takes an effective status from its own contents, and the first source that yields one decides: a numeric code on the error, or a numeric code, status or status_code in its body or the body's error object; an explicit status in its message ([504], HTTP 502, status 503); or type server_error (500). Its busy check also reads each run of _, - and whitespace as one space, so rate-limit and RATE_LIMIT_EXECUTION_TIMEOUT match "rate limit". Retry budget, backoff and circuit breaker then treat it as the equivalent HTTP error. Errors that carry a status are classified exactly as before, and the quota, auth and burst-rate checks still run first. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(llm-error): note how status-less errors are classified Add one sentence to item 9 of the middlewares guide, and an llm entry under Fixed in both changelogs, for the retry of provider errors reported inside a 200 stream. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(llm-error): pin the in-band status source order and leftmost match Two rules of _extract_in_band_status had no test: a numeric status field decides before the message is read, and within the message the leftmost in-range status wins. Either rule could be inverted with every test still green. Add one stays-generic case for each: code 400 with "[503]" in the message, and "HTTP 400 after [503]". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(llm-error): do not read an array index as an in-band status The bracketed alternative of the in-band status regex had no lookbehind, so `messages[502]` or `tools[0][503]` matched and their number decided the error's status. The bracketed form now starts with the lookbehind `(?<![\w\]])`: a `[` directly after a word character or a `]` opens an index, not a status. The `http` and `status` alternatives are unchanged; `[504]: ...` in running text still counts. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(llm-error): let an in-band status preempt the separator busy check The separator-normalised busy check ran for every status-less error, even when the in-band lookup had already found a status: "rate-limit header malformed" with `code: 400` was retried as busy, while HTTP 400 is not. It now runs only when no source yields a status, so an in-band 4xx decides like an HTTP one. The same text with no status anywhere still counts as busy. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(llm-error): let only 4xx and 5xx values decide the in-band status The in-band lookup accepted 100-599 values from its first two sources, so a 1xx-3xx status could decide the classification: `HTTP 200` in a message whose server_error type means 500, or `code: 200` alongside `status: 503`. Only 400-599 now count: `_as_http_status` skips a smaller value as it skips a non-numeric one, so the candidate loop moves on, and an out-of-range message match is skipped so the scan continues. An error that carries a status stays as it was, and the server_error source is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(llm-error): describe the narrowed in-band status rules The guide, the CHANGELOG and the CHANGELOG_zh now describe the three narrowed rules: only a 4xx or 5xx value decides, an index such as `messages[502]` is not a status, and the separator-matched busy check runs only when no source yields a status. Only the sentences that describe the affected forms changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(llm-error): pin the remaining in-band status guards Cover the nested error.status_code field, a code above 599 that yields to a later status field, a code written in non-ASCII digits, and an in-band 4xx whose plain busy text still decides, as it does for an HTTP 4xx. These pin current behaviour; no code changes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(llm-error): say which in-band fields are read on the error itself Only `code` is read on the exception; `status` and `status_code` are read in its body and in the body's `error` object. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(changelog): link the in-stream retry entry to #6897 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(llm-error): require a separator and ASCII digits in an in-band status The message regex read `status503` and `statuscode503` as a status because the separator after `status` was optional, and it read Arabic-Indic or fullwidth digits because `\d` and `int()` accept every Unicode decimal digit. A status-less error with such text was retried as a 503. The `status` form now needs spaces or tabs, or a `:` or `=`, before the number, and all three forms read only ASCII digits. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(llm-error): pin the retry of an in-stream error after partial output An `error` event that follows content chunks is retried like one that comes first: the whole call re-runs and its result holds only the new answer, while the tokens the first attempt already streamed to the callbacks stay delivered, as with a mid-stream `ReadError`. The new test pins that outcome, and a comment in `_classify_error` notes it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Handle transient errors based on status codes Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
ajayr
GitHub
parent
585f87d3b3
commit
51bc0f160e
15
CHANGELOG.md
15
CHANGELOG.md
@ -1247,6 +1247,20 @@ This release closes that milestone with **621 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **llm:** A provider error that arrives inside a streaming response that has
|
||||
already returned `200` is now retried like the same error sent as an HTTP
|
||||
status. An OpenAI-compatible gateway that answers `200` early, to keep slow
|
||||
requests alive with SSE keepalives, reports a later 504 as an SSE `error`
|
||||
event; the openai SDK raises that as an `APIError` with no status, so
|
||||
`LLMErrorHandlingMiddleware` classified it as generic and the run failed on
|
||||
the first attempt. Such a status-less error now takes a 4xx or 5xx status
|
||||
from a numeric `code` or `status` field, an explicit `[504]`, `HTTP 502` or
|
||||
`status 503` in its message (not an index such as `messages[502]`), or
|
||||
`type: server_error` (500); only when none yields one does its busy check
|
||||
also match spellings such as `rate-limit` and `RATE_LIMIT_…`. Retry budget,
|
||||
backoff and circuit breaker treat it as the equivalent HTTP error; errors
|
||||
that carry a status are classified as before. (issue #6896, [#6897])
|
||||
|
||||
- **channels:** Runs started from an IM channel no longer die with the Gateway
|
||||
instance that received the message. The manager holds `runs.wait` /
|
||||
`runs.stream` open for the whole run, and the Gateway cancels a run whose
|
||||
@ -11732,3 +11746,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6852]: https://github.com/bytedance/deer-flow/pull/6852
|
||||
[#6865]: https://github.com/bytedance/deer-flow/pull/6865
|
||||
[#6876]: https://github.com/bytedance/deer-flow/pull/6876
|
||||
[#6897]: https://github.com/bytedance/deer-flow/pull/6897
|
||||
|
||||
@ -1002,6 +1002,16 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **llm：** 流式响应已返回 `200` 之后才在流内报告的 provider 错误，现在会像以 HTTP
|
||||
状态码返回的同一错误一样重试。为了用 SSE keepalive 维持慢请求而提前返回 `200` 的
|
||||
OpenAI 兼容网关，会把之后发生的 504 作为 SSE `error` 事件发出；openai SDK 将其抛出为
|
||||
不带状态码的 `APIError`，因此 `LLMErrorHandlingMiddleware` 此前把它归为普通错误，
|
||||
运行在第一次尝试后即失败。现在这类不带状态码的错误会依次从数字型 `code` 或 `status`
|
||||
字段、消息中明确写出的 `[504]`、`HTTP 502` 或 `status 503`（`messages[502]` 这类
|
||||
数组下标不算）、以及 `type: server_error`（视为 500）中取得 4xx 或 5xx 状态码；都
|
||||
取不到时，繁忙判断才会额外匹配 `rate-limit`、`RATE_LIMIT_…` 等写法。重试预算、退避
|
||||
和熔断器按对应的 HTTP 错误处理；带状态码的错误分类不变。 (issue #6896, [#6897])
|
||||
|
||||
- **渠道：** 从 IM 渠道发起的运行不再随接收消息的 Gateway 实例一起消亡。manager 在整个
|
||||
运行期间保持 `runs.wait` / `runs.stream` 请求打开，而 Gateway 会在创建请求断开时取消
|
||||
运行，除非该请求声明了 `on_disconnect: continue`；因此 leader 切换、滚动重启或接收
|
||||
@ -9358,3 +9368,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
|
||||
[#6852]: https://github.com/bytedance/deer-flow/pull/6852
|
||||
[#6865]: https://github.com/bytedance/deer-flow/pull/6865
|
||||
[#6876]: https://github.com/bytedance/deer-flow/pull/6876
|
||||
[#6897]: https://github.com/bytedance/deer-flow/pull/6897
|
||||
|
||||
@ -81,7 +81,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
|
||||
their narrower discovery allowlists never rebuild the shared thread view or
|
||||
force eager sandbox acquisition.
|
||||
8. **DanglingToolCallMiddleware** - Injects placeholder ToolMessages for AIMessage tool_calls that lack responses (e.g., user interruption), preserving raw provider tool-call payloads in `additional_kwargs["tool_calls"]`; malformed tool-call names and arguments are sanitized in the model-bound request so strict OpenAI-compatible providers do not reject the next request
|
||||
9. **LLMErrorHandlingMiddleware** - Assistant error recovery. HTTP 529 is transient by status; retries re-enter admission. Honor numeric/date hints up to 24h via `utils.retry_after`, otherwise jitter. Only current circuit-generation owners settle/release sync/async half-open probes. Cancellation propagates without retry/failure accounting.
|
||||
9. **LLMErrorHandlingMiddleware** - Assistant error recovery. HTTP 529 is transient by status; retries re-enter admission. Honor numeric/date hints up to 24h via `utils.retry_after`, otherwise jitter. Only current circuit-generation owners settle/release sync/async half-open probes. Cancellation propagates without retry/failure accounting. A status-less (in-stream) error takes a 4xx/5xx status from a numeric `code`/`status`, a `[504]`/`HTTP 502` in its text, or `type: server_error`; else its busy match reads `_`/`-` as spaces.
|
||||
10. **ArtifactResolutionMiddleware** - See [tool artifacts](TOOL_ARTIFACTS.md).
|
||||
11. **Authorization / GuardrailMiddleware** - Up to two independent pre-tool-call gates run here. When `authorization.enabled`, the `AuthorizationProvider` instance already used for Layer 1 capability filtering is wrapped by `GuardrailAuthorizationAdapter` and reused for Layer 2 execution checks. A generated `tool_search` bypasses the adapter's second provider call only when the current build has a concrete deferred setup; its catalog was already filtered by Layer 1, and an ordinary same-named tool without that deferred setup receives no exemption. When `guardrails.enabled`, the explicitly configured `GuardrailProvider` is appended after authorization and still evaluates every call, including `tool_search`. Authorization therefore runs outermost and can deny before an external guardrail call; both use the existing middleware's fail-closed, audit, sync/async, and error-`ToolMessage` behavior. See the authorization RFC and [docs/GUARDRAILS.md](../../../../../docs/GUARDRAILS.md).
|
||||
|
||||
|
||||
@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from collections import deque
|
||||
@ -169,6 +170,22 @@ _REASON_RETRY_BUDGETS: dict[str, int] = {
|
||||
# streamed response (HTTP 200) yields no generation chunk at all.
|
||||
_EMPTY_STREAM_MARKER = "no generations found in stream"
|
||||
|
||||
# An HTTP status written into the text of an error that carries no status:
|
||||
# ``[504]`` - but not an index such as ``messages[502]``, which the bracketed
|
||||
# form's lookbehind rejects; ``HTTP 502``, ``(HTTP 502)``, ``HTTP/1.1 502``;
|
||||
# ``status 503``, ``status: 503``, ``status_code=503``, ``"status": 503``. The
|
||||
# keyword is required and must be separated from the number - by spaces or tabs
|
||||
# after ``HTTP``, and also by ``:`` or ``=`` after ``status`` - so a bare number
|
||||
# (``130504 tokens``), one on the next line or one run into the keyword
|
||||
# (``status503``) never counts. Digits are ASCII only: ``\d`` and ``int()``
|
||||
# also accept other scripts' digits. The spaces before and after ``:``/``=``
|
||||
# are separate runs, which keeps matching linear in the length of the message.
|
||||
_IN_BAND_STATUS_RE = re.compile(r"(?<![\w\]])\[([0-9]{3})\]|\bhttp(?:/[0-9](?:\.[0-9])?)?[ \t]+([0-9]{3})\b|\bstatus(?:[ _]?code)?(?:[\"']?[ \t]*[:=][ \t]*|[\"']?[ \t]+)([0-9]{3})\b")
|
||||
|
||||
# Runs of ``_``, ``-`` and whitespace. A status-less error's busy check reads each
|
||||
# run as one space, so ``rate-limit`` and ``RATE_LIMIT_...`` match ``rate limit``.
|
||||
_SEPARATOR_RUN_RE = re.compile(r"[\s_\-]+")
|
||||
|
||||
# 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,
|
||||
@ -611,6 +628,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
lowered = detail.lower()
|
||||
error_code = _extract_error_code(exc)
|
||||
status_code = _extract_status_code(exc)
|
||||
# A provider that has already answered ``200`` reports a later failure
|
||||
# inside the stream; the openai SDK raises it as a bare ``APIError`` with
|
||||
# no status code and no response. Only such status-less errors take a
|
||||
# status from their own fields and text (``_extract_in_band_status``).
|
||||
# As with a mid-stream ``ReadError``, a retry re-runs the whole call:
|
||||
# tokens the failed attempt already streamed are not taken back.
|
||||
status_less = status_code is None
|
||||
|
||||
if isinstance(exc, EmptyModelResponseError):
|
||||
return True, "empty_response"
|
||||
@ -659,10 +683,16 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
# Same upstream-payload glitch, same retry/backoff path.
|
||||
if isinstance(exc, ValueError) and _EMPTY_STREAM_MARKER in lowered:
|
||||
return True, "transient"
|
||||
if status_less:
|
||||
status_code = _extract_in_band_status(exc)
|
||||
if status_code is not None and 500 <= status_code <= 599:
|
||||
return True, "transient"
|
||||
if status_code in _RETRIABLE_STATUS_CODES:
|
||||
return True, "transient"
|
||||
if _matches_any(lowered, _BUSY_PATTERNS):
|
||||
return True, "busy"
|
||||
if status_code is None and (_matches_any(_normalize_separators(detail), _BUSY_PATTERNS) or _matches_any(_normalize_separators(str(error_code)), _BUSY_PATTERNS)):
|
||||
return True, "busy"
|
||||
|
||||
return False, "generic"
|
||||
|
||||
@ -1078,6 +1108,57 @@ def _extract_status_code(exc: BaseException) -> int | None:
|
||||
return status if isinstance(status, int) else None
|
||||
|
||||
|
||||
def _extract_in_band_status(exc: BaseException) -> int | None:
|
||||
"""Return the HTTP status that a status-less error carries in its contents.
|
||||
|
||||
Three sources, in order; the first that yields a status decides:
|
||||
a numeric ``code`` on the error, or a numeric ``code``/``status``/
|
||||
``status_code`` in its body or the body's ``error`` object; an explicit
|
||||
status in the message (``_IN_BAND_STATUS_RE``, leftmost match in 400-599);
|
||||
and ``type == "server_error"``, which OpenAI-compatible APIs use for 5xx
|
||||
failures (500).
|
||||
"""
|
||||
body = getattr(exc, "body", None)
|
||||
body = body if isinstance(body, dict) else {}
|
||||
error = body.get("error")
|
||||
error = error if isinstance(error, dict) else {}
|
||||
for value in (
|
||||
getattr(exc, "code", None),
|
||||
body.get("code"),
|
||||
error.get("code"),
|
||||
body.get("status"),
|
||||
body.get("status_code"),
|
||||
error.get("status"),
|
||||
error.get("status_code"),
|
||||
):
|
||||
status = _as_http_status(value)
|
||||
if status is not None:
|
||||
return status
|
||||
for match in _IN_BAND_STATUS_RE.finditer(_extract_error_detail(exc).lower()):
|
||||
status = int(next(group for group in match.groups() if group))
|
||||
if 400 <= status <= 599:
|
||||
return status
|
||||
for value in (getattr(exc, "type", None), body.get("type"), error.get("type")):
|
||||
if isinstance(value, str) and value.strip().lower() == "server_error":
|
||||
return 500
|
||||
return None
|
||||
|
||||
|
||||
def _as_http_status(value: Any) -> int | None:
|
||||
"""Return ``value`` as an HTTP status if it is an int or a three-digit string in 400-599."""
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
value = int(text) if len(text) == 3 and text.isascii() and text.isdigit() else None
|
||||
if isinstance(value, int) and not isinstance(value, bool) and 400 <= value <= 599:
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_separators(text: str) -> str:
|
||||
"""Lower-case ``text`` and replace each run of ``_``, ``-`` and whitespace with one space."""
|
||||
return _SEPARATOR_RUN_RE.sub(" ", text.lower())
|
||||
|
||||
|
||||
def _extract_retry_after_ms(exc: BaseException) -> int | None:
|
||||
response = getattr(exc, "response", None)
|
||||
headers = getattr(response, "headers", None)
|
||||
|
||||
@ -2,6 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
@ -10,18 +11,23 @@ from email.utils import format_datetime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import openai
|
||||
import pytest
|
||||
from langchain.agents import create_agent
|
||||
from langchain.agents.middleware.types import ModelResponse
|
||||
from langchain_core.callbacks import AsyncCallbackHandler
|
||||
from langchain_core.language_models import BaseChatModel
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
from langchain_core.tools import StructuredTool
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langgraph.errors import GraphBubbleUp
|
||||
|
||||
from deerflow.agents.middlewares.llm_error_handling_middleware import (
|
||||
EmptyModelResponseError,
|
||||
LLMErrorHandlingMiddleware,
|
||||
_extract_in_band_status,
|
||||
_extract_retry_after_ms,
|
||||
)
|
||||
from deerflow.config.app_config import AppConfig, LlmCallConfig
|
||||
@ -3058,3 +3064,385 @@ def test_circuit_recovery_survives_a_backward_wall_clock_step(monkeypatch: pytes
|
||||
assert middleware._check_circuit() is False
|
||||
assert middleware._circuit_state == "half_open"
|
||||
assert middleware._circuit_probe_in_flight is True
|
||||
|
||||
|
||||
# ---------- Provider errors reported inside a 200 stream ----------
|
||||
#
|
||||
# A provider (or a gateway in front of one) that has already answered ``200``
|
||||
# reports a later failure as an SSE ``error`` event. The openai SDK raises it as
|
||||
# the base ``APIError``: ``body`` is the event's ``error`` object, ``code`` and
|
||||
# ``type`` are copied from it, and there is no status code and no response.
|
||||
# ``_in_stream_error`` builds exactly that object.
|
||||
|
||||
# What OmniRoute (an OpenAI-compatible gateway) sent after committing to ``200``
|
||||
# with SSE keepalives: its combo's error, wrapping the target's own 504 frame.
|
||||
_OMNIROUTE_COMBO_504 = (
|
||||
"openai-compatible-chat-conn:7b9b6797/glm-5.3: provider — data: "
|
||||
'{"error":{"message":"[504]: Request exceeded OmniRoute\'s local rate-limit execution expiration '
|
||||
"(legacy resilienceSettings.requestQueue.maxWaitMs=15000ms) for openai-compatible-chat-conn:7b9b6797/glm-5.3. "
|
||||
"Bottleneck applies this deadline only after dispatch; it does not bound queue wait and is not an "
|
||||
'upstream-generated timeout.","type":"server_error","code":"RATE_LIMIT_EXECUTION_TIMEOUT"}}'
|
||||
)
|
||||
_OMNIROUTE_TARGET_504 = (
|
||||
"[504]: Request exceeded OmniRoute's local rate-limit execution expiration "
|
||||
"(legacy resilienceSettings.requestQueue.maxWaitMs=15000ms) for openai-compatible-chat-conn:7b9b6797/glm-5.3. "
|
||||
"Bottleneck applies this deadline only after dispatch; it does not bound queue wait and is not an "
|
||||
"upstream-generated timeout."
|
||||
)
|
||||
_STREAM_REQUEST = httpx.Request("POST", "https://llm.invalid/v1/chat/completions")
|
||||
|
||||
|
||||
def _in_stream_error(message: str, **error_fields: Any) -> openai.APIError:
|
||||
"""Build the ``APIError`` the openai SDK raises for an SSE ``error`` event."""
|
||||
return openai.APIError(message, _STREAM_REQUEST, body={"message": message, **error_fields})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(_in_stream_error(_OMNIROUTE_COMBO_504, type="server_error", code="gateway_timeout"), id="omniroute-combo"),
|
||||
pytest.param(_in_stream_error(_OMNIROUTE_TARGET_504, type="server_error", code="RATE_LIMIT_EXECUTION_TIMEOUT"), id="omniroute-target"),
|
||||
],
|
||||
)
|
||||
def test_omniroute_in_stream_504_is_transient(exc: openai.APIError) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == (True, "transient")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"message",
|
||||
[
|
||||
pytest.param("[504]: upstream timed out", id="bracketed"),
|
||||
pytest.param("upstream answered HTTP 502", id="http"),
|
||||
pytest.param("upstream failed (HTTP 502)", id="http-in-parens"),
|
||||
pytest.param("HTTP/1.1 502 Bad Gateway", id="http-1.1"),
|
||||
pytest.param("HTTP/2 503 from upstream", id="http-2"),
|
||||
pytest.param("upstream status 503", id="status-space"),
|
||||
pytest.param("upstream status: 503", id="status-colon"),
|
||||
pytest.param("upstream status=503", id="status-equals"),
|
||||
pytest.param("upstream status code: 503", id="status-code"),
|
||||
pytest.param("upstream status_code=503", id="status_code"),
|
||||
pytest.param('upstream replied {"status": 503}', id="json-status"),
|
||||
pytest.param('upstream replied {"status":503}', id="json-status-without-space"),
|
||||
pytest.param('upstream replied {"statusCode": 503}', id="json-statusCode"),
|
||||
pytest.param("gateway [999] relayed HTTP 503", id="out-of-range-match-skipped"),
|
||||
pytest.param("error: [503] overloaded", id="bracketed-in-running-text"),
|
||||
pytest.param("openai-compatible-chat-conn:7b9b6797/glm-5.3: provider — [504]: upstream timed out (HTTP 504)", id="omniroute-combo-text"),
|
||||
pytest.param("upstream answered HTTP 200, then failed with [503]", id="status-200-skipped-for-later-match"),
|
||||
],
|
||||
)
|
||||
def test_in_stream_error_takes_its_status_from_the_message(message: str) -> None:
|
||||
assert _build_middleware()._classify_error(_in_stream_error(message)) == (True, "transient")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error_fields",
|
||||
[
|
||||
pytest.param({"code": 503}, id="int-code"),
|
||||
pytest.param({"code": "503"}, id="str-code"),
|
||||
pytest.param({"code": " 503 "}, id="str-code-with-whitespace"),
|
||||
pytest.param({"code": 503, "status": "UNAVAILABLE"}, id="google-style"),
|
||||
pytest.param({"status": 502}, id="status"),
|
||||
pytest.param({"status_code": 502}, id="status_code"),
|
||||
pytest.param({"error": {"code": 503}}, id="nested-code"),
|
||||
pytest.param({"error": {"status": 502}}, id="nested-status"),
|
||||
pytest.param({"error": {"status_code": 502}}, id="nested-status_code"),
|
||||
pytest.param({"type": "server_error"}, id="server-error-type"),
|
||||
pytest.param({"type": "SERVER_ERROR"}, id="server-error-type-upper-case"),
|
||||
pytest.param({"error": {"type": "server_error"}}, id="nested-server-error-type"),
|
||||
pytest.param({"code": 200, "type": "server_error"}, id="code-200-skipped-for-server-error-type"),
|
||||
pytest.param({"code": 200, "status": 503}, id="code-200-skipped-for-status"),
|
||||
pytest.param({"code": 999, "status": 503}, id="code-999-skipped-for-status"),
|
||||
],
|
||||
)
|
||||
def test_in_stream_error_takes_its_status_from_structured_fields(error_fields: dict[str, Any]) -> None:
|
||||
exc = _in_stream_error("Upstream request failed.", **error_fields)
|
||||
assert _build_middleware()._classify_error(exc) == (True, "transient")
|
||||
|
||||
|
||||
def test_in_stream_non_error_status_does_not_hide_server_error_type() -> None:
|
||||
exc = _in_stream_error("stream aborted after HTTP 200", type="server_error")
|
||||
assert _build_middleware()._classify_error(exc) == (True, "transient")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(_in_stream_error("stream aborted after HTTP 200"), id="message-http-200"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code=101), id="code-101"),
|
||||
pytest.param(_in_stream_error("upstream redirected: status 302"), id="message-status-302"),
|
||||
],
|
||||
)
|
||||
def test_in_band_lookup_skips_non_error_statuses(exc: openai.APIError) -> None:
|
||||
assert _extract_in_band_status(exc) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(_in_stream_error("upstream rate-limit hit"), id="hyphenated-message"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code="rate_limit_exceeded"), id="snake-case-code"),
|
||||
pytest.param(_in_stream_error("Request exceeded the local execution expiration.", code="RATE_LIMIT_EXECUTION_TIMEOUT"), id="upper-snake-case-code"),
|
||||
pytest.param(RuntimeError("Codex API response.failed (rate_limit_exceeded): slow down"), id="codex-response-failed"),
|
||||
pytest.param(_in_stream_error("rate-limit header malformed"), id="separator-busy-without-a-status"),
|
||||
],
|
||||
)
|
||||
def test_status_less_error_matches_busy_patterns_across_separators(exc: Exception) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == (True, "busy")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(
|
||||
_in_stream_error("The response was filtered due to the prompt triggering content management policy.", type="invalid_request_error", code="content_filter"),
|
||||
id="content-filter",
|
||||
),
|
||||
pytest.param(
|
||||
_in_stream_error("This model's maximum context length is 131072 tokens. However, your messages resulted in 130504 tokens.", type="invalid_request_error", code="context_length_exceeded"),
|
||||
id="context-length",
|
||||
),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code=400), id="code-400"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code=400, type="server_error"), id="code-400-beats-server-error-type"),
|
||||
pytest.param(_in_stream_error("upstream returned HTTP 400", type="server_error"), id="message-400-beats-server-error-type"),
|
||||
pytest.param(_in_stream_error("upstream failed [999]"), id="out-of-range-status"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code=True), id="bool-code"),
|
||||
pytest.param(_in_stream_error("upstream failed: [5040]"), id="bracketed-four-digits"),
|
||||
pytest.param(_in_stream_error("upstream failed: [50]"), id="bracketed-two-digits"),
|
||||
pytest.param(_in_stream_error("upstream failed: HTTP502"), id="http-without-space"),
|
||||
pytest.param(_in_stream_error("upstream failed: https://host:5020"), id="url-port"),
|
||||
pytest.param(_in_stream_error("upstream failed: status ok"), id="status-word"),
|
||||
pytest.param(_in_stream_error("upstream failed: status 5030"), id="status-four-digits"),
|
||||
pytest.param(_in_stream_error("upstream failed: status\n503"), id="status-number-on-next-line"),
|
||||
pytest.param(_in_stream_error("upstream failed: status503"), id="status-without-separator"),
|
||||
pytest.param(_in_stream_error("upstream failed: statuscode503"), id="statuscode-without-separator"),
|
||||
pytest.param(_in_stream_error("upstream failed: status_code503"), id="status_code-without-separator"),
|
||||
# Arabic-Indic and fullwidth digits for 503: int() accepts them, the status regex must not.
|
||||
pytest.param(_in_stream_error("upstream answered HTTP \u0665\u0660\u0663"), id="http-arabic-indic-digits"),
|
||||
pytest.param(_in_stream_error("upstream answered HTTP \uff15\uff10\uff13"), id="http-fullwidth-digits"),
|
||||
pytest.param(_in_stream_error("upstream failed: [\u0665\u0660\u0663]"), id="bracketed-arabic-indic-digits"),
|
||||
pytest.param(_in_stream_error("upstream status: \u0665\u0660\u0663"), id="status-arabic-indic-digits"),
|
||||
pytest.param(_in_stream_error("upstream failed: error 504"), id="number-without-keyword"),
|
||||
pytest.param(_in_stream_error("upstream said [503]", code=400), id="code-400-beats-message-503"),
|
||||
pytest.param(_in_stream_error("upstream returned HTTP 400 after [503]"), id="leftmost-message-status-decides"),
|
||||
pytest.param(_in_stream_error("invalid value at messages[502].content"), id="message-index"),
|
||||
pytest.param(_in_stream_error("invalid tool call at tools[0][503].function"), id="nested-index"),
|
||||
pytest.param(_in_stream_error("rate-limit header malformed", code=400), id="code-400-beats-separator-busy"),
|
||||
pytest.param(_in_stream_error("HTTP 400: rate-limit header malformed"), id="message-400-beats-separator-busy"),
|
||||
],
|
||||
)
|
||||
def test_in_stream_error_without_a_retriable_signal_stays_generic(exc: openai.APIError) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == (False, "generic")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(openai.APIError("An error occurred during streaming", _STREAM_REQUEST, body="upstream exploded"), id="string-body"),
|
||||
pytest.param(openai.APIError("An error occurred during streaming", _STREAM_REQUEST, body=["upstream", "exploded"]), id="list-body"),
|
||||
pytest.param(openai.APIError("An error occurred during streaming", _STREAM_REQUEST, body=None), id="no-body"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", error="upstream timeout"), id="string-error-field"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code={"value": 503}), id="dict-code"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", status=["503"]), id="list-status"),
|
||||
# Arabic-Indic digits for 503: int() accepts them, the status check must not.
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code="\u0665\u0660\u0663"), id="non-ascii-digit-code"),
|
||||
],
|
||||
)
|
||||
def test_in_stream_error_with_a_malformed_body_stays_generic(exc: openai.APIError) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == (False, "generic")
|
||||
|
||||
|
||||
def test_in_stream_status_parsing_stays_fast_on_a_long_message() -> None:
|
||||
# Spaces before and after ``:``/``=`` must be separate runs in the status
|
||||
# regex; two adjacent ``[ \t]*`` would backtrack quadratically here.
|
||||
exc = _in_stream_error("status" + " " * 50_000 + "x")
|
||||
started = time.perf_counter()
|
||||
assert _build_middleware()._classify_error(exc) == (False, "generic")
|
||||
assert time.perf_counter() - started < 1.0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("exc", "expected"),
|
||||
[
|
||||
pytest.param(
|
||||
_in_stream_error("You exceeded your current quota, please check your plan and billing details.", type="server_error", code="insufficient_quota"),
|
||||
(False, "quota"),
|
||||
id="quota",
|
||||
),
|
||||
pytest.param(_in_stream_error("Unauthorized", code=401), (False, "auth"), id="auth"),
|
||||
pytest.param(_in_stream_error("Upstream request failed.", code="limit_burst_rate"), (True, "burst_rate"), id="burst-rate"),
|
||||
],
|
||||
)
|
||||
def test_in_stream_quota_auth_and_burst_rate_keep_precedence(exc: openai.APIError, expected: tuple[bool, str]) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(_in_stream_error("rate limit exceeded", code=400), id="in-band-400"),
|
||||
pytest.param(FakeError("rate limit exceeded", status_code=400), id="http-400"),
|
||||
],
|
||||
)
|
||||
def test_in_band_4xx_keeps_the_plain_busy_match(exc: Exception) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == (True, "busy")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
pytest.param(FakeError("rate-limit exceeded", status_code=400), id="hyphenated-busy-text"),
|
||||
pytest.param(FakeError("upstream said [503]", status_code=400), id="bracketed-status"),
|
||||
pytest.param(FakeError("upstream said [503]", status_code=200), id="stream-status-200"),
|
||||
],
|
||||
)
|
||||
def test_error_with_a_status_ignores_in_band_signals(exc: FakeError) -> None:
|
||||
assert _build_middleware()._classify_error(exc) == (False, "generic")
|
||||
|
||||
|
||||
def test_async_in_stream_504_retries_with_the_full_budget_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
|
||||
attempts = 0
|
||||
events: list[dict] = []
|
||||
|
||||
async def fake_sleep(_delay: float) -> None:
|
||||
return None
|
||||
|
||||
async def fake_emit_custom_event(payload, *, writer):
|
||||
writer(payload)
|
||||
|
||||
async def handler(_request) -> AIMessage:
|
||||
nonlocal attempts
|
||||
attempts += 1
|
||||
if attempts < 3:
|
||||
raise _in_stream_error(_OMNIROUTE_COMBO_504, type="server_error", code="gateway_timeout")
|
||||
return AIMessage(content="recovered")
|
||||
|
||||
monkeypatch.setattr("asyncio.sleep", fake_sleep)
|
||||
monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: events.append)
|
||||
monkeypatch.setattr("deerflow.agents.middlewares.llm_error_handling_middleware.aemit_custom_event", fake_emit_custom_event)
|
||||
|
||||
result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))
|
||||
|
||||
assert isinstance(result, AIMessage)
|
||||
assert result.content == "recovered"
|
||||
assert attempts == 3
|
||||
assert [(event["type"], event["reason"], event["max_attempts"]) for event in events] == [("llm_retry", "transient", 3)] * 2
|
||||
|
||||
|
||||
def test_sync_in_stream_504_exhausts_the_full_budget_into_the_transient_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
|
||||
attempts = 0
|
||||
|
||||
def handler(_request) -> AIMessage:
|
||||
nonlocal attempts
|
||||
attempts += 1
|
||||
raise _in_stream_error(_OMNIROUTE_COMBO_504, type="server_error", code="gateway_timeout")
|
||||
|
||||
monkeypatch.setattr("time.sleep", lambda _delay: None)
|
||||
|
||||
result = middleware.wrap_model_call(SimpleNamespace(), handler)
|
||||
|
||||
assert attempts == 3
|
||||
assert isinstance(result, AIMessage)
|
||||
assert "temporarily unavailable" in str(result.content)
|
||||
assert result.additional_kwargs["deerflow_error_fallback"] is True
|
||||
assert result.additional_kwargs["error_type"] == "APIError"
|
||||
assert result.additional_kwargs["error_reason"] == "transient"
|
||||
assert result.additional_kwargs["error_detail"] == _OMNIROUTE_COMBO_504
|
||||
|
||||
|
||||
def test_exhausted_in_stream_5xx_failures_open_the_circuit_breaker(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr("time.sleep", lambda _delay: None)
|
||||
middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=10, retry_cap_delay_ms=10, circuit_failure_threshold=2)
|
||||
calls = 0
|
||||
|
||||
def handler(_request) -> AIMessage:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
raise _in_stream_error("Upstream request failed.", type="server_error")
|
||||
|
||||
middleware.wrap_model_call(SimpleNamespace(), handler)
|
||||
assert middleware._circuit_failure_count == 1
|
||||
middleware.wrap_model_call(SimpleNamespace(), handler)
|
||||
assert middleware._circuit_state == "open"
|
||||
|
||||
result = middleware.wrap_model_call(SimpleNamespace(), handler)
|
||||
|
||||
assert calls == 4
|
||||
assert result.additional_kwargs["error_reason"] == "circuit_open"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_in_stream_error_from_a_real_openai_stream_is_retried() -> None:
|
||||
"""End to end: the first stream answers ``200`` and then carries the error event."""
|
||||
middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
|
||||
requests: list[httpx.Request] = []
|
||||
|
||||
def chunk(delta: dict[str, Any], finish_reason: str | None = None) -> dict[str, Any]:
|
||||
return {"id": "chatcmpl-retry", "object": "chat.completion.chunk", "created": 0, "model": "glm-5.3", "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}]}
|
||||
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
requests.append(request)
|
||||
if len(requests) == 1:
|
||||
error = {"error": {"message": _OMNIROUTE_COMBO_504, "type": "server_error", "code": "gateway_timeout"}}
|
||||
body = f": keepalive\n\ndata: {json.dumps(error)}\n\n"
|
||||
else:
|
||||
body = "".join(f"data: {json.dumps(event)}\n\n" for event in (chunk({"role": "assistant", "content": "recovered"}), chunk({}, "stop"))) + "data: [DONE]\n\n"
|
||||
return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=body)
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
|
||||
model = ChatOpenAI(model="glm-5.3", api_key="offline-test-key", base_url="https://llm.invalid/v1", streaming=True, max_retries=0, http_async_client=client)
|
||||
|
||||
async def handler(_request) -> AIMessage:
|
||||
return await model.ainvoke([HumanMessage(content="hello")])
|
||||
|
||||
result = await middleware.awrap_model_call(SimpleNamespace(), handler)
|
||||
|
||||
assert len(requests) == 2
|
||||
assert isinstance(result, AIMessage)
|
||||
assert result.content == "recovered"
|
||||
assert "deerflow_error_fallback" not in result.additional_kwargs
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_in_stream_error_after_partial_output_is_retried_from_the_start() -> None:
|
||||
"""An ``error`` event after content chunks retries the whole call.
|
||||
|
||||
The result holds only the retried answer, but the tokens the first attempt
|
||||
already streamed reached the callbacks (which feed a live stream) and are
|
||||
not taken back. Mid-stream ``ReadError`` retries behave the same way; this
|
||||
pins that accepted outcome.
|
||||
"""
|
||||
middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
|
||||
requests: list[httpx.Request] = []
|
||||
streamed: list[str] = []
|
||||
|
||||
class RecordTokens(AsyncCallbackHandler):
|
||||
async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
|
||||
streamed.append(token)
|
||||
|
||||
def chunk(delta: dict[str, Any], finish_reason: str | None = None) -> dict[str, Any]:
|
||||
return {"id": "chatcmpl-retry", "object": "chat.completion.chunk", "created": 0, "model": "glm-5.3", "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}]}
|
||||
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
requests.append(request)
|
||||
if len(requests) == 1:
|
||||
error = {"error": {"message": "[504]: upstream timed out", "type": "server_error", "code": "gateway_timeout"}}
|
||||
body = "".join(f"data: {json.dumps(event)}\n\n" for event in (chunk({"role": "assistant", "content": "partial answ"}), error))
|
||||
else:
|
||||
body = "".join(f"data: {json.dumps(event)}\n\n" for event in (chunk({"role": "assistant", "content": "recovered"}), chunk({}, "stop"))) + "data: [DONE]\n\n"
|
||||
return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=body)
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
|
||||
model = ChatOpenAI(model="glm-5.3", api_key="offline-test-key", base_url="https://llm.invalid/v1", streaming=True, max_retries=0, http_async_client=client)
|
||||
|
||||
async def handler(_request) -> AIMessage:
|
||||
return await model.ainvoke([HumanMessage(content="hello")], config={"callbacks": [RecordTokens()]})
|
||||
|
||||
result = await middleware.awrap_model_call(SimpleNamespace(), handler)
|
||||
|
||||
assert len(requests) == 2
|
||||
assert isinstance(result, AIMessage)
|
||||
assert result.content == "recovered"
|
||||
assert "deerflow_error_fallback" not in result.additional_kwargs
|
||||
assert "".join(streamed) == "partial answrecovered"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user