From cd14378554e4f0a6f2a3ec77ae3d0b9905c9436e Mon Sep 17 00:00:00 2001 From: yujia Date: Mon, 7 Jul 2025 17:39:26 +0800 Subject: [PATCH 1/3] feat: add code detect fasttext model --- llm_web_kit/model/code_detector.py | 217 +++++-------- tests/llm_web_kit/model/test_code_detector.py | 295 +++++++++++++++--- 2 files changed, 335 insertions(+), 177 deletions(-) diff --git a/llm_web_kit/model/code_detector.py b/llm_web_kit/model/code_detector.py index c419d1c4..f0e40fff 100644 --- a/llm_web_kit/model/code_detector.py +++ b/llm_web_kit/model/code_detector.py @@ -1,189 +1,130 @@ -# Copyright (c) Opendatalab. All rights reserved. import os import re -from typing import Dict, Tuple +from typing import Dict import fasttext +import jieba from llm_web_kit.config.cfg_reader import load_config from llm_web_kit.libs.logger import mylogger as logger from llm_web_kit.model.resource_utils import (CACHE_DIR, download_auto_file, - get_unzip_dir, - singleton_resource_manager, - unzip_local_file) + singleton_resource_manager) class CodeClassification: - """code classification using fasttext model.""" + """Code classification using fasttext model with jieba preprocessing.""" - def __init__(self, model_path: str = None): - """Initialize CodeClassification model Will download the v3_1223.bin - model if model_path is not provided. + # Constants for text preprocessing + MAX_WORD_NUM = 20480 + + def __init__(self, model_path: str = None, resource_name=None): + """Initialize CodeClassification model. Args: model_path (str, optional): Path to the model. Defaults to None. """ - if not model_path: - model_path = self.auto_download() + model_path = self.auto_download(resource_name) self.model = fasttext.load_model(model_path) - def auto_download(self): - """Default download the v3_1223.bin model.""" - resource_name = 'code_fasttext_models_v3_1223' + def auto_download(self, resource_name: str = None) -> str: + if resource_name is None: + resource_name = 'code_detect_v4_0409' + resource_config = load_config()['resources'] - code_cl_v3_config: dict = resource_config[resource_name] - code_cl_v3_path = code_cl_v3_config['download_path'] - code_cl_v3_md5 = code_cl_v3_config.get('md5', '') - # get the zip path calculated by the s3 path - zip_path = os.path.join(CACHE_DIR, f'{resource_name}.zip') - # the unzip path is calculated by the zip path - unzip_path = get_unzip_dir(zip_path) - logger.info(f'try to make unzip_path: {unzip_path}') - # if the unzip path does not exist, download the zip file and unzip it - if not os.path.exists(unzip_path): - logger.info(f'unzip_path: {unzip_path} does not exist') - logger.info(f'try to unzip from zip_path: {zip_path}') - if not os.path.exists(zip_path): - logger.info(f'zip_path: {zip_path} does not exist') - logger.info(f'downloading {code_cl_v3_path}') - zip_path = download_auto_file(code_cl_v3_path, zip_path, code_cl_v3_md5) - logger.info(f'unzipping {zip_path}') - unzip_path = unzip_local_file(zip_path, unzip_path) - else: - logger.info(f'unzip_path: {unzip_path} exist') - return os.path.join(unzip_path, f'{resource_name}.bin') + code_cl_v4_config: dict = resource_config[resource_name] + model_download_path = code_cl_v4_config['download_path'] + model_md5 = code_cl_v4_config.get('md5', '') + + target_dir = os.path.join(CACHE_DIR, 'resource_name') + os.makedirs(target_dir, exist_ok=True) + target_path = os.path.join(target_dir, f'{resource_name}.bin') + + logger.info(f'开始下载模型 {resource_name} -> {target_path}') + return download_auto_file(resource_path=model_download_path, target_path=target_path, md5_sum=model_md5) @property def version(self) -> str: - """ - Get the version of the model - The version is determined by the update of model - now have v3 version from : s3://web-parse-huawei/shared_resource/code/code_fasttext_models_v3_1223.zip + """Get the version of the model. Returns: str: The version of the model """ if not hasattr(self, '_version'): - self._version = 'v3_1223' + self._version = 'v4_0409' return self._version - def predict(self, text: str, k: int = 2) -> Tuple[Tuple[str], Tuple[float]]: - """Predict code of the given text Return all predictions. Default k is - 2. - + def jieba_cut(self, text: str) -> str: + """Preprocess text using jieba segmentation and cleaning. Args: - text (str): Text to predict code - k (int, optional): Number of predictions to return. Defaults to 2. + text (str): Input text to preprocess Returns: - Tuple[Tuple[str], Tuple[float]]: Tuple of predictions and probabilities return top k predictions + str: Processed text string """ - assert k > 0, 'k should be greater than 0' - - # remove new lines - text = text.replace('\n', ' ') + # Define stop words including the specified ones + stop_words = {'\r', '\n', ' ', '', '\r\n'} - # returns top k predictions - predictions, probabilities = self.model.predict(text, k=k) + words = jieba.lcut(text) + filtered_words = [word for word in words if word not in stop_words and word.strip()] + if len(filtered_words) > self.MAX_WORD_NUM: + filtered_words = filtered_words[: self.MAX_WORD_NUM] - return predictions, probabilities + seg_text = ' '.join(filtered_words) + seg_text = re.sub(r'([.\!?,\'/()\"\n])', '', seg_text) + seg_text = re.sub(r'\s+', ' ', seg_text) + seg_text = seg_text.lower() + seg_text = seg_text.strip() + return seg_text -def get_singleton_code_detect() -> CodeClassification: - """Get the singleton code classification model. - - returns: - CodeClassification: The code classification model - """ - if not singleton_resource_manager.has_name('code_detect'): - singleton_resource_manager.set_resource('code_detect', CodeClassification()) - return singleton_resource_manager.get_resource('code_detect') + def predict(self, text: str) -> Dict[str, float]: + """Predict code classification. + Args: + text (str): Input text to classify -def detect_latex_env(content_str: str) -> bool: - """Detect if the content string contains latex environment.""" - latex_env_pattern = re.compile(r'\\begin\{.*?\}.*\\end\{.*\}', re.DOTALL) - return latex_env_pattern.search(content_str) is not None + Returns: + Dict[str, float]: Dictionary with key "has_code_prob_0409" and prediction value + """ + seg_text = self.jieba_cut(text) + predictions, probabilities = self.model.predict(seg_text) + + if len(predictions) > 0 and len(probabilities) > 0: + label = predictions[0] + confidence = probabilities[0] + + label = label.replace('__label__', '') + confidence = min(confidence, 1.0) + if label == '1': + prediction_value = confidence + else: + prediction_value = 1.0 - confidence + else: + prediction_value = 0.0 + return {'has_code_prob_0409': prediction_value} -CODE_CL_SUPPORTED_VERSIONS = ['v3_1223'] +def get_singleton_code_detect() -> CodeClassification: + """Get the singleton code classification v2 model. -def decide_code_func(content_str: str, code_detect: CodeClassification) -> float: - """Decide code based on the content string. This function will truncate the - content string if it is too long. + Returns: + CodeClassification: The code classification model + """ + if not singleton_resource_manager.has_name('code_detect_v4_0409'): + singleton_resource_manager.set_resource('code_detect_v4_0409', CodeClassification()) + return singleton_resource_manager.get_resource('code_detect_v4_0409') - Raises: - ValueError: Unsupported version. - The prediction str is different for different versions of fasttext model. - So the version should be specified. - Now only support version "v3_1223" - Warning: - The too long content string may be truncated. - Some text with latex code block may be misclassified. +def update_code_prob_by_str(content_str: str) -> Dict[str, float]: + """Predict code classification using model with detailed output. Args: - content_str (str): The content string to decide code - code_detect (CodeClassification): The code classification model + content_str (str): The content string to classify Returns: - float: the probability of code text + Dict[str, float]: Dictionary with key "has_code_prob_0409" and prediction value """ - - # truncate the content string if it is too long - str_len = len(content_str) - if str_len > 10000: - logger.warning('Content string is too long, truncate to 10000 characters') - start_idx = (str_len - 10000) // 2 - content_str = content_str[start_idx : start_idx + 10000] - - # check if the content string contains latex environment - if detect_latex_env(content_str): - logger.warning( - 'Content string contains latex environment, may be misclassified' - ) - - def decide_code_by_prob_v3( - predictions: Tuple[str], probabilities: Tuple[float] - ) -> float: - idx = predictions.index('__label__1') - true_prob = probabilities[idx] - return true_prob - - if code_detect.version == 'v3_1223': - predictions, probabilities = code_detect.predict(content_str) - result = decide_code_by_prob_v3(predictions, probabilities) - else: - raise ValueError( - f'Unsupported version: {code_detect.version}. Supported versions: {[CODE_CL_SUPPORTED_VERSIONS]}' - ) - return result - - -def decide_code_by_str(content_str: str) -> float: - """Decide code based on the content string, based on decide_code_func.""" - lang_detect = get_singleton_code_detect() - - return decide_code_func(content_str, lang_detect) - - -def update_code_by_str(content_str: str) -> Dict[str, float]: - """Decide code based on the content string, based on decide_code_func.""" - return {'code': decide_code_by_str(content_str)} - - -if __name__ == '__main__': - cclass = CodeClassification() - text = '示例文本' - predictions, probabilities = cclass.predict(text) - print(predictions, probabilities) - print(update_code_by_str(text)) - - text = 'import pandas as pd' - print(update_code_by_str(text)) - - text = "# 这是一个代码测试模块\nimport pandas as pd\ndf=pd.read_csv('file.csv')" - predictions, probabilities = cclass.predict(text) - print(update_code_by_str(text)) + code_detect = get_singleton_code_detect() + return code_detect.predict(content_str) diff --git a/tests/llm_web_kit/model/test_code_detector.py b/tests/llm_web_kit/model/test_code_detector.py index 89877ba0..1be4fc99 100644 --- a/tests/llm_web_kit/model/test_code_detector.py +++ b/tests/llm_web_kit/model/test_code_detector.py @@ -1,64 +1,281 @@ -import unittest +import logging +import warnings from unittest import TestCase from unittest.mock import MagicMock, patch +from llm_web_kit.exception.exception import ModelResourceException from llm_web_kit.model.code_detector import (CodeClassification, - decide_code_by_str, - decide_code_func, - detect_latex_env, - update_code_by_str) + get_singleton_code_detect, + update_code_prob_by_str) +# Suppress external dependency warnings for clean pytest runs +warnings.filterwarnings('ignore', category=DeprecationWarning, module='jieba') +warnings.filterwarnings('ignore', category=DeprecationWarning, module='pkg_resources') +warnings.filterwarnings('ignore', category=DeprecationWarning, module='jupyter_client') -class TestCodeClassification(TestCase): + +class TestEnvironment: + """Manages test environment and determines if real model downloads are + available.""" + + _real_model_available = None + + @classmethod + def can_download_real_model(cls) -> bool: + """Check if real model download is possible.""" + if cls._real_model_available is not None: + return cls._real_model_available + + try: + CodeClassification() + cls._real_model_available = True + return True + except (ModelResourceException, ConnectionError, TimeoutError, OSError, Exception) as e: + logging.info(f'Real model download failed, using mocked tests: {e}') + cls._real_model_available = False + return False + + @classmethod + def reset(cls): + """Reset environment detection for testing.""" + cls._real_model_available = None + + +class BaseTestCase(TestCase): + """Base test case with common functionality.""" + + def assert_prediction_format(self, result): + """Assert that prediction result has correct format.""" + self.assertIsInstance(result, dict) + self.assertIn('has_code_prob_0409', result) + self.assertIsInstance(result['has_code_prob_0409'], (int, float)) + self.assertGreaterEqual(result['has_code_prob_0409'], 0.0) + self.assertLessEqual(result['has_code_prob_0409'], 1.0) + + def get_test_samples(self): + """Get test samples for code detection.""" + return { + 'code': ['import pandas as pd', 'def hello():\n pass', 'SELECT * FROM users'], + 'non_code': ['这是普通文本', 'Regular English text', 'Lorem ipsum'], + 'edge': ['', ' \t\n\r ', '!@#$%', 'a' * 1000] + } + + def setup_mock_classifier(self, mock_auto_download, mock_load_model): + """Setup a mocked classifier with standard configuration.""" + mock_auto_download.return_value = '/fake/model/path' + classifier = CodeClassification() + return classifier + + +class TestCodeClassificationIntegration(BaseTestCase): + """Integration tests using real model downloads when available.""" + + @classmethod + def setUpClass(cls): + """Set up class-level resources.""" + cls.use_real_model = TestEnvironment.can_download_real_model() + if not cls.use_real_model: + logging.info('Real model unavailable, skipping integration tests') + + def setUp(self): + """Set up test fixtures.""" + if not self.use_real_model: + self.skipTest('Real model not available') + + def test_real_model_functionality(self): + """Test real model initialization and basic functionality.""" + classifier = CodeClassification() + self.assertIsNotNone(classifier.model) + self.assertEqual(classifier.version, 'v4_0409') + + # Test with code and non-code samples + samples = self.get_test_samples() + + # Test code detection + for code_sample in samples['code']: + result = classifier.predict(code_sample) + self.assert_prediction_format(result) + + # Test non-code detection + for non_code_sample in samples['non_code']: + result = classifier.predict(non_code_sample) + self.assert_prediction_format(result) + + # Test edge cases + for edge_case in samples['edge']: + result = classifier.predict(edge_case) + self.assert_prediction_format(result) + + def test_real_model_singleton(self): + """Test singleton behavior with real model.""" + instance1 = get_singleton_code_detect() + instance2 = get_singleton_code_detect() + self.assertIs(instance1, instance2) + + result1 = instance1.predict('import pandas as pd') + result2 = instance2.predict('import pandas as pd') + self.assertEqual(result1, result2) + + +class TestCodeClassificationMocked(BaseTestCase): + """Mocked tests that always run regardless of model availability.""" @patch('llm_web_kit.model.code_detector.fasttext.load_model') @patch('llm_web_kit.model.code_detector.CodeClassification.auto_download') - def test_init(self, mock_auto_download, mock_load_model): + def test_initialization(self, mock_auto_download, mock_load_model): + """Test initialization with both default and custom paths.""" + # Test default path mock_auto_download.return_value = '/fake/model/path' - # Test with default model path - _ = CodeClassification() - mock_load_model.assert_called_once_with('/fake/model/path') + CodeClassification() + mock_auto_download.assert_called_once_with(None) + mock_load_model.assert_called_with('/fake/model/path') - # Test with custom model path + # Test custom path mock_load_model.reset_mock() - _ = CodeClassification('custom_model_path') - mock_load_model.assert_called_once_with('custom_model_path') + CodeClassification('custom_model_path') + mock_load_model.assert_called_with('custom_model_path') @patch('llm_web_kit.model.code_detector.fasttext.load_model') @patch('llm_web_kit.model.code_detector.CodeClassification.auto_download') - def test_predict(self, mock_auto_download, mock_load_model): - code_cl = CodeClassification() - code_cl.model.predict.return_value = (['label1', 'label2'], [0.9, 0.1]) - predictions, probabilities = code_cl.predict('test text') - assert predictions == ['label1', 'label2'] - assert probabilities == [0.9, 0.1] + @patch('llm_web_kit.model.code_detector.jieba.lcut') + def test_jieba_cut_functionality(self, mock_jieba_lcut, mock_auto_download, mock_load_model): + """Test jieba_cut text preprocessing functionality.""" + classifier = self.setup_mock_classifier(mock_auto_download, mock_load_model) + + # Test basic functionality + mock_jieba_lcut.return_value = ['这是', '一个', '测试', '文本'] + result = classifier.jieba_cut('这是一个测试文本') + self.assertIsInstance(result, str) + self.assertIn('这是', result) + self.assertIn('测试', result) + + # Test stop words removal + mock_jieba_lcut.return_value = ['这是', '\r', '一个', '\n', '测试', ' ', '文本', ''] + result = classifier.jieba_cut('这是\r一个\n测试 文本') + self.assertNotIn('\r', result) + self.assertNotIn('\n', result) + self.assertIn('这是', result) + + # Test truncation + long_word_list = ['word'] * (CodeClassification.MAX_WORD_NUM + 100) + mock_jieba_lcut.return_value = long_word_list + result = classifier.jieba_cut('very long text') + word_count = len(result.split()) + self.assertLessEqual(word_count, CodeClassification.MAX_WORD_NUM) + + # Test special character removal + mock_jieba_lcut.return_value = ['测试', '文本', '!', '?', ','] + result = classifier.jieba_cut('测试文本!?,') + self.assertNotIn('!', result) + self.assertNotIn('?', result) + + # Test edge cases + mock_jieba_lcut.return_value = [] + result = classifier.jieba_cut('') + self.assertEqual(result, '') + + @patch('llm_web_kit.model.code_detector.fasttext.load_model') + @patch('llm_web_kit.model.code_detector.CodeClassification.auto_download') + def test_predict_functionality(self, mock_auto_download, mock_load_model): + """Test predict method with various scenarios.""" + classifier = self.setup_mock_classifier(mock_auto_download, mock_load_model) + + # Test code label (1) + classifier.model.predict.return_value = (['__label__1'], [0.8]) + result = classifier.predict('import pandas as pd') + self.assert_prediction_format(result) + self.assertEqual(result['has_code_prob_0409'], 0.8) + + # Test non-code label (0) + classifier.model.predict.return_value = (['__label__0'], [0.7]) + result = classifier.predict('这是普通文本') + self.assert_prediction_format(result) + self.assertAlmostEqual(result['has_code_prob_0409'], 0.3, places=5) # 1 - 0.7 + + # Test confidence capping + classifier.model.predict.return_value = (['__label__1'], [1.5]) + result = classifier.predict('test') + self.assertEqual(result['has_code_prob_0409'], 1.0) + + # Test empty predictions + classifier.model.predict.return_value = ([], []) + result = classifier.predict('test') + self.assertEqual(result['has_code_prob_0409'], 0.0) + + # Test version property + self.assertEqual(classifier.version, 'v4_0409') + + +class TestErrorHandlingAndUtilities(BaseTestCase): + """Test cases for error handling and utility functions.""" + + @patch('llm_web_kit.model.code_detector.CodeClassification.auto_download') + @patch('llm_web_kit.model.code_detector.fasttext.load_model') + def test_download_failure_scenarios(self, mock_load_model, mock_auto_download): + """Test various download failure scenarios.""" + # Test network failure + mock_auto_download.side_effect = ConnectionError('Network unreachable') + with self.assertRaises(ConnectionError): + CodeClassification() + + # Test model resource exception + mock_auto_download.side_effect = ModelResourceException('Model download failed') + with self.assertRaises(ModelResourceException): + CodeClassification() + # Test timeout + mock_auto_download.side_effect = TimeoutError('Download timeout') + with self.assertRaises(TimeoutError): + CodeClassification() -def test_detect_latex_env(): - assert detect_latex_env('\\begin{equation}\nx = y\n\\end{equation}') - assert not detect_latex_env('This is not a latex environment') + # Test file not found + mock_auto_download.side_effect = None + mock_auto_download.return_value = '/nonexistent/path' + mock_load_model.side_effect = OSError('File not found') + with self.assertRaises(OSError): + CodeClassification() + def test_environment_detection_caching(self): + """Test that environment detection results are cached properly.""" + TestEnvironment.reset() + result1 = TestEnvironment.can_download_real_model() + result2 = TestEnvironment.can_download_real_model() + self.assertEqual(result1, result2) -def test_decide_code_func(): - code_detect = MagicMock() - code_detect.version = 'v3_1223' - code_detect.predict.return_value = (['__label__1', '__label__0'], [0.6, 0.4]) - assert decide_code_func('test text', code_detect) == 0.6 + TestEnvironment.reset() + result3 = TestEnvironment.can_download_real_model() + self.assertEqual(result1, result3) + @patch('llm_web_kit.model.code_detector.singleton_resource_manager') + @patch('llm_web_kit.model.code_detector.CodeClassification') + def test_singleton_functionality(self, mock_code_classification, mock_singleton_manager): + """Test singleton pattern functionality.""" + # Test new instance creation + mock_singleton_manager.has_name.return_value = False + mock_instance = MagicMock() + mock_code_classification.return_value = mock_instance + mock_singleton_manager.get_resource.return_value = mock_instance -def test_decide_code_by_str(): - with patch('llm_web_kit.model.code_detector.get_singleton_code_detect') as mock_get_singleton_code_detect, patch( - 'llm_web_kit.model.code_detector.decide_code_func') as mock_decide_code_func: - mock_get_singleton_code_detect.return_value = MagicMock() - mock_decide_code_func.return_value = 0.6 - assert decide_code_by_str('test text') == 0.6 + result = get_singleton_code_detect() + mock_singleton_manager.set_resource.assert_called_once_with('code_detect_v4_0409', mock_instance) + self.assertIsNotNone(result) + # Test existing instance + mock_singleton_manager.reset_mock() + mock_singleton_manager.has_name.return_value = True + mock_singleton_manager.get_resource.return_value = mock_instance -def test_update_code_by_str(): - with patch('llm_web_kit.model.code_detector.decide_code_by_str') as mock_decide_code_by_str: - mock_decide_code_by_str.return_value = 0.6 - assert update_code_by_str('test text') == {'code': 0.6} + result = get_singleton_code_detect() + mock_singleton_manager.set_resource.assert_not_called() + self.assertEqual(result, mock_instance) + @patch('llm_web_kit.model.code_detector.get_singleton_code_detect') + def test_update_code_prob_by_str(self, mock_get_singleton): + """Test update_code_prob_by_str function.""" + mock_classifier = MagicMock() + mock_classifier.predict.return_value = {'has_code_prob_0409': 0.75} + mock_get_singleton.return_value = mock_classifier -if __name__ == '__main__': - unittest.main() + result = update_code_prob_by_str('import pandas as pd') + self.assertEqual(result, {'has_code_prob_0409': 0.75}) + mock_classifier.predict.assert_called_once_with('import pandas as pd') From 66d2efeae81fba53d8246e1f562998d174d52745 Mon Sep 17 00:00:00 2001 From: yujia Date: Mon, 7 Jul 2025 17:50:09 +0800 Subject: [PATCH 2/3] feat: add jieba in requirement. --- requirements/runtime.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/requirements/runtime.txt b/requirements/runtime.txt index 10b17d5b..549c8de1 100644 --- a/requirements/runtime.txt +++ b/requirements/runtime.txt @@ -6,6 +6,7 @@ commentjson==0.9.0 fasttext-wheel==0.9.2 filelock==3.16.1 html-alg-lib==2.0.2 +jieba>=0.42 jieba-fast==0.53 jupyter==1.1.1 langdetect_zh==1.0.4 From 80546fad9960592f5d04972e5c390c5aa3b37b33 Mon Sep 17 00:00:00 2001 From: yujia Date: Mon, 7 Jul 2025 19:22:05 +0800 Subject: [PATCH 3/3] opt: replace jieba to jieba_fast --- llm_web_kit/model/code_detector.py | 4 ++-- requirements/runtime.txt | 1 - 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/llm_web_kit/model/code_detector.py b/llm_web_kit/model/code_detector.py index f0e40fff..a9543ebd 100644 --- a/llm_web_kit/model/code_detector.py +++ b/llm_web_kit/model/code_detector.py @@ -3,7 +3,7 @@ from typing import Dict import fasttext -import jieba +import jieba_fast as jieba from llm_web_kit.config.cfg_reader import load_config from llm_web_kit.libs.logger import mylogger as logger @@ -36,7 +36,7 @@ def auto_download(self, resource_name: str = None) -> str: model_download_path = code_cl_v4_config['download_path'] model_md5 = code_cl_v4_config.get('md5', '') - target_dir = os.path.join(CACHE_DIR, 'resource_name') + target_dir = os.path.join(CACHE_DIR, resource_name) os.makedirs(target_dir, exist_ok=True) target_path = os.path.join(target_dir, f'{resource_name}.bin') diff --git a/requirements/runtime.txt b/requirements/runtime.txt index 549c8de1..10b17d5b 100644 --- a/requirements/runtime.txt +++ b/requirements/runtime.txt @@ -6,7 +6,6 @@ commentjson==0.9.0 fasttext-wheel==0.9.2 filelock==3.16.1 html-alg-lib==2.0.2 -jieba>=0.42 jieba-fast==0.53 jupyter==1.1.1 langdetect_zh==1.0.4