[v1 loader UT]add v1 loader unitest

fastdeploy/model_executor/layers/moe/fused_moe_backend_base.py

@@ -210,3 +210,5 @@ def create_weights(self, layer: nn.Layer, **extra_weight_attrs):
                 shape=gate_correction_bias_shape,
                 dtype="float32",
             )
+        else:
+            layer.gate_correction_bias = None

test/conftest.py

@@ -0,0 +1,89 @@
+# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from typing import Any, Union
+
+import pytest
+from e2e.utils import clean_ports
+
+
+class FDRunner:
+    def __init__(
+        self,
+        model_name_or_path: str,
+        tensor_parallel_size: int = 1,
+        max_model_len: int = 1024,
+        load_choices: str = "default",
+        enable_custom_all_reduce: bool = False,
+        use_cudagraph: bool = False,
+        quantization: str = "None",
+        num_gpu_blocks_override: int = 1024,
+        **kwargs,
+    ) -> None:
+        from fastdeploy.entrypoints.llm import LLM
+
+        ports_to_clean = []
+        if "engine_worker_queue_port" in kwargs:
+            ports_to_clean.append(kwargs["engine_worker_queue_port"])
+        clean_ports(ports_to_clean)
+        self.llm = LLM(
+            model=model_name_or_path,
+            num_gpu_blocks_override=num_gpu_blocks_override,
+            tensor_parallel_size=tensor_parallel_size,
+            max_model_len=max_model_len,
+            load_choices=load_choices,
+            enable_custom_all_reduce=enable_custom_all_reduce,
+            use_cudagraph=use_cudagraph,
+            quantization=quantization,
+            **kwargs,
+        )
+
+    def generate(
+        self,
+        prompts: list[str],
+        sampling_params,
+        **kwargs: Any,
+    ) -> list[tuple[list[list[int]], list[str]]]:
+
+        req_outputs = self.llm.generate(prompts, sampling_params=sampling_params, **kwargs)
+        outputs: list[tuple[list[list[int]], list[str]]] = []
+        sample_output_ids: list[list[int]] = []
+        sample_output_strs: list[str] = []
+        for output in req_outputs:
+            sample_output_ids.append(output.outputs.token_ids)
+            sample_output_strs.append(output.outputs.text)
+            outputs.append((sample_output_ids, sample_output_strs))
+        return outputs
+
+    def generate_topp0(
+        self,
+        prompts: Union[list[str]],
+        max_tokens: int,
+        **kwargs: Any,
+    ) -> list[tuple[list[int], str]]:
+        from fastdeploy.engine.sampling_params import SamplingParams
+
+        topp_params = SamplingParams(temperature=0.1, top_p=0, max_tokens=max_tokens)
+        outputs = self.generate(prompts, topp_params, **kwargs)
+        return outputs
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        del self.llm
+
+
+@pytest.fixture(scope="session")
+def fd_runner():
+    return FDRunner

test/e2e/test_common_model_v1_loader_offline.py

@@ -0,0 +1,125 @@
+# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import traceback
+from multiprocessing import Process, Queue
+
+import pytest
+from utils import check_tokens_id_and_text_close
+
+FD_ENGINE_QUEUE_PORT = int(os.getenv("FD_ENGINE_QUEUE_PORT", 8313))
+MAX_WAIT_SECONDS = 60 * 5
+
+prompts = ["解释下“温故而知新", "Hello, how are you?"]
+
+
+def form_model_get_output(
+    fd_runner, model_path, tensor_parallel_size, max_model_len, max_tokens, quantization, load_choices, result_queue
+):
+    try:
+        with fd_runner(
+            model_path,
+            tensor_parallel_size=tensor_parallel_size,
+            max_model_len=max_model_len,
+            load_choices=load_choices,
+            quantization=quantization,
+            engine_worker_queue_port=FD_ENGINE_QUEUE_PORT,
+        ) as fd_model:
+            fd_outputs = fd_model.generate_topp0(prompts, max_tokens=max_tokens)
+            result_queue.put(fd_outputs)
+    except Exception:
+        print(f"Failed using {load_choices} laoder to load model from {model_path}.")
+        traceback.print_exc()
+        pytest.fail(f"Failed to initialize LLM model from {model_path}")
+
+
+@pytest.mark.parametrize(
+    "model_name_or_path,tensor_parallel_size,max_model_len",
+    [
+        pytest.param(
+            "Qwen3-30B-A3B",
+            2,
+            1024,
+            marks=[pytest.mark.core_model],
+        ),
+        pytest.param(
+            "Qwen3-0.6B",
+            1,
+            1024,
+            marks=[pytest.mark.core_model],
+        ),
+        pytest.param(
+            "ernie-4_5-21b-a3b-bf16-paddle",
+            2,
+            1024,
+            marks=[pytest.mark.core_model],
+        ),
+    ],
+)
+@pytest.mark.parametrize("quantization", ["None"])
+@pytest.mark.parametrize("max_tokens", [32])
+def test_v1_loader_models(
+    fd_runner,
+    model_name_or_path: str,
+    tensor_parallel_size: int,
+    max_model_len: int,
+    max_tokens: int,
+    quantization: str,
+) -> None:
+    base_path = os.getenv("MODEL_PATH")
+    if base_path:
+        model_path = os.path.join(base_path, model_name_or_path)
+    else:
+        model_path = model_name_or_path
+    result_queue = Queue()
+    p = Process(
+        target=form_model_get_output,
+        args=(
+            fd_runner,
+            model_path,
+            tensor_parallel_size,
+            max_model_len,
+            max_tokens,
+            quantization,
+            "default",
+            result_queue,
+        ),
+    )
+    p.start()
+    p.join()
+    fd_outputs_v0 = result_queue.get(timeout=60)
+
+    p = Process(
+        target=form_model_get_output,
+        args=(
+            fd_runner,
+            model_path,
+            tensor_parallel_size,
+            max_model_len,
+            max_tokens,
+            quantization,
+            "default_v1",
+            result_queue,
+        ),
+    )
+    p.start()
+    p.join()
+    fd_outputs_v1 = result_queue.get(timeout=60)
+    check_tokens_id_and_text_close(
+        outputs_0_lst=fd_outputs_v0,
+        outputs_1_lst=fd_outputs_v1,
+        name_0="default loader",
+        name_1="default_v1 loader",
+    )

test/e2e/utils.py

@@ -0,0 +1,92 @@
+# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import os
+import signal
+import socket
+import subprocess
+import warnings
+
+TokensIdText = list[tuple[list[int], str]]
+
+
+# (token_ids, text)
+def kill_process_on_port(port: int):
+    """
+    Kill processes that are listening on the given port.
+    Uses `lsof` to find process ids and sends SIGKILL.
+    """
+    try:
+        output = subprocess.check_output(f"lsof -i:{port} -t", shell=True).decode().strip()
+        for pid in output.splitlines():
+            os.kill(int(pid), signal.SIGKILL)
+            print(f"Killed process on port {port}, pid={pid}")
+    except subprocess.CalledProcessError:
+        pass
+
+
+def clean_ports(ports_to_clean: list[int]):
+    """
+    Kill all processes occupying the ports listed in PORTS_TO_CLEAN.
+    """
+    for port in ports_to_clean:
+        kill_process_on_port(port)
+
+
+def is_port_open(host: str, port: int, timeout=1.0):
+    """
+    Check if a TCP port is open on the given host.
+    Returns True if connection succeeds, False otherwise.
+    """
+    try:
+        with socket.create_connection((host, port), timeout):
+            return True
+    except Exception:
+        return False
+
+
+def check_tokens_id_and_text_close(
+    *,
+    outputs_0_lst: TokensIdText,
+    outputs_1_lst: TokensIdText,
+    name_0: str,
+    name_1: str,
+    warn_on_mismatch: bool = True,
+) -> None:
+    assert len(outputs_0_lst) == len(outputs_1_lst)
+
+    for prompt_idx, (outputs_0, outputs_1) in enumerate(zip(outputs_0_lst, outputs_1_lst)):
+        assert len(outputs_0) == len(outputs_1)
+        output_ids_0, output_str_0 = outputs_0
+        output_ids_1, output_str_1 = outputs_1
+
+        # Loop through generated tokens.
+        for idx, (output_id_0, output_id_1) in enumerate(zip(output_ids_0, output_ids_1)):
+            is_tok_mismatch = output_id_0 != output_id_1
+            if is_tok_mismatch and warn_on_mismatch:
+                fail_msg = (
+                    f"Test{prompt_idx}:"
+                    f"\nMatched tokens:\t{output_ids_0[:idx]}"
+                    f"\n{name_0}:\t{output_str_0!r}"
+                    f"\n{name_1}:\t{output_str_1!r}"
+                )
+                with warnings.catch_warnings():
+                    warnings.simplefilter("always")
+                    warnings.warn(fail_msg, stacklevel=2)
+                break
+    else:
+        if output_str_0 != output_str_1 and warn_on_mismatch:
+            fail_msg = f"Test{prompt_idx}:" f"\n{name_0}:\t{output_str_0!r}" f"\n{name_1}:\t{output_str_1!r}"
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(fail_msg, stacklevel=2)