fix(api): harden OpenAI streaming/non-streaming error handling

lightllm/server/api_http.py

@@ -376,6 +376,9 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request)
     except ClientDisconnected as e:
         logger.warning(str(e))
         return Response(status_code=499)
+    except Exception as e:
+        logger.error("An error occurred: %s", str(e), exc_info=True)
+        return create_error_response(HTTPStatus.EXPECTATION_FAILED, str(e))
     return resp
 
 
@@ -396,6 +399,9 @@ async def completions(request: CompletionRequest, raw_request: Request) -> Respo
     except ClientDisconnected as e:
         logger.warning(str(e))
         return Response(status_code=499)
+    except Exception as e:
+        logger.error("An error occurred: %s", str(e), exc_info=True)
+        return create_error_response(HTTPStatus.EXPECTATION_FAILED, str(e))
     return resp
 
 

lightllm/server/api_openai.py

@@ -30,7 +30,7 @@
 from .httpserver_for_pd_master.manager import HttpServerManagerForPDMaster
 from .api_lightllm import lightllm_get_score
 from lightllm.utils.envs_utils import get_env_start_args, get_lightllm_websocket_max_message_size
-from lightllm.utils.error_utils import ClientDisconnected
+from lightllm.utils.error_utils import ClientDisconnected, ServerBusyError
 
 from lightllm.utils.log_utils import init_logger
 from lightllm.server.metrics.manager import MetricClient
@@ -60,19 +60,46 @@
 logger = init_logger(__name__)
 
 
+def _record_request_failure_metric():
+    try:
+        from .api_http import g_objs
+
+        if g_objs.metric_client is not None:
+            g_objs.metric_client.counter_inc("lightllm_request_failure")
+    except Exception:
+        logger.warning("Failed to record lightllm_request_failure metric", exc_info=True)
+
+
+def _stream_error_chunk(message: str, err_type: str, code: Optional[int] = None):
+    error = {"message": message, "type": err_type}
+    if code is not None:
+        error["code"] = code
+    return f"data: {json.dumps({'error': error}, ensure_ascii=False)}\n\n"
+
+
 async def _safe_stream_wrapper(stream_generator):
     """Wrap a streaming generator to catch ValueError (e.g. input too long) and yield an SSE error
     event instead of letting the exception propagate to Starlette which prints a long traceback."""
     try:
         async for item in stream_generator:
             yield item
     except ValueError as e:
-        error_data = json.dumps({"error": {"message": str(e), "type": "invalid_request_error"}}, ensure_ascii=False)
-        yield f"data: {error_data}\n\n"
+        _record_request_failure_metric()
+        yield _stream_error_chunk(str(e), "invalid_request_error")
+    except ServerBusyError as e:
+        logger.error("%s", str(e), exc_info=True)
+        _record_request_failure_metric()
+        yield _stream_error_chunk(str(e), "ServerBusyError", e.status_code)
     except ClientDisconnected as e:
         logger.warning(str(e))
         # Client is gone — there's no point yielding more SSE chunks. Stop quietly.
         return
+    except asyncio.CancelledError:
+        raise
+    except Exception as e:
+        logger.error("An error occurred in streaming response: %s", str(e), exc_info=True)
+        _record_request_failure_metric()
+        yield _stream_error_chunk(str(e), "InternalServerError")
 
 
 def _serialize_sse_chunk(chunk, choice_nulls=(), response_nulls=()):

unit_tests/server/test_openai_stream_error_handling.py

@@ -0,0 +1,98 @@
+import asyncio
+
+import ujson as json
+
+from lightllm.server import api_http
+from lightllm.server.api_openai import _safe_stream_wrapper
+from lightllm.utils.error_utils import ServerBusyError
+
+
+class FakeMetricClient:
+    def __init__(self):
+        self.counters = []
+
+    def counter_inc(self, name):
+        self.counters.append(name)
+
+
+def _decode_sse_payload(chunk):
+    if isinstance(chunk, bytes):
+        chunk = chunk.decode("utf-8")
+    assert chunk.startswith("data: ")
+    return json.loads(chunk.removeprefix("data: ").strip())
+
+
+def _collect(gen):
+    async def drain():
+        return [c async for c in gen]
+
+    return asyncio.run(drain())
+
+
+def test_safe_stream_wrapper_converts_unexpected_exception_to_sse_error(monkeypatch):
+    metric_client = FakeMetricClient()
+    monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client)
+
+    async def failing_stream():
+        if False:
+            yield b""
+        raise RuntimeError("backend failed")
+
+    chunks = _collect(_safe_stream_wrapper(failing_stream()))
+
+    assert len(chunks) == 1
+    payload = _decode_sse_payload(chunks[0])
+    assert payload["error"]["message"] == "backend failed"
+    assert payload["error"]["type"] == "InternalServerError"
+    assert metric_client.counters == ["lightllm_request_failure"]
+
+
+def test_safe_stream_wrapper_maps_server_busy_error(monkeypatch):
+    metric_client = FakeMetricClient()
+    monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client)
+
+    async def busy_stream():
+        if False:
+            yield b""
+        raise ServerBusyError("server overloaded")
+
+    chunks = _collect(_safe_stream_wrapper(busy_stream()))
+
+    assert len(chunks) == 1
+    payload = _decode_sse_payload(chunks[0])
+    assert payload["error"]["type"] == "ServerBusyError"
+    assert payload["error"]["code"] == 503
+    assert metric_client.counters == ["lightllm_request_failure"]
+
+
+def test_safe_stream_wrapper_keeps_value_error_as_invalid_request(monkeypatch):
+    metric_client = FakeMetricClient()
+    monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client)
+
+    async def bad_stream():
+        if False:
+            yield b""
+        raise ValueError("input too long")
+
+    chunks = _collect(_safe_stream_wrapper(bad_stream()))
+
+    payload = _decode_sse_payload(chunks[0])
+    assert payload["error"]["type"] == "invalid_request_error"
+    assert metric_client.counters == ["lightllm_request_failure"]
+
+
+def test_safe_stream_wrapper_swallows_client_disconnect(monkeypatch):
+    from lightllm.utils.error_utils import ClientDisconnected
+
+    metric_client = FakeMetricClient()
+    monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client)
+
+    async def gone_stream():
+        if False:
+            yield b""
+        raise ClientDisconnected("client gone")
+
+    chunks = _collect(_safe_stream_wrapper(gone_stream()))
+
+    assert chunks == []
+    assert metric_client.counters == []