diff --git a/fastdeploy/model_executor/layers/moe/fused_moe_backend_base.py b/fastdeploy/model_executor/layers/moe/fused_moe_backend_base.py index d72ce1232eb..32d27ba219a 100644 --- a/fastdeploy/model_executor/layers/moe/fused_moe_backend_base.py +++ b/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 diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 00000000000..cea384b2524 --- /dev/null +++ b/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 diff --git a/test/e2e/__init__.py b/test/e2e/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/test/e2e/test_common_model_v1_loader_offline.py b/test/e2e/test_common_model_v1_loader_offline.py new file mode 100644 index 00000000000..9a16bb83cbe --- /dev/null +++ b/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", + ) diff --git a/test/e2e/utils.py b/test/e2e/utils.py new file mode 100644 index 00000000000..fb6c32a0c02 --- /dev/null +++ b/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)