diff --git a/docs/admin/settings/settings_ai_summary.rst b/docs/admin/settings/settings_ai_summary.rst index 0d9fe2d07..37d355f1d 100644 --- a/docs/admin/settings/settings_ai_summary.rst +++ b/docs/admin/settings/settings_ai_summary.rst @@ -201,7 +201,16 @@ Troubleshooting **The summary box shows an error.** SearXNG could not reach the LLM server, or the server rejected the request. Check ``base_url`` from the SearXNG machine, confirm the model name exists on - that server (``ollama list``), and check the SearXNG log. + that server (``ollama list``), and check the SearXNG log -- it records the + status and the message the server replied with, which usually names the cause. + +**The first search after a while fails, the next one works.** + An idle LLM server unloads the model and has to load it again, and it sends + nothing at all while doing so. If that takes longer than ``read_timeout`` + (30 s by default) the request is abandoned. SearXNG repeats the request once, + which covers a normal load, but a large model on slow storage can need more: + raise ``read_timeout``, or keep the model in memory -- with Ollama, set + ``OLLAMA_KEEP_ALIVE`` (for example ``-1`` to never unload it). **The summary starts, then stops mid-sentence.** The answer exceeded ``stream_timeout`` (120 s by default). Large models on diff --git a/searx/plugins/ai_summary.py b/searx/plugins/ai_summary.py index 69f4d6643..f3d22e366 100644 --- a/searx/plugins/ai_summary.py +++ b/searx/plugins/ai_summary.py @@ -47,6 +47,18 @@ if t.TYPE_CHECKING: VALID_ROLES = ("user", "assistant") MODEL_NAME_REGEXP = re.compile(r"[A-Za-z0-9._:/-]{1,128}") +log = logging.getLogger("searx.plugins.ai_summary") + +UPSTREAM_RETRIES = 1 +"""How often a request to the LLM server is repeated when the server does not +answer in time. + +An idle LLM server unloads the model, and loads it again on the next request -- +which can take longer than :py:obj:`read_timeout +`, because no byte of the +response is sent while the model is loading. The request that runs into this +is also the request that starts the load, so repeating it usually succeeds.""" + def _get_client(base_url: str, cfg: SettingsAISummary, api_key: str = "") -> httpx.Client: """HTTP client for one request to the LLM server at ``base_url``. The @@ -238,6 +250,47 @@ def _validate_payload(payload: t.Any, cfg: SettingsAISummary) -> tuple[list[dict return _validate_messages(payload.get("messages"), cfg), _validate_context(payload.get("context", []), cfg) +def _open_upstream(client: httpx.Client, payload: dict[str, t.Any]) -> tuple[t.Any, t.Any]: + """Start the streaming completion on the LLM server. + + Returns the (already entered) stream context and the response, or + ``(None, None)`` if no usable response was received. A server that does not + answer in time is asked again (:py:obj:`UPSTREAM_RETRIES`); a server that + *did* answer, but with an error status, is not -- a wrong API key or an + unknown model name does not become right when asked twice.""" + + for attempt in range(UPSTREAM_RETRIES + 1): + stream_ctx = client.stream("POST", "/chat/completions", json=payload) + try: + resp = stream_ctx.__enter__() # pylint: disable=unnecessary-dunder-call + except httpx.TransportError as exc: + if attempt < UPSTREAM_RETRIES: + log.debug("LLM server did not answer (%s), asking again", exc) + continue + log.warning("LLM server did not answer: %s", exc) + return None, None + except httpx.HTTPError as exc: + log.warning("request to the LLM server failed: %s", exc) + return None, None + + if resp.status_code == 200: + return stream_ctx, resp + + # the body of an error response is short and usually names the cause, + # e.g. an unknown model; without it a misconfiguration is invisible + detail = "" + try: + resp.read() + detail = resp.text.strip()[:200] + except (httpx.HTTPError, UnicodeDecodeError): # pragma: no cover + pass + log.warning("LLM server responded with HTTP %s %s", resp.status_code, detail) + stream_ctx.__exit__(None, None, None) + return None, None + + return None, None # pragma: no cover - the loop always returns + + def ai_summary_view() -> flask.Response: """Stream an AI generated answer for the messages in the request body, response is NDJSON: ``{"delta": ..}`` lines followed by one final @@ -271,22 +324,13 @@ def ai_summary_view() -> flask.Response: # reported as HTTP 502 instead of a line in an already started stream user_api_key = str(sxng_request.preferences.get_value("ai_summary_api_key") or "").strip() client = _get_client(server, cfg, _server_api_key(cfg, server, user_api_key)) - stream_ctx = client.stream("POST", "/chat/completions", json=chat_payload) - upstream = None - try: - upstream = stream_ctx.__enter__() # pylint: disable=unnecessary-dunder-call - if upstream.status_code != 200: - stream_ctx.__exit__(None, None, None) - upstream = None - except httpx.HTTPError: - upstream = None - if upstream is None: + stream_ctx, upstream = _open_upstream(client, chat_payload) + if stream_ctx is None or upstream is None: client.close() return flask.Response(json.dumps({"error": "upstream error"}), status=502, mimetype="application/json") # from here on nothing must be read from the request context, the # generator runs after the request context has been torn down - log = logging.getLogger("searx.plugins.ai_summary") def ndjson(obj: dict[str, t.Any]) -> bytes: # the generator bypasses flask's response encoding (direct_passthrough) diff --git a/tests/unit/test_plugin_ai_summary.py b/tests/unit/test_plugin_ai_summary.py index b35da38eb..3735dcc80 100644 --- a/tests/unit/test_plugin_ai_summary.py +++ b/tests/unit/test_plugin_ai_summary.py @@ -31,6 +31,8 @@ def sse_stream_mock(lines: list[dict], status_code: int = 200) -> Mock: upstream = Mock(status_code=status_code) upstream.iter_lines.return_value = iter([f"data: {json.dumps(line)}" for line in lines] + ["data: [DONE]"]) + # httpx exposes the body of an error response as a str; the endpoint logs it + upstream.text = "" @contextmanager def stream(*_args, **_kwargs): @@ -425,6 +427,71 @@ class PluginAISummary(SearxTestCase): self.assertIn('tab-label-ai"', html) self.assertIn("ai_summary_api_key", html) + def test_endpoint_retries_a_server_that_does_not_answer(self): + # a cold LLM server loads the model before it answers anything, which + # can outlast read_timeout; the retry is what makes the first search + # after an idle period work + import httpx # pylint: disable=import-outside-toplevel + + attempts = [] + ok = sse_stream_mock([{"choices": [{"delta": {"content": "hi"}}]}]) + + @contextmanager + def stream(*_args, **_kwargs): + attempts.append(1) + if len(attempts) == 1: + raise httpx.ReadTimeout("timed out") + with ok.stream() as resp: + yield resp + + client_mock = Mock() + client_mock.stream = stream + self.mock_upstream(client_mock) + + res = self.client.post("/ai_summary", json={"messages": [{"role": "user", "content": "hi"}]}) + self.assertEqual(res.status_code, 200) + self.assertEqual(2, len(attempts)) + lines = [json.loads(line) for line in res.data.decode().splitlines() if line] + self.assertEqual(lines[0], {"delta": "hi"}) + + def test_endpoint_gives_up_after_the_retry(self): + import httpx # pylint: disable=import-outside-toplevel + + attempts = [] + + @contextmanager + def stream(*_args, **_kwargs): + attempts.append(1) + raise httpx.ConnectTimeout("nope") + yield # pylint: disable=unreachable + + client_mock = Mock() + client_mock.stream = stream + self.mock_upstream(client_mock) + + res = self.client.post("/ai_summary", json={"messages": [{"role": "user", "content": "hi"}]}) + self.assertEqual(res.status_code, 502) + self.assertEqual(2, len(attempts)) + + def test_endpoint_does_not_retry_an_error_status(self): + # a wrong API key or an unknown model does not fix itself + attempts = [] + bad = sse_stream_mock([], status_code=401) + + @contextmanager + def stream(*_args, **_kwargs): + attempts.append(1) + with bad.stream() as resp: + yield resp + + client_mock = Mock() + client_mock.stream = stream + self.mock_upstream(client_mock) + + res = self.client.post("/ai_summary", json={"messages": [{"role": "user", "content": "hi"}]}) + self.assertEqual(res.status_code, 502) + self.assertEqual(1, len(attempts)) + def test_endpoint_upstream_error(self): self.mock_upstream(sse_stream_mock([], status_code=500))