From bc121f030b38fd3d15fff76fd4705318bc5a8edf Mon Sep 17 00:00:00 2001 From: xming521 <1223398803@qq.com> Date: Thu, 19 Jun 2025 16:38:09 +0800 Subject: [PATCH] =?UTF-8?q?0.2.23=20=E8=BF=90=E8=A1=8C=20weclone-cli=20ser?= =?UTF-8?q?ver=20=E6=8A=A5=E9=94=99=20Fixes=20#158?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Qwen2.5-VL.jsonc} | 0 .../{full_pipe.jsonc => configs/qwen3.jsonc} | 42 +++-- tests/test_full_pipe.py | 173 ----------------- tests/test_full_pipeV2.py | 176 +++++++++++------- weclone/server/api_service.py | 2 +- 5 files changed, 137 insertions(+), 256 deletions(-) rename tests/{full_pipeV2.jsonc => configs/Qwen2.5-VL.jsonc} (100%) rename tests/{full_pipe.jsonc => configs/qwen3.jsonc} (64%) delete mode 100644 tests/test_full_pipe.py diff --git a/tests/full_pipeV2.jsonc b/tests/configs/Qwen2.5-VL.jsonc similarity index 100% rename from tests/full_pipeV2.jsonc rename to tests/configs/Qwen2.5-VL.jsonc diff --git a/tests/full_pipe.jsonc b/tests/configs/qwen3.jsonc similarity index 64% rename from tests/full_pipe.jsonc rename to tests/configs/qwen3.jsonc index ba070d6..62b819b 100644 --- a/tests/full_pipe.jsonc +++ b/tests/configs/qwen3.jsonc @@ -1,11 +1,14 @@ { - "version": "0.2.2", + "version": "0.2.22", "common_args": { - "model_name_or_path": "./models/Qwen2.5-3B-Instruct", + "model_name_or_path": "./models/Qwen3-4B", "adapter_name_or_path": "./model_output", //同时做为train_sft_args的output_dir - "template": "qwen", + "template": "qwen3", "default_system": "请你扮演一名人类，不要说自己是人工智能", "finetuning_type": "lora", + "media_dir": "dataset/media", + "image_max_pixels": 209920, //720P + "enable_thinking": false, "trust_remote_code": true }, "cli_args": { @@ -15,25 +18,28 @@ //数据处理配置 "platform": "wechat", "include_type": [ - "text" - ], - "blocked_words": [ // 禁用词 - "例如 姓名", - "例如 密码", - "//....." + "text", + // "image" ], + "max_image_num": 2, // 单条数据最大图片数量 "single_combine_strategy": "time_window", // 单人组成单句策略 "qa_match_strategy": "time_window", // 组成qa策略 "single_combine_time_window": 2, // 单人组成单句时间窗口（分钟）, "qa_match_time_window": 5, // 组成qa时间窗口（分钟）, "combine_msg_max_length": 256, // 组合后消息最大长度 配合cutoff_len 使用 - "prompt_with_history": false, // 是否在prompt中包含历史对话 "clean_dataset": { - "enable_clean": true, + "enable_clean": false, "clean_strategy": "llm", "llm": { "accept_score": 2, //可以接受的llm打分阈值,1分最差，5分最好,低于此分数的数据不会用于训练 } + }, + "vision_api": { + "enable": false, // 设置为 true 来开启此功能 + "api_key": "xxx", + "api_url": "https://xxx/v1", // 例如阿里云，或替换为其他兼容OpenAI的API地址 + "model_name": "xxx", // 要使用的多模态模型名称,例如qwen-vl-max + "max_workers": 5 // 并行调用API的线程数，最多不要超过8 } }, "test_model_args": { @@ -44,16 +50,17 @@ "stage": "sft", "dataset": "wechat-sft", "dataset_dir": "./dataset/res_csv/sft", + "freeze_multi_modal_projector": false, //MLLM 训练时是否冻结多模态投影器。 "use_fast_tokenizer": true, - "lora_target": "q_proj,v_proj", - "lora_rank": 4, + "lora_target": "q_proj,v_proj,visual.merger.mlp.0,visual.merger.mlp.2", + "lora_rank": 2, "lora_dropout": 0.3, "weight_decay": 0.1, "overwrite_cache": true, - "per_device_train_batch_size": 8, - "gradient_accumulation_steps": 4, + "per_device_train_batch_size": 4, + "gradient_accumulation_steps": 8, "lr_scheduler_type": "cosine", - "cutoff_len": 256, + "cutoff_len": 1024, "logging_steps": 5, "save_steps": 10, "learning_rate": 1e-4, @@ -69,8 +76,5 @@ "temperature": 0.5, "max_length": 50, "top_p": 0.65 - }, - "vllm_args": { - "gpu_memory_utilization": 0.90 } } diff --git a/tests/test_full_pipe.py b/tests/test_full_pipe.py deleted file mode 100644 index a183c61..0000000 --- a/tests/test_full_pipe.py +++ /dev/null @@ -1,173 +0,0 @@ -import functools -import os -import shutil -import subprocess -import sys -import time -from typing import Optional, Union, cast -from unittest import mock - -import pytest - -from weclone.utils.config import load_config -from weclone.utils.config_models import DataModality, WCMakeDatasetConfig -from weclone.utils.log import logger - -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) -PROJECT_ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) -server_process: Optional[subprocess.Popen] = None - -test_logger = logger.bind() -test_logger.remove() -test_logger.add( - sys.stderr, - format="{message}", - colorize=True, - level="INFO", -) - -def print_test_header(test_name: str): - line_length = 100 - test_logger.info("\n" + "─" * line_length) - title = f" Testing Phase: {test_name} " - padding_total = line_length - len(title) - padding_left = padding_total // 2 - padding_right = padding_total - padding_left - test_logger.info(" " * padding_left + title + " " * padding_right) - test_logger.info("─" * line_length) - -def setup_make_dataset_test_data(): - PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) - DATASET_CSV_DIR = os.path.join(PROJECT_ROOT, "dataset", "csv") - - TESTS_DIR = os.path.dirname(__file__) - TEST_DATA_PERSON_DIR = os.path.join(TESTS_DIR, "tests_data", "test_person") - - # 先删除目录，再重新创建 - if os.path.exists(DATASET_CSV_DIR): - shutil.rmtree(DATASET_CSV_DIR) - os.makedirs(DATASET_CSV_DIR) - - # 创建test_person子目录 - test_person_csv_dir = os.path.join(DATASET_CSV_DIR, "test_person") - os.makedirs(test_person_csv_dir) - - # 复制测试数据到test_person目录 - for item_name in os.listdir(TEST_DATA_PERSON_DIR): - source_item_path = os.path.join(TEST_DATA_PERSON_DIR, item_name) - if os.path.isfile(source_item_path) and item_name.lower().endswith('.csv'): - destination_item_path = os.path.join(test_person_csv_dir, item_name) - shutil.copy2(source_item_path, destination_item_path) - -def run_cli_command(command: list[str], timeout: int | None = None, background: bool = False) -> Union[subprocess.CompletedProcess, subprocess.Popen]: - """Execute a CLI command and return the result. - - Args: - command: List of commands to execute. - timeout: Timeout in seconds. - background: Whether to run in the background. - - Returns: - If background=True, returns a Popen object; otherwise, returns a CompletedProcess object. - """ - env = os.environ.copy() - env["WECLONE_CONFIG_PATH"] = "tests/full_pipe.jsonc" # Set environment variable - - if background: - process = subprocess.Popen( - [sys.executable, "-m", "weclone.cli"] + command, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - text=True, - cwd=PROJECT_ROOT_DIR, - env=env - ) - time.sleep(2) - return process - else: - process = subprocess.run( - [sys.executable, "-m", "weclone.cli"] + command, - stderr=None, - stdout=None, - text=True, - cwd=PROJECT_ROOT_DIR, # Execute in the project root directory - timeout=timeout, - env=env # Pass the modified environment variables - ) - return process - -@pytest.mark.order(1) -def test_cli_make_dataset(): - """Test the make-dataset command.""" - print_test_header("make-dataset") - config: WCMakeDatasetConfig = cast(WCMakeDatasetConfig, load_config("make_dataset")) - if DataModality.IMAGE in config.include_type: - #复制图片到media_dir/iamges - os.makedirs(config.media_dir, exist_ok=True) - os.makedirs(os.path.join(config.media_dir, "images"), exist_ok=True) - for file in os.listdir(os.path.join(PROJECT_ROOT_DIR, "tests", "tests_data", "test_person")): - shutil.copy(os.path.join(PROJECT_ROOT_DIR, "tests", "tests_data", "test_person", file), os.path.join(config.media_dir, "images", file)) - - setup_make_dataset_test_data() - result = run_cli_command(["make-dataset"]) - assert result.returncode == 0, "make-dataset command execution failed" - -@pytest.mark.order(2) -def test_cli_train_sft(): - """Test the train-sft command.""" - print_test_header("train-sft") - if os.path.exists("model_output"): - shutil.rmtree("model_output") - try: - result = run_cli_command(["train-sft"]) - assert result.returncode == 0, "train-sft command failed or did not fail fast as expected" - except subprocess.TimeoutExpired: - test_logger.info("train-sft command terminated due to timeout, which is acceptable in testing, indicating the command has started execution.") - pass - except Exception as e: - pytest.fail(f"An unexpected error occurred during train-sft command execution: {e}") - -@pytest.mark.order(3) -def test_cli_webchat_demo(): - """Test the webchat-demo command.""" - print_test_header("webchat-demo") - - with mock.patch("weclone.eval.web_demo.main") as mock_main: - mock_main.return_value = None - try: - result = run_cli_command(["webchat-demo"], timeout=5) - assert result.returncode == 0, "webchat-demo command execution failed" - except subprocess.TimeoutExpired: - pass - -@pytest.mark.order(4) -def test_cli_server(): - """Test the server command. - - Start the server in the background, without blocking subsequent tests. - """ - print_test_header("server (background)") - global server_process - server_process = cast(subprocess.Popen, run_cli_command(["server"], background=True)) - assert server_process.poll() is None, "Server startup failed" - test_logger.info("服务器已在后台启动") - -@pytest.mark.order(5) -def test_cli_test_model(): - """Test the test-model command. - - Use the server for testing, and shut down the server after the test is complete. - """ - print_test_header("test-model") - try: - result = run_cli_command(["test-model"]) - assert result.returncode == 0, "test-model command execution failed" - finally: - global server_process - if server_process is not None and server_process.poll() is None: - test_logger.info("测试完成，正在关闭服务器...") - server_process.terminate() - server_process.wait(timeout=5) - if server_process.poll() is None: - server_process.kill() # Force kill if the process hasn't terminated - test_logger.info("服务器已关闭") diff --git a/tests/test_full_pipeV2.py b/tests/test_full_pipeV2.py index e244b57..e905779 100644 --- a/tests/test_full_pipeV2.py +++ b/tests/test_full_pipeV2.py @@ -15,7 +15,10 @@ from weclone.utils.log import logger sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) PROJECT_ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) -server_process: Optional[subprocess.Popen] = None +PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +DATASET_CSV_DIR = os.path.join(PROJECT_ROOT, "dataset", "csv") +TESTS_DIR = os.path.dirname(__file__) +TEST_DATA_PERSON_DIR = os.path.join(TESTS_DIR, "tests_data", "test_person") test_logger = logger.bind() test_logger.remove() @@ -26,33 +29,49 @@ test_logger.add( level="INFO", ) -def print_test_header(test_name: str): +def get_config_files(): + """获取所有配置文件""" + configs_dir = os.path.join(os.path.dirname(__file__), "configs") + config_files = [] + for file in os.listdir(configs_dir): + if file.endswith('.jsonc'): + config_files.append(f"tests/configs/{file}") + return config_files + +def print_test_header(test_name: str, config_file: str = ""): line_length = 100 test_logger.info("\n" + "─" * line_length) - title = f" Testing Phase: {test_name} " + if config_file: + title = f" Testing Phase: {test_name} | Config: {os.path.basename(config_file)} " + else: + title = f" Testing Phase: {test_name} " padding_total = line_length - len(title) padding_left = padding_total // 2 padding_right = padding_total - padding_left test_logger.info(" " * padding_left + title + " " * padding_right) test_logger.info("─" * line_length) -def setup_make_dataset_test_data(): - PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) - DATASET_CSV_DIR = os.path.join(PROJECT_ROOT, "dataset", "csv") - - TESTS_DIR = os.path.dirname(__file__) - TEST_DATA_PERSON_DIR = os.path.join(TESTS_DIR, "tests_data", "test_person") +def print_config_header(config_file: str): + """打印配置文件开始测试的头部""" + line_length = 120 + test_logger.info("\n" + "═" * line_length) + title = f" 开始测试配置文件: {os.path.basename(config_file)} " + padding_total = line_length - len(title) + padding_left = padding_total // 2 + padding_right = padding_total - padding_left + test_logger.info(" " * padding_left + title + " " * padding_right) + test_logger.info("═" * line_length) - # 先删除目录，再重新创建 +def set_test_env(): + if os.path.exists("model_output"): + shutil.rmtree("model_output") if os.path.exists(DATASET_CSV_DIR): shutil.rmtree(DATASET_CSV_DIR) os.makedirs(DATASET_CSV_DIR) - # 创建test_person子目录 test_person_csv_dir = os.path.join(DATASET_CSV_DIR, "test_person") os.makedirs(test_person_csv_dir) - # 复制测试数据到test_person目录 for item_name in os.listdir(TEST_DATA_PERSON_DIR): source_item_path = os.path.join(TEST_DATA_PERSON_DIR, item_name) if os.path.isfile(source_item_path) and item_name.lower().endswith('.csv'): @@ -60,11 +79,12 @@ def setup_make_dataset_test_data(): shutil.copy2(source_item_path, destination_item_path) -def run_cli_command(command: list[str], timeout: int | None = None, background: bool = False) -> Union[subprocess.CompletedProcess, subprocess.Popen]: +def run_cli_command(command: list[str], config_path: str, timeout: int | None = None, background: bool = False) -> Union[subprocess.CompletedProcess, subprocess.Popen]: """Execute a CLI command and return the result. Args: command: List of commands to execute. + config_path: Path to the configuration file. timeout: Timeout in seconds. background: Whether to run in the background. @@ -72,13 +92,13 @@ def run_cli_command(command: list[str], timeout: int | None = None, background: If background=True, returns a Popen object; otherwise, returns a CompletedProcess object. """ env = os.environ.copy() - env["WECLONE_CONFIG_PATH"] = "tests/full_pipeV2.jsonc" # Set environment variable + env["WECLONE_CONFIG_PATH"] = config_path # Set environment variable if background: process = subprocess.Popen( [sys.executable, "-m", "weclone.cli"] + command, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, + stderr=None, + stdout=None, text=True, cwd=PROJECT_ROOT_DIR, env=env @@ -97,11 +117,25 @@ def run_cli_command(command: list[str], timeout: int | None = None, background: ) return process -@pytest.mark.order(5) -def test_cli_make_dataset(): - """Test the make-dataset command.""" - print_test_header("make-dataset") - config: WCMakeDatasetConfig = cast(WCMakeDatasetConfig, load_config("make_dataset")) +def load_config_with_path(config_file: str, config_section: str): + """临时设置环境变量并加载配置""" + original_env = os.environ.get("WECLONE_CONFIG_PATH") + os.environ["WECLONE_CONFIG_PATH"] = config_file + + try: + return load_config(config_section) + finally: + # 恢复原始环境变量 + if original_env is not None: + os.environ["WECLONE_CONFIG_PATH"] = original_env + elif "WECLONE_CONFIG_PATH" in os.environ: + del os.environ["WECLONE_CONFIG_PATH"] + +def run_make_dataset_test(config_file: str): + """执行 make-dataset 测试""" + print_test_header("make-dataset", config_file) + + config: WCMakeDatasetConfig = cast(WCMakeDatasetConfig, load_config_with_path(config_file, "make_dataset")) if DataModality.IMAGE in config.include_type: #复制图片到media_dir/iamges os.makedirs(config.media_dir, exist_ok=True) @@ -109,69 +143,85 @@ def test_cli_make_dataset(): for file in os.listdir(os.path.join(PROJECT_ROOT_DIR, "tests", "tests_data", "images")): shutil.copy(os.path.join(PROJECT_ROOT_DIR, "tests", "tests_data", "images", file), os.path.join(config.media_dir, "images", file)) - setup_make_dataset_test_data() - result = run_cli_command(["make-dataset"]) - assert result.returncode == 0, "make-dataset command execution failed" + result = run_cli_command(["make-dataset"], config_file) + assert result.returncode == 0, f"make-dataset command execution failed for config {config_file}" -@pytest.mark.order(6) -def test_cli_train_sft(): - """Test the train-sft command.""" - print_test_header("train-sft") - if os.path.exists("model_output"): - shutil.rmtree("model_output") +def run_train_sft_test(config_file: str): + """执行 train-sft 测试""" + print_test_header("train-sft", config_file) + try: - result = run_cli_command(["train-sft"]) - assert result.returncode == 0, "train-sft command failed or did not fail fast as expected" + result = run_cli_command(["train-sft"], config_file) + assert result.returncode == 0, f"train-sft command failed or did not fail fast as expected for config {config_file}" except subprocess.TimeoutExpired: - test_logger.info("train-sft command terminated due to timeout, which is acceptable in testing, indicating the command has started execution.") + test_logger.info(f"train-sft command terminated due to timeout for config {config_file}, which is acceptable in testing, indicating the command has started execution.") pass except Exception as e: - pytest.fail(f"An unexpected error occurred during train-sft command execution: {e}") + pytest.fail(f"An unexpected error occurred during train-sft command execution for config {config_file}: {e}") -@pytest.mark.order(7) -def test_cli_webchat_demo(): - """Test the webchat-demo command.""" - print_test_header("webchat-demo") +def run_webchat_demo_test(config_file: str): + """执行 webchat-demo 测试""" + print_test_header("webchat-demo", config_file) with mock.patch("weclone.eval.web_demo.main") as mock_main: mock_main.return_value = None try: - result = run_cli_command(["webchat-demo"], timeout=30) - assert result.returncode == 0, "webchat-demo command execution failed" + result = run_cli_command(["webchat-demo"], config_file, timeout=20) + assert result.returncode == 0, f"webchat-demo command execution failed for config {config_file}" except subprocess.TimeoutExpired: pass -@pytest.mark.order(8) -def test_cli_server(): - """Test the server command. - - Start the server in the background, without blocking subsequent tests. - """ - print_test_header("server (background)") - global server_process - server_process = cast(subprocess.Popen, run_cli_command(["server"], background=True)) - assert server_process.poll() is None, "Server startup failed" - test_logger.info("服务器已在后台启动") +def run_server_test(config_file: str) -> subprocess.Popen: + """执行 server 测试，返回进程对象""" + print_test_header("server (background)", config_file) + server_process = cast(subprocess.Popen, run_cli_command(["server"], config_file, background=True)) + test_logger.info("等待服务器启动，20秒后检查状态...") + time.sleep(20) + assert server_process.poll() is None, f"Server startup failed for config {config_file}" + test_logger.info(f"使用配置 {config_file} 的服务器已在后台启动") + return server_process -@pytest.mark.order(9) -def test_cli_test_model(): - """Test the test-model command. - - Use the server for testing, and shut down the server after the test is complete. - """ - print_test_header("test-model") +def run_test_model_test(config_file: str, server_process: subprocess.Popen): + """执行 test-model 测试并关闭服务器""" + print_test_header("test-model", config_file) try: - result = run_cli_command(["test-model"]) - assert result.returncode == 0, "test-model command execution failed" + result = run_cli_command(["test-model"], config_file) + assert result.returncode == 0, f"test-model command execution failed for config {config_file}" finally: - global server_process if server_process is not None and server_process.poll() is None: - test_logger.info("测试完成，正在关闭服务器...") + test_logger.info(f"测试完成，正在关闭使用配置 {config_file} 的服务器...") server_process.terminate() server_process.wait(timeout=5) if server_process.poll() is None: server_process.kill() # Force kill if the process hasn't terminated test_logger.info("服务器已关闭") +@pytest.mark.parametrize("config_file", get_config_files()) +def test_full_pipeline_for_config(config_file): + """为每个配置文件完整执行所有测试步骤""" + print_config_header(config_file) + + set_test_env() + + server_process = None + try: + # 按顺序执行所有测试步骤 + run_make_dataset_test(config_file) + run_train_sft_test(config_file) + run_webchat_demo_test(config_file) + server_process = run_server_test(config_file) + run_test_model_test(config_file, server_process) + + test_logger.info(f"✅ 配置文件 {os.path.basename(config_file)} 的所有测试已完成") + + except Exception as e: + test_logger.error(f"❌ 配置文件 {os.path.basename(config_file)} 测试失败: {e}") + if server_process is not None and server_process.poll() is None: + server_process.terminate() + server_process.wait(timeout=5) + if server_process.poll() is None: + server_process.kill() + raise + if __name__ == "__main__": - setup_make_dataset_test_data() + set_test_env() diff --git a/weclone/server/api_service.py b/weclone/server/api_service.py index b07d027..cb2a289 100644 --- a/weclone/server/api_service.py +++ b/weclone/server/api_service.py @@ -9,7 +9,7 @@ from weclone.utils.config import load_config def main(): config = load_config("api_service") - chat_model = ChatModel(config) + chat_model = ChatModel(config.model_dump(mode="json")) app = create_app(chat_model) print("Visit http://localhost:{}/docs for API document.".format(os.environ.get("API_PORT", 8005))) uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("API_PORT", 8005)), workers=1)