diff --git a/lightllm/server/api_http.py b/lightllm/server/api_http.py index 9c8981730..11417dac4 100755 --- a/lightllm/server/api_http.py +++ b/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 diff --git a/lightllm/server/api_openai.py b/lightllm/server/api_openai.py index 0d934c44c..33a97835c 100644 --- a/lightllm/server/api_openai.py +++ b/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,6 +60,23 @@ 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.""" @@ -67,12 +84,22 @@ async def _safe_stream_wrapper(stream_generator): 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=()): diff --git a/unit_tests/server/test_openai_stream_error_handling.py b/unit_tests/server/test_openai_stream_error_handling.py new file mode 100644 index 000000000..c7d2f23f8 --- /dev/null +++ b/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 == []