From 2a75fa4f609f139f3cf329764fa8396c7f36b76a Mon Sep 17 00:00:00 2001 From: sufubao Date: Thu, 6 Aug 2026 22:05:36 +0800 Subject: [PATCH] fix(api): align OpenAI validation and stop handling --- lightllm/server/api_http.py | 31 ++++ lightllm/server/api_openai.py | 75 +++++++-- .../server/test_openai_stop_sequences.py | 145 ++++++++++++++++++ .../server/test_openai_validation_error.py | 45 ++++++ 4 files changed, 286 insertions(+), 10 deletions(-) create mode 100644 unit_tests/server/test_openai_stop_sequences.py create mode 100644 unit_tests/server/test_openai_validation_error.py diff --git a/lightllm/server/api_http.py b/lightllm/server/api_http.py index 1cda03b96..97e47ac86 100755 --- a/lightllm/server/api_http.py +++ b/lightllm/server/api_http.py @@ -37,6 +37,7 @@ from typing import Callable from lightllm.server import TokenLoad from fastapi import BackgroundTasks, FastAPI, Request +from fastapi.exceptions import RequestValidationError from fastapi.responses import Response, StreamingResponse, JSONResponse from lightllm.server.core.objs.sampling_params import SamplingParams from lightllm.server.core.objs import StartArgs @@ -178,6 +179,36 @@ def create_server_busy_response(exc: ServerBusyError) -> JSONResponse: return create_error_response(status, str(exc), err_type="RateLimitError") +@app.exception_handler(RequestValidationError) +async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse: + errors = exc.errors() + if not errors: + return create_error_response( + HTTPStatus.UNPROCESSABLE_ENTITY, + str(exc), + err_type="invalid_request_error", + ) + + error = errors[0] + location = error.get("loc", ()) + param_parts = [str(part) for part in location if part != "body"] + param = ".".join(param_parts) or None + + if error.get("type") == "missing" and param is not None: + message = f"Missing required parameter: '{param}'." + elif param is not None: + message = f"Invalid value for '{param}': {error.get('msg', 'Request validation failed')}" + else: + message = error.get("msg", "Request validation failed") + + return create_error_response( + HTTPStatus.UNPROCESSABLE_ENTITY, + message, + err_type="invalid_request_error", + param=param, + ) + + @app.exception_handler(ServerBusyError) async def server_busy_exception_handler(request: Request, exc: ServerBusyError) -> JSONResponse: logger.warning(str(exc)) diff --git a/lightllm/server/api_openai.py b/lightllm/server/api_openai.py index 413d05c99..968f7c39e 100644 --- a/lightllm/server/api_openai.py +++ b/lightllm/server/api_openai.py @@ -125,6 +125,58 @@ def _serialize_sse_chunk(chunk, choice_nulls=(), response_nulls=()): return json.dumps(d, ensure_ascii=False) +class _StopSequenceFilter: + """Hide stop strings while preserving text that only partially matches one.""" + + def __init__(self, stop_sequences: List[str]): + self.stop_sequences = [sequence for sequence in stop_sequences if sequence] + self.pending = "" + self.stopped = False + + def process(self, text: str, *, final: bool = False) -> str: + if self.stopped: + return "" + + self.pending += text + stop_index = None + for stop_sequence in self.stop_sequences: + index = self.pending.find(stop_sequence) + if index != -1 and (stop_index is None or index < stop_index): + stop_index = index + + if stop_index is not None: + output = self.pending[:stop_index] + self.pending = "" + self.stopped = True + return output + + if final or not self.stop_sequences: + output = self.pending + self.pending = "" + return output + + partial_match_length = 0 + for stop_sequence in self.stop_sequences: + max_length = min(len(self.pending), len(stop_sequence) - 1) + for length in range(max_length, 0, -1): + if self.pending.endswith(stop_sequence[:length]): + partial_match_length = max(partial_match_length, length) + break + + if partial_match_length == 0: + output = self.pending + self.pending = "" + return output + + output = self.pending[:-partial_match_length] + self.pending = self.pending[-partial_match_length:] + return output + + +def _remove_stop_sequences(text: str, stop_sequences: List[str]) -> str: + return _StopSequenceFilter(stop_sequences).process(text, final=True) + + def create_error_response( status_code: HTTPStatus, message: str, err_type: str = None, param: str = None ) -> JSONResponse: @@ -426,6 +478,8 @@ async def chat_completions_impl(request: ChatCompletionRequest, raw_request: Req finish_reason = finish_reason_dict[sub_req_id] text = "".join(final_output_dict[sub_req_id]) + if finish_reason == "stop": + text = _remove_stop_sequences(text, sampling_params.stop_sequences.to_strings()) # Handle reasoning content reasoning_text = None @@ -513,6 +567,10 @@ async def stream_results() -> AsyncGenerator[bytes, None]: has_emitted_tool_calls: Dict[int, bool] = collections.defaultdict(bool) has_emitted_first_chunk: Dict[int, bool] = collections.defaultdict(bool) stream_tool_call_ids: Dict[Tuple[int, int], str] = {} + stop_sequences = sampling_params.stop_sequences.to_strings() + stop_filters: Dict[int, _StopSequenceFilter] = collections.defaultdict( + lambda: _StopSequenceFilter(stop_sequences) + ) from .req_id_generator import convert_sub_id_to_group_id prompt_tokens = 0 @@ -525,8 +583,8 @@ async def stream_results() -> AsyncGenerator[bytes, None]: group_request_id = convert_sub_id_to_group_id(sub_req_id) choice_index = sub_req_id - group_request_id - delta = request_output current_finish_reason = finish_status.get_finish_reason() + delta = stop_filters[sub_req_id].process(request_output, final=current_finish_reason is not None) # Emit the initial role-only chunk once per choice, as required by the # OpenAI SSE spec: role appears only in the first delta with content="". @@ -960,6 +1018,10 @@ async def stream_results() -> AsyncGenerator[bytes, None]: prompt_tokens = 0 completion_tokens = 0 cached_tokens = 0 + stop_sequences = sampling_params.stop_sequences.to_strings() + stop_filters: Dict[int, _StopSequenceFilter] = collections.defaultdict( + lambda: _StopSequenceFilter(stop_sequences) + ) async for sub_req_id, request_output, metadata, finish_status in results_generator: group_request_id = convert_sub_id_to_group_id(sub_req_id) @@ -971,7 +1033,7 @@ async def stream_results() -> AsyncGenerator[bytes, None]: if finish_status.is_finished(): current_finish_reason = finish_status.get_finish_reason() - output_text = request_output + output_text = stop_filters[sub_req_id].process(request_output, final=current_finish_reason is not None) if request.echo and metadata.get("is_first_token", False): prompt_str = prompt if isinstance(prompt, list): @@ -1050,16 +1112,9 @@ async def _collect_generation_results( prompt_tokens = metadata["prompt_tokens"] prompt_cache_len = metadata.get("prompt_cache_len", 0) - # 处理停止序列剔除 final_text = "".join(final_output) if finish_reason == "stop" and sampling_params.stop_sequences.size > 0: - valid_stop_strings = sampling_params.stop_sequences.to_strings() - for stop_str in valid_stop_strings: - stop_index = final_text.rfind(stop_str, max(0, len(final_text) - len(stop_str) - 20), len(final_text)) - if stop_index != -1: - logger.debug(f"removed stop sequence in tail: '{final_text[stop_index:]}'") - final_text = final_text[:stop_index] - break + final_text = _remove_stop_sequences(final_text, sampling_params.stop_sequences.to_strings()) return { "index": prompt_index, diff --git a/unit_tests/server/test_openai_stop_sequences.py b/unit_tests/server/test_openai_stop_sequences.py new file mode 100644 index 000000000..49587c7f2 --- /dev/null +++ b/unit_tests/server/test_openai_stop_sequences.py @@ -0,0 +1,145 @@ +import asyncio +import json +from types import SimpleNamespace + +import pytest + +from lightllm.server import api_http, api_openai +from lightllm.server.api_models import ChatCompletionRequest, CompletionRequest +from lightllm.server.core.objs import FinishStatus +from lightllm.server.core.objs import sampling_params as sampling_params_module + + +STOP_SEQUENCE = "" + + +class FakeTokenizer: + def encode(self, text, add_special_tokens=False): + return [ord(character) for character in text] + + def decode(self, token_ids, skip_special_tokens=False): + return "".join(chr(token_id) for token_id in token_ids) + + +class FakeHttpServerManager: + def __init__(self, chunks): + self.tokenizer = FakeTokenizer() + self.chunks = chunks + + def generate(self, prompt, sampling_params, multimodal_params, request): + async def results(): + for index, (text, status) in enumerate(self.chunks): + metadata = { + "prompt_tokens": 1, + "prompt_cache_len": 0, + "is_first_token": index == 0, + } + yield 8, text, metadata, FinishStatus(status) + + return results() + + +@pytest.fixture +def fake_generation(monkeypatch): + manager = FakeHttpServerManager( + [ + ("Hello <", FinishStatus.NO_FINISH), + ("END", FinishStatus.NO_FINISH), + (">", FinishStatus.FINISHED_STOP), + ] + ) + monkeypatch.setattr(api_http.g_objs, "httpserver_manager", manager) + monkeypatch.setattr(api_openai, "get_env_start_args", lambda: SimpleNamespace(reasoning_parser=None)) + monkeypatch.setattr( + sampling_params_module, + "get_env_start_args", + lambda: SimpleNamespace(enable_prompt_logprobs=False), + ) + + async def fake_build_prompt(request, tools): + return "prompt" + + monkeypatch.setattr(api_openai, "build_prompt", fake_build_prompt) + return manager + + +async def collect_sse(response): + events = [] + async for chunk in response.body_iterator: + if isinstance(chunk, bytes): + chunk = chunk.decode("utf-8") + for line in chunk.splitlines(): + if not line.startswith("data: ") or line == "data: [DONE]": + continue + events.append(json.loads(line.removeprefix("data: "))) + return events + + +def streamed_text(events, field): + output = [] + for event in events: + for choice in event.get("choices", []): + if field == "delta": + output.append(choice.get("delta", {}).get("content") or "") + else: + output.append(choice.get(field) or "") + return "".join(output) + + +def test_stop_sequence_filter_handles_chunk_boundaries_and_partial_matches(): + stop_filter = api_openai._StopSequenceFilter([STOP_SEQUENCE]) + + assert stop_filter.process("safe<") == "safe" + assert stop_filter.process("END") == "" + assert stop_filter.process(">must-not-leak", final=True) == "" + + partial_filter = api_openai._StopSequenceFilter([STOP_SEQUENCE]) + assert partial_filter.process("safe