fix(llm): retain finite provider request timeouts

This commit is contained in:
jxxghp
2026-11-07 05:52:11 +08:00
parent 0851b373d2
commit 263ae3fbc7
3 changed files with 154 additions and 4 deletions

View File

@@ -25,6 +25,8 @@ if TYPE_CHECKING:
_TEMPERATURE_UNSET = object()
# SDK 的请求级 None 会关闭 HTTP 客户端默认超时;非 Agent 的同步调用也必须有网络等待上限。
_LLM_REQUEST_TIMEOUT = 180.0
class LLMTestError(RuntimeError):
@@ -198,23 +200,27 @@ def _build_httpx_client(
proxy_url: str | None,
*,
async_client: bool = False,
timeout: float | None = None,
timeout: float = _LLM_REQUEST_TIMEOUT,
) -> "httpx.Client | httpx.AsyncClient":
"""
构造显式代理策略的 httpx 客户端。
当关闭 LLM 代理时也返回 trust_env=False 的客户端,避免 httpx 自动读取
进程环境变量中的代理配置。
进程环境变量中的代理配置。连接、读取、写入和连接池等待均保留有限超时。
:param proxy_url: 显式代理地址;为空时直连
:param async_client: 是否创建异步客户端
:param timeout: 各类网络等待的超时秒数,SDK 请求层需同时保留此限制
:return: 由模型持有并在不再使用时关闭的独占客户端
"""
import httpx
client_cls = httpx.AsyncClient if async_client else httpx.Client
kwargs: dict[str, Any] = {
"trust_env": False,
"timeout": timeout,
**_build_httpx_proxy_kwargs(proxy_url),
}
if timeout is not None:
kwargs["timeout"] = timeout
return client_cls(**kwargs)
@@ -1601,6 +1607,10 @@ class LLMHelper:
) -> Any:
"""
获取 LLM 实例;调用方负责在请求或缓存图不再使用模型时调用 close_llm。
非 Bedrock 模型显式设置 180 秒的网络请求超时,防止 SDK 以 None 关闭超时。
此限制针对网络等待;包含重试的总时长由 Agent 或插件调用方控制。
:param streaming: 是否启用流式输出
:param provider: LLM提供商,默认为配置项LLM_PROVIDER
:param model: 模型名称,默认为配置项LLM_MODEL
@@ -1727,6 +1737,7 @@ class LLMHelper:
model=model_name,
api_key=runtime["api_key"],
retries=3,
timeout=_LLM_REQUEST_TIMEOUT,
**({"temperature": temperature_value} if temperature_value is not None else {}),
streaming=streaming,
client_args=_build_google_client_args(llm_proxy),
@@ -1753,6 +1764,7 @@ class LLMHelper:
api_key=runtime["api_key"],
api_base=runtime["base_url"],
max_retries=3,
request_timeout=_LLM_REQUEST_TIMEOUT,
temperature=temperature_value,
streaming=streaming,
stream_usage=True,
@@ -1810,6 +1822,7 @@ class LLMHelper:
api_key=runtime["api_key"],
base_url=runtime["base_url"],
max_retries=3,
timeout=_LLM_REQUEST_TIMEOUT,
temperature=temperature_value,
streaming=streaming,
stream_usage=True,
@@ -1834,6 +1847,7 @@ class LLMHelper:
model=model_name,
api_key=runtime["api_key"],
max_retries=3,
request_timeout=_LLM_REQUEST_TIMEOUT,
base_url=runtime.get("base_url"),
temperature=temperature_value,
streaming=streaming,

View File

@@ -66,6 +66,9 @@ APScheduler 3 的标准线程池没有任意数值 `priority` 的调度能力。
## 超时
- `LLMHelper` 为 OpenAI 兼容、DeepSeek、Anthropic 兼容和 Google 模型显式设置 180 秒的
网络请求超时,同步插件调用也生效;Bedrock 保留既有的读取超时配置。网络超时限制单次
I/O 等待,供应商重试或持续回包可能延长总耗时,不能代替调用方的整次请求预算。
- `AGENT_LLM_TIMEOUT`:单次根 Agent 模型请求总预算,默认 180 秒,覆盖流式与非流式生成
以及该请求的供应商重试;`0` 表示不限时。工具执行时间由原 `LLM_TOOL_TIMEOUT` 管理。
- `SCHEDULER_JOB_TIMEOUT`:协程调度任务的默认执行预算,默认 3600 秒,等待避让不计入执行时间;

View File

@@ -0,0 +1,133 @@
"""
验证模型请求保留有限超时,覆盖真实 SDK 的同步和异步无响应连接。
"""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from langchain_core.messages import HumanMessage
from openai import APITimeoutError
from app.agent.llm import helper as llm_module
from app.agent.llm.helper import LLMHelper
@pytest.fixture
def anyio_backend():
"""
本地 HTTP 服务和 SDK 客户端共享 asyncio 事件循环。
"""
return "asyncio"
async def _make_model(runtime, *, base_url="https://offline.example/v1"):
"""
注入离线供应商信息,使用锁定版本的真实模型 SDK。
"""
provider, model = {
"openai_compatible": ("alibaba", "qwen-test"),
"deepseek": ("deepseek", "deepseek-chat"),
"anthropic_compatible": ("anthropic", "claude-test"),
"google": ("google", "gemini-2.5-flash"),
}[runtime]
return await LLMHelper.get_llm(
provider=provider,
model=model,
api_key="offline-key",
use_proxy=False,
thinking_level="off",
api_protocol="chat_completions",
web_search_mode="disabled",
provider_runtime=SimpleNamespace(resolve_runtime=AsyncMock(return_value={
"provider_id": provider, "runtime": runtime,
"model_id": model, "api_key": "offline-key", "base_url": base_url,
})),
)
async def _close_model(model, runtime):
"""
在创建模型的循环内回收各 SDK 拥有的同步和异步连接。
"""
if runtime == "anthropic_compatible":
model._client.close()
await model._async_client.close()
elif runtime == "google":
await model.client.aio.aclose()
model.client.close()
else:
assert await LLMHelper.close_llm(model)
@pytest.mark.anyio
@pytest.mark.parametrize("proxy", [None, "http://127.0.0.1:7890"])
@pytest.mark.parametrize("runtime", [
"openai_compatible", "deepseek", "anthropic_compatible", "google",
])
async def test_real_models_have_finite_request_timeouts(monkeypatch, runtime, proxy):
"""
供应商请求层必须显式设置超时,不能以 None 覆盖 HTTP 客户端默认值。
"""
monkeypatch.setattr(llm_module, "_resolve_llm_proxy", lambda _use_proxy: proxy)
model = await _make_model(runtime)
try:
if runtime == "google":
assert model.timeout == 180
elif runtime == "anthropic_compatible":
assert model.default_request_timeout == 180
assert model._client.timeout == 180
assert model._async_client.timeout == 180
else:
assert model.request_timeout == 180
assert model.root_client.timeout == 180
assert model.root_async_client.timeout == 180
finally:
await _close_model(model, runtime)
@pytest.mark.anyio
@pytest.mark.parametrize("runtime", ["openai_compatible", "deepseek"])
@pytest.mark.parametrize("synchronous", [True, False])
async def test_silent_upstream_times_out_and_releases_connection(monkeypatch, runtime, synchronous):
"""
上游接受连接后不回包时,真实同步 generate 和异步调用均超时并断开连接。
"""
monkeypatch.setattr(llm_module, "_LLM_REQUEST_TIMEOUT", 0.05)
connections = set()
closed = asyncio.Queue()
async def silent_response(reader, writer):
"""
读完请求后保持静默,直到客户端超时关闭连接。
"""
connections.add(writer)
try:
await reader.read()
finally:
connections.discard(writer)
writer.close()
await writer.wait_closed()
closed.put_nowait(True)
server = await asyncio.start_server(silent_response, "127.0.0.1", 0)
base_url = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}/v1"
model = await _make_model(runtime, base_url=base_url)
# 本用例只验证一次真实读超时,避免 SDK 退避时间影响回归耗时。
model.root_client.max_retries = 0
model.root_async_client.max_retries = 0
try:
call = (
asyncio.to_thread(model.generate, [[HumanMessage(content="title")]])
if synchronous else model.ainvoke("title")
)
with pytest.raises(APITimeoutError):
await asyncio.wait_for(call, timeout=2)
assert await asyncio.wait_for(closed.get(), timeout=2)
finally:
for writer in tuple(connections):
writer.close()
server.close()
await server.wait_closed()
await _close_model(model, runtime)