fix(api): align OpenAI validation and stop handling

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))

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,

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 = "<END>"
+
+
+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<EN") == "safe"
+    assert partial_filter.process("", final=True) == "<EN"
+
+
+def test_chat_completion_omits_stop_sequence(fake_generation):
+    request = ChatCompletionRequest(
+        model="test-model",
+        messages=[{"role": "user", "content": "hello"}],
+        stop=STOP_SEQUENCE,
+    )
+
+    response = asyncio.run(api_openai.chat_completions_impl(request, SimpleNamespace()))
+
+    assert response.choices[0].message.content == "Hello "
+    assert response.choices[0].finish_reason == "stop"
+
+
+def test_streaming_chat_completion_omits_stop_sequence(fake_generation):
+    request = ChatCompletionRequest(
+        model="test-model",
+        messages=[{"role": "user", "content": "hello"}],
+        stop=STOP_SEQUENCE,
+        stream=True,
+    )
+
+    response = asyncio.run(api_openai.chat_completions_impl(request, SimpleNamespace()))
+    events = asyncio.run(collect_sse(response))
+
+    assert streamed_text(events, "delta") == "Hello "
+    assert any(choice.get("finish_reason") == "stop" for event in events for choice in event.get("choices", []))
+
+
+def test_completion_omits_stop_sequence(fake_generation):
+    request = CompletionRequest(model="test-model", prompt="hello", stop=STOP_SEQUENCE)
+
+    response = asyncio.run(api_openai.completions_impl(request, SimpleNamespace()))
+
+    assert response.choices[0].text == "Hello "
+    assert response.choices[0].finish_reason == "stop"
+
+
+def test_streaming_completion_omits_stop_sequence(fake_generation):
+    request = CompletionRequest(model="test-model", prompt="hello", stop=STOP_SEQUENCE, stream=True)
+
+    response = asyncio.run(api_openai.completions_impl(request, SimpleNamespace()))
+    events = asyncio.run(collect_sse(response))
+
+    assert streamed_text(events, "text") == "Hello "
+    assert any(choice.get("finish_reason") == "stop" for event in events for choice in event.get("choices", []))

unit_tests/server/test_openai_validation_error.py

@@ -0,0 +1,45 @@
+from types import SimpleNamespace
+
+import pytest
+from fastapi.testclient import TestClient
+
+from lightllm.server import api_http
+
+
+@pytest.fixture
+def client(monkeypatch):
+    monkeypatch.setattr(api_http.g_objs, "metric_client", SimpleNamespace(counter_inc=lambda *args: None))
+    test_client = TestClient(api_http.app, raise_server_exceptions=False)
+    yield test_client
+    test_client.close()
+
+
+def test_missing_required_parameter_uses_openai_error_envelope(client):
+    response = client.post("/v1/completions", json={"prompt": "hello"})
+
+    assert response.status_code == 422
+    assert response.json() == {
+        "error": {
+            "message": "Missing required parameter: 'model'.",
+            "type": "invalid_request_error",
+            "param": "model",
+            "code": 422,
+        }
+    }
+
+
+def test_invalid_parameter_uses_openai_error_envelope(client):
+    response = client.post(
+        "/v1/chat/completions",
+        json={"model": "test-model", "messages": [{"role": "user", "content": "hello"}], "seed": -2},
+    )
+
+    assert response.status_code == 422
+    assert response.json() == {
+        "error": {
+            "message": "Invalid value for 'seed': Input should be greater than or equal to -1",
+            "type": "invalid_request_error",
+            "param": "seed",
+            "code": 422,
+        }
+    }