diff --git a/app/agent/llm/helper.py b/app/agent/llm/helper.py index aaea553f0..da56f1604 100644 --- a/app/agent/llm/helper.py +++ b/app/agent/llm/helper.py @@ -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, diff --git a/docs/agent-execution.md b/docs/agent-execution.md index 0b27d33e2..76f3b8800 100644 --- a/docs/agent-execution.md +++ b/docs/agent-execution.md @@ -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 秒，等待避让不计入执行时间； diff --git a/tests/test_llm_request_timeout.py b/tests/test_llm_request_timeout.py new file mode 100644 index 000000000..3f1b150e2 --- /dev/null +++ b/tests/test_llm_request_timeout.py @@ -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)