组件

Finetune

lazyllm.components.finetune.AlpacaloraFinetune

Bases: LazyLLMFinetuneBase

Source code in lazyllm\components\finetune\alpacalora.py

class AlpacaloraFinetune(LazyLLMFinetuneBase):
    defatult_kw = ArgsDict({
        'data_path': None,
        'batch_size': 64,
        'micro_batch_size': 4,
        'num_epochs': 2,
        'learning_rate': 5.e-4,
        'cutoff_len': 1030,
        'filter_nums': 1024,
        'val_set_size': 200,
        'lora_r': 8,
        'lora_alpha': 32,
        'lora_dropout': 0.05,
        'lora_target_modules': '[wo,wqkv]',
        'modules_to_save': '[tok_embeddings,output]',
        'deepspeed': '',
        'prompt_template_name': 'alpaca',
        'train_on_inputs': True,
        'show_prompt': False,
        'nccl_port': 19081,
    })
    auto_map = {'micro_batch_size': 'micro_batch_size'}

    def __init__(self,
                 base_model,
                 target_path,
                 merge_path=None,
                 model_name='LLM',
                 cp_files='tokeniz*',
                 launcher=launchers.remote(ngpus=1),  # noqa B008
                 **kw
                 ):
        if not merge_path:
            save_path = os.path.join(lazyllm.config['train_target_root'], target_path)
            target_path, merge_path = os.path.join(save_path, 'lazyllm_lora'), os.path.join(save_path, 'lazyllm_merge')
            os.makedirs(target_path, exist_ok=True)
            os.makedirs(merge_path, exist_ok=True)
        super().__init__(
            base_model,
            target_path,
            launcher=launcher,
        )
        self.folder_path = os.path.dirname(os.path.abspath(__file__))
        deepspeed_config_path = os.path.join(self.folder_path, 'alpaca-lora', 'ds.json')
        self.kw = copy.deepcopy(self.defatult_kw)
        self.kw['deepspeed'] = deepspeed_config_path
        self.kw['nccl_port'] = random.randint(19000, 20500)
        self.kw.check_and_update(kw)
        self.merge_path = merge_path
        self.cp_files = cp_files
        self.model_name = model_name

    def cmd(self, trainset, valset=None) -> str:
        thirdparty.check_packages(['datasets', 'deepspeed', 'fire', 'numpy', 'peft', 'torch', 'transformers'])
        if not os.path.exists(trainset):
            defatult_path = os.path.join(lazyllm.config['data_path'], trainset)
            if os.path.exists(defatult_path):
                trainset = defatult_path
        if not self.kw['data_path']:
            self.kw['data_path'] = trainset

        run_file_path = os.path.join(self.folder_path, 'alpaca-lora', 'finetune.py')
        cmd = (f'python {run_file_path} '
               f'--base_model={self.base_model} '
               f'--output_dir={self.target_path} '
            )
        cmd += self.kw.parse_kwargs()
        cmd += f' 2>&1 | tee {os.path.join(self.target_path, self.model_name)}_$(date +"%Y-%m-%d_%H-%M-%S").log'

        if self.merge_path:
            run_file_path = os.path.join(self.folder_path, 'alpaca-lora', 'utils', 'merge_weights.py')

            cmd = [cmd,
                   f'python {run_file_path} '
                   f'--base={self.base_model} '
                   f'--adapter={self.target_path} '
                   f'--save_path={self.merge_path} ',
                   f' cp {os.path.join(self.base_model, self.cp_files)} {self.merge_path} '
                ]

        # cmd = 'realpath .'
        return cmd

cmd(trainset, valset=None)

Source code in lazyllm\components\finetune\alpacalora.py

def cmd(self, trainset, valset=None) -> str:
    thirdparty.check_packages(['datasets', 'deepspeed', 'fire', 'numpy', 'peft', 'torch', 'transformers'])
    if not os.path.exists(trainset):
        defatult_path = os.path.join(lazyllm.config['data_path'], trainset)
        if os.path.exists(defatult_path):
            trainset = defatult_path
    if not self.kw['data_path']:
        self.kw['data_path'] = trainset

    run_file_path = os.path.join(self.folder_path, 'alpaca-lora', 'finetune.py')
    cmd = (f'python {run_file_path} '
           f'--base_model={self.base_model} '
           f'--output_dir={self.target_path} '
        )
    cmd += self.kw.parse_kwargs()
    cmd += f' 2>&1 | tee {os.path.join(self.target_path, self.model_name)}_$(date +"%Y-%m-%d_%H-%M-%S").log'

    if self.merge_path:
        run_file_path = os.path.join(self.folder_path, 'alpaca-lora', 'utils', 'merge_weights.py')

        cmd = [cmd,
               f'python {run_file_path} '
               f'--base={self.base_model} '
               f'--adapter={self.target_path} '
               f'--save_path={self.merge_path} ',
               f' cp {os.path.join(self.base_model, self.cp_files)} {self.merge_path} '
            ]

    # cmd = 'realpath .'
    return cmd

lazyllm.components.finetune.CollieFinetune

Bases: LazyLLMFinetuneBase

Source code in lazyllm\components\finetune\collie.py

class CollieFinetune(LazyLLMFinetuneBase):
    defatult_kw = ArgsDict({
        'data_path': None,
        'batch_size': 64,
        'micro_batch_size': 4,
        'num_epochs': 3,
        'learning_rate': 5.e-4,
        'dp_size': 8,
        'pp_size': 1,
        'tp_size': 1,
        'lora_r': 8,
        'lora_alpha': 16,
        'lora_dropout': 0.05,
        'lora_target_modules': '[wo,wqkv]',
        'modules_to_save': '[tok_embeddings,output]',
        'prompt_template_name': 'alpaca',
    })
    auto_map = {
        'ddp': 'dp_size',
        'micro_batch_size': 'micro_batch_size',
        'tp': 'tp_size',
    }

    def __init__(self,
                 base_model,
                 target_path,
                 merge_path=None,
                 model_name='LLM',
                 cp_files='tokeniz*',
                 launcher=launchers.remote(ngpus=1),  # noqa B008
                 **kw
                 ):
        if not merge_path:
            save_path = os.path.join(lazyllm.config['train_target_root'], target_path)
            target_path, merge_path = os.path.join(save_path, 'lazyllm_lora'), os.path.join(save_path, 'lazyllm_merge')
            os.makedirs(target_path, exist_ok=True)
            os.makedirs(merge_path, exist_ok=True)
        super().__init__(
            base_model,
            target_path,
            launcher=launcher,
        )
        self.folder_path = os.path.dirname(os.path.abspath(__file__))
        self.kw = copy.deepcopy(self.defatult_kw)
        self.kw.check_and_update(kw)
        self.merge_path = merge_path
        self.cp_files = cp_files
        self.model_name = model_name

    def cmd(self, trainset, valset=None) -> str:
        thirdparty.check_packages(['numpy', 'peft', 'torch', 'transformers'])
        if not os.path.exists(trainset):
            defatult_path = os.path.join(lazyllm.config['data_path'], trainset)
            if os.path.exists(defatult_path):
                trainset = defatult_path
        if not self.kw['data_path']:
            self.kw['data_path'] = trainset

        run_file_path = os.path.join(self.folder_path, 'collie', 'finetune.py')
        cmd = (f'python {run_file_path} '
               f'--base_model={self.base_model} '
               f'--output_dir={self.target_path} '
            )
        cmd += self.kw.parse_kwargs()
        cmd += f' 2>&1 | tee {os.path.join(self.target_path, self.model_name)}_$(date +"%Y-%m-%d_%H-%M-%S").log'

        if self.merge_path:
            run_file_path = os.path.join(self.folder_path, 'alpaca-lora', 'utils', 'merge_weights.py')

            cmd = [cmd,
                   f'python {run_file_path} '
                   f'--base={self.base_model} '
                   f'--adapter={self.target_path} '
                   f'--save_path={self.merge_path} ',
                   f' cp {os.path.join(self.base_model, self.cp_files)} {self.merge_path} '
                ]

        return cmd

lazyllm.components.finetune.LlamafactoryFinetune

Bases: LazyLLMFinetuneBase

Source code in lazyllm\components\finetune\llamafactory.py

class LlamafactoryFinetune(LazyLLMFinetuneBase):
    auto_map = {
        'gradient_step': 'gradient_accumulation_steps',
        'micro_batch_size': 'per_device_train_batch_size',
    }

    def __init__(self,
                 base_model,
                 target_path,
                 merge_path=None,
                 config_path=None,
                 export_config_path=None,
                 lora_r=None,
                 modules_to_save=None,
                 lora_target_modules=None,
                 launcher=launchers.remote(ngpus=1, sync=True),  # noqa B008
                 **kw
                 ):
        if not os.path.exists(base_model):
            default_path = os.path.join(lazyllm.config['model_path'], base_model)
            if os.path.exists(default_path):
                base_model = default_path
        if not merge_path:
            normalized_target = os.path.normpath(target_path)
            if normalized_target.endswith('lazyllm_lora'):
                merge_path = normalized_target.replace('lazyllm_lora', 'lazyllm_merge')
            else:
                save_path = os.path.join(lazyllm.config['train_target_root'], target_path)
                target_path = os.path.join(save_path, 'lazyllm_lora')
                merge_path = os.path.join(save_path, 'lazyllm_merge')
            os.makedirs(target_path, exist_ok=True)
            os.makedirs(merge_path, exist_ok=True)
        super().__init__(
            base_model,
            target_path,
            launcher=launcher,
        )
        self.merge_path = merge_path
        self.temp_yaml_file = None
        self.temp_export_yaml_file = None
        self.config_path = config_path
        self.export_config_path = export_config_path
        self.config_folder_path = os.path.dirname(os.path.abspath(__file__))

        default_config_path = os.path.join(self.config_folder_path, 'llama_factory', 'sft.yaml')
        self.template_dict = ArgsDict(self._load_yaml(default_config_path))

        if self.config_path:
            self.template_dict.update(self._load_yaml(self.config_path))

        if lora_r:
            self.template_dict['lora_rank'] = lora_r
        if modules_to_save:
            self.template_dict['additional_target'] = modules_to_save.strip('[]')
        if lora_target_modules:
            self.template_dict['lora_target'] = lora_target_modules.strip('[]')
        self.template_dict['model_name_or_path'] = base_model
        self.template_dict['output_dir'] = target_path
        self.template_dict['template'] = self._get_template_name(base_model)

        # Filter kw to only include keys that exist in template_dict
        # This ensures check_and_update won't fail due to unexpected keys
        # Keys not in template_dict will be silently ignored (they may be used elsewhere)
        ignored_keys = set(kw.keys()) - set(self.template_dict.keys())
        if ignored_keys:
            lazyllm.LOG.info(f'Ignored parameters not in template_dict: {sorted(ignored_keys)}')
        filtered_kw = {k: v for k, v in kw.items() if k in self.template_dict}

        self.template_dict.check_and_update(filtered_kw)

        default_export_config_path = os.path.join(self.config_folder_path, 'llama_factory', 'lora_export.yaml')
        self.export_dict = ArgsDict(self._load_yaml(default_export_config_path))

        if self.export_config_path:
            self.export_dict.update(self._load_yaml(self.export_config_path))

        self.export_dict['model_name_or_path'] = base_model
        self.export_dict['adapter_name_or_path'] = target_path
        self.export_dict['export_dir'] = merge_path
        self.export_dict['template'] = self.template_dict['template']

        self.temp_folder = os.path.join(lazyllm.config['temp_dir'], 'llamafactory_config', str(uuid.uuid4())[:10])
        if not os.path.exists(self.temp_folder):
            os.makedirs(self.temp_folder)
        self.log_file_path = None

    def _get_template_name(self, base_model):
        base_name = os.path.basename(base_model).lower()
        key_value = match_longest_prefix(base_name)
        if key_value:
            return key_value
        else:
            raise RuntimeError(f'Cannot find prfix of base_model({base_model}) '
                               f'in DEFAULT_TEMPLATE of LLaMA_Factory: {llamafactory_mapping_dict}')

    def _load_yaml(self, config_path):
        with open(config_path, 'r') as file:
            config_dict = yaml.safe_load(file)
        return config_dict

    def _build_temp_yaml(self, updated_template_str, prefix='train_'):
        fd, temp_yaml_file = tempfile.mkstemp(prefix=prefix, suffix='.yaml', dir=self.temp_folder)
        with os.fdopen(fd, 'w') as temp_file:
            temp_file.write(updated_template_str)
        return temp_yaml_file

    def _build_temp_dataset_info(self, datapaths, stage=None):  # noqa C901
        """
        Build dataset_info.json based on training stage and dataset format.
        """
        if isinstance(datapaths, str):
            datapaths = [datapaths]
        elif isinstance(datapaths, list) and all(isinstance(item, str) for item in datapaths):
            pass
        else:
            raise TypeError(f'datapaths({datapaths}) should be str or list of str.')

        if stage is None:
            stage = self.template_dict.get('stage', 'sft').lower()

        supported_stages = ['sft', 'pt', 'dpo']
        if stage not in supported_stages:
            raise ValueError(
                f'Unsupported training stage: {stage}. '
                f'Only supported stages are: {", ".join(supported_stages)}'
            )

        temp_dataset_dict = dict()
        for datapath in datapaths:
            datapath = os.path.join(lazyllm.config['data_path'], datapath)
            assert os.path.isfile(datapath)
            file_name, _ = os.path.splitext(os.path.basename(datapath))
            temp_dataset_dict[file_name] = {'file_name': datapath}

            formatting = None
            first_item = None

            if stage == 'pt':
                formatting = None
                try:
                    with open(datapath, 'r', encoding='utf-8') as file:
                        first_bytes = file.read(1024)
                        file.seek(0)

                        if not first_bytes.strip().startswith(('[', '{')):
                            lines = file.readlines()
                            if not lines:
                                raise ValueError(f'PT stage: Dataset file {datapath} is empty')

                            first_item = {'text': lines[0].strip() if lines else ''}
                            lazyllm.LOG.info(
                                f'PT stage: Dataset {file_name} detected as plain text format '
                                f'({len(lines)} lines). LLaMA-Factory will handle conversion.'
                            )
                        else:
                            try:
                                data = json.load(file)
                                if isinstance(data, list):
                                    if not data:
                                        raise ValueError(f'PT stage: Dataset file {datapath} is empty (empty list)')
                                    first_item = data[0]
                                elif isinstance(data, dict):
                                    first_item = data
                                else:
                                    raise ValueError(
                                        f'PT stage: Dataset file {datapath} has invalid JSON structure. '
                                        f'Expected list or dict, got {type(data).__name__}'
                                    )
                                lazyllm.LOG.info(
                                    f'PT stage: Dataset {file_name} detected as JSON format. '
                                    f'Looking for "text" field.'
                                )
                            except json.JSONDecodeError as json_err:
                                raise ValueError(
                                    f'PT stage: Dataset file {datapath} is neither valid plain text nor valid JSON. '
                                    f'JSON parse error: {str(json_err)}'
                                )

                    if not first_item:
                        raise ValueError(f'PT stage: Failed to extract first item from dataset {datapath}')

                    self._build_alpaca_dataset_info(
                        temp_dataset_dict, file_name, first_item, stage
                    )

                except Exception as e:
                    error_msg = (
                        f'PT stage: Failed to process dataset {file_name} from {datapath}. '
                        f'Error: {str(e)}. '
                        f'PT mode requires either: '
                        f'(1) Plain text format (one text per line), or '
                        f'(2) JSON format with "text" field in each object.'
                    )
                    lazyllm.LOG.error(error_msg)
                    raise ValueError(error_msg) from e
            else:
                formatting = 'alpaca'
                try:
                    with open(datapath, 'r', encoding='utf-8') as file:
                        data = json.load(file)
                    if not data:
                        raise ValueError(f'Dataset file {datapath} is empty')

                    first_item = data[0]

                    if 'messages' in first_item:
                        formatting = 'sharegpt'
                        self._build_sharegpt_dataset_info(
                            temp_dataset_dict, file_name, first_item, stage
                        )
                    else:
                        self._build_alpaca_dataset_info(
                            temp_dataset_dict, file_name, first_item, stage
                        )

                except Exception as e:
                    lazyllm.LOG.warning(
                        f'Failed to analyze dataset {datapath} for stage {stage}: {e}. '
                        f'Using default formatting.'
                    )

            if formatting is not None:
                temp_dataset_dict[file_name].update({'formatting': formatting})

        self.temp_dataset_info_path = os.path.join(self.temp_folder, 'dataset_info.json')
        with open(self.temp_dataset_info_path, 'w') as json_file:
            json.dump(temp_dataset_dict, json_file, indent=4)
        return self.temp_dataset_info_path, ','.join(temp_dataset_dict.keys())

    def _build_alpaca_dataset_info(self, dataset_dict, file_name, first_item, stage):  # noqa C901
        """
        Build dataset info for Alpaca format based on training stage.
        """
        columns = {}
        ranking = False

        media_types = []
        for media in ['images', 'videos', 'audios']:
            if media in first_item:
                media_types.append(media)

        if stage == 'pt':
            if 'text' in first_item:
                columns['prompt'] = 'text'
            else:
                if 'instruction' in first_item:
                    columns['prompt'] = 'instruction'
                elif 'output' in first_item:
                    columns['prompt'] = 'output'
                else:
                    lazyllm.LOG.warning(
                        f'PT stage: No "text" field found in dataset {file_name}, '
                        f'using "instruction" or "output" as fallback'
                    )
                    columns['prompt'] = 'instruction' if 'instruction' in first_item else 'output'

        elif stage == 'dpo':
            ranking = True
            if 'chosen' in first_item and 'rejected' in first_item:
                columns['prompt'] = 'instruction' if 'instruction' in first_item else None
                columns['query'] = 'input' if 'input' in first_item else None
                columns['chosen'] = 'chosen'
                columns['rejected'] = 'rejected'
                columns = {k: v for k, v in columns.items() if v is not None}
            else:
                raise ValueError(
                    f'DPO stage requires "chosen" and "rejected" fields in dataset, '
                    f'but found: {list(first_item.keys())}'
                )

        elif stage == 'sft':
            if 'instruction' in first_item:
                columns['prompt'] = 'instruction'
            if 'input' in first_item:
                columns['query'] = 'input'
            if 'output' in first_item:
                columns['response'] = 'output'
            if 'system' in first_item:
                columns['system'] = 'system'
            if 'history' in first_item:
                columns['history'] = 'history'
        else:
            raise ValueError(f'Unsupported stage: {stage}. Only sft, pt, dpo are supported.')

        if media_types:
            multimodal_columns = {item: item for item in media_types}
            multimodal_columns.update(columns)
            columns = multimodal_columns

        update_dict = {'columns': columns}
        if ranking:
            update_dict['ranking'] = True

        dataset_dict[file_name].update(update_dict)

    def _build_sharegpt_dataset_info(self, dataset_dict, file_name, first_item, stage):  # noqa C901
        """
        Build dataset info for ShareGPT format based on training stage.
        """
        columns = {}
        ranking = False

        media_types = []
        for media in ['images', 'videos', 'audios']:
            if media in first_item:
                media_types.append(media)

        if stage == 'dpo':
            ranking = True
            if 'chosen' in first_item and 'rejected' in first_item:
                columns['messages'] = 'conversations' if 'conversations' in first_item else 'messages'
                columns['chosen'] = 'chosen'
                columns['rejected'] = 'rejected'
            else:
                raise ValueError(
                    f'DPO stage requires "chosen" and "rejected" fields in dataset, '
                    f'but found: {list(first_item.keys())}'
                )
        elif stage == 'sft':
            columns['messages'] = 'conversations' if 'conversations' in first_item else 'messages'
            if 'system' in first_item:
                columns['system'] = 'system'
            if 'tools' in first_item:
                columns['tools'] = 'tools'

            if 'messages' in first_item and isinstance(first_item['messages'], list):
                if len(first_item['messages']) > 0:
                    msg = first_item['messages'][0]
                    if 'role' in msg and 'content' in msg:
                        dataset_dict[file_name].update({
                            'tags': {
                                'role_tag': 'role',
                                'content_tag': 'content',
                                'user_tag': 'user',
                                'assistant_tag': 'assistant',
                                'system_tag': 'system'
                            }
                        })
        else:
            raise ValueError(f'Unsupported stage: {stage}. Only sft, pt, dpo are supported.')

        if media_types:
            multimodal_columns = {item: item for item in media_types}
            multimodal_columns.update(columns)
            columns = multimodal_columns

        update_dict = {'columns': columns}
        if ranking:
            update_dict['ranking'] = True

        dataset_dict[file_name].update(update_dict)

    def _rm_temp_yaml(self):
        if self.temp_yaml_file:
            if os.path.exists(self.temp_yaml_file):
                os.remove(self.temp_yaml_file)
            self.temp_yaml_file = None

    def cmd(self, trainset, valset=None) -> str:
        thirdparty.check_packages(['datasets', 'deepspeed', 'numpy', 'peft', 'torch', 'transformers', 'trl'])
        if 'dataset_dir' in self.template_dict and self.template_dict['dataset_dir'] == 'lazyllm_temp_dir':
            stage = self.template_dict.get('stage', 'sft')
            _, datasets = self._build_temp_dataset_info(trainset, stage=stage)
            self.template_dict['dataset_dir'] = self.temp_folder
        else:
            datasets = trainset
        self.template_dict['dataset'] = datasets

        if self.template_dict['finetuning_type'] == 'lora':
            # For LoRA/QLoRA: use llamafactory-cli export to merge adapter with base model
            # For Full finetuning: model copy handling is done in cmds below
            updated_export_str = yaml.dump(dict(self.export_dict), default_flow_style=False)
            self.temp_export_yaml_file = self._build_temp_yaml(updated_export_str, prefix='merge_')

        updated_template_str = yaml.dump(dict(self.template_dict), default_flow_style=False)
        self.temp_yaml_file = self._build_temp_yaml(updated_template_str)

        formatted_date = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
        random_value = random.randint(1000, 9999)
        self.log_file_path = f'{self.target_path}/train_log_{formatted_date}_{random_value}.log'

        # Use bash instead of sh to support pipefail
        # This ensures export only runs if training succeeds
        cmds = (f'export DISABLE_VERSION_CHECK=1 && bash -c "set -o pipefail && '
                f'llamafactory-cli train {self.temp_yaml_file} 2>&1 | '
                f'tee {self.log_file_path}"')
        if self.temp_export_yaml_file:
            # For LoRA/QLoRA: merge adapter with base model
            # Only run export if training succeeds (exit code 0)
            # With pipefail, tee will preserve the exit code from llamafactory-cli
            cmds += f' && llamafactory-cli export {self.temp_export_yaml_file}'
        elif self.template_dict['finetuning_type'] == 'full':
            # For Full finetuning: copy model from lazyllm_lora to lazyllm_merge
            # This maintains consistency with LoRA/QLoRA workflow
            # Only copy if training succeeds (exit code 0)
            # Only copy model files, exclude training process information (checkpoints, logs, etc.)
            # to save storage space since lazyllm_merge is only used for exporting to models directory
            exclude_patterns = [
                '--exclude=checkpoint-*',  # Exclude checkpoint directories
                '--exclude=train_log_*.log',  # Exclude training logs
                '--exclude=trainer_state.json',  # Exclude trainer state
                '--exclude=trainer_log.jsonl',  # Exclude trainer log
                '--exclude=train_results.json',  # Exclude training results
                '--exclude=all_results.json',  # Exclude all results
                '--exclude=eval_results.json',  # Exclude evaluation results
                '--exclude=training_loss.png',  # Exclude training loss plot
                '--exclude=runs/',  # Exclude tensorboard logs
                '--exclude=training_args.bin',  # Exclude training arguments
            ]
            exclude_str = ' '.join(exclude_patterns)
            cmds += f' && rsync -a {exclude_str} {self.target_path}/ {self.merge_path}/ 2>/dev/null || true'
        return cmds

lazyllm.components.deploy.LazyLLMDeployBase

Bases: ComponentBase

Source code in lazyllm\components\deploy\base.py

class LazyLLMDeployBase(ComponentBase):
    keys_name_handle = None
    message_format = None
    default_headers = {'Content-Type': 'application/json'}
    stream_url_suffix = ''
    stream_parse_parameters = {}

    encoder_map = dict(image=_image_to_base64, audio=_audio_to_base64, ocr_files=ocr_to_base64)

    @staticmethod
    def extract_result(output, inputs):
        return output

    def __init__(self, *, launcher=launchers.remote()):  # noqa B008
        super().__init__(launcher=launcher)

lazyllm.components.deploy.LazyLLMDeployBase.extract_result(output, inputs) staticmethod

Source code in lazyllm\components\deploy\base.py

@staticmethod
def extract_result(output, inputs):
    return output

lazyllm.components.finetune.FlagembeddingFinetune

Bases: LazyLLMFinetuneBase

Source code in lazyllm\components\finetune\flagembedding.py

class FlagembeddingFinetune(LazyLLMFinetuneBase):
    defatult_embed_kw = ArgsDict({
        'train_group_size': 8,
        'query_max_len': 512,
        'passage_max_len': 512,
        'pad_to_multiple_of': 8,
        'query_instruction_for_retrieval': 'Represent this sentence for searching relevant passages: ',
        'query_instruction_format': '{}{}',
        'learning_rate': 1e-5,
        'num_train_epochs': 1,
        'per_device_train_batch_size': 2,
        'gradient_accumulation_steps': 1,
        'dataloader_drop_last': True,
        'warmup_ratio': 0.1,
        'weight_decay': 0.01,
        'deepspeed': '',
        'logging_steps': 1,
        'save_steps': 1000,
        'temperature': 0.02,
        'sentence_pooling_method': 'cls',
        'normalize_embeddings': True,
        'kd_loss_type': 'kl_div',
        'overwrite_output_dir': True,
        'fp16': True,
        'gradient_checkpointing': True,
        'negatives_cross_device': True
    })
    defatult_rerank_kw = ArgsDict({
        'train_group_size': 8,
        'query_max_len': 256,
        'passage_max_len': 256,
        'pad_to_multiple_of': 8,
        'learning_rate': 6e-5,
        'num_train_epochs': 1,
        'per_device_train_batch_size': 2,
        'gradient_accumulation_steps': 1,
        'dataloader_drop_last': True,
        'warmup_ratio': 0.1,
        'weight_decay': 0.01,
        'deepspeed': '',
        'logging_steps': 1,
        'save_steps': 1000,
        'overwrite_output_dir': True,
        'fp16': True,
        'gradient_checkpointing': True
    })
    store_true_embed_kw = {'overwrite_output_dir', 'fp16', 'gradient_checkpointing', 'negatives_cross_device'}
    store_true_rerank_kw = {'overwrite_output_dir', 'fp16', 'gradient_checkpointing'}

    def __init__(
        self,
        base_model,
        target_path,
        launcher=launchers.remote(ngpus=1, sync=True),  # noqa B008
        **kw
    ):
        model_type = ModelManager.get_model_type(base_model.split('/')[-1])
        if model_type not in ('embed', 'rerank'):
            raise RuntimeError(f'Not supported {model_type} type to finetune.')
        if not os.path.exists(base_model):
            defatult_path = os.path.join(lazyllm.config['model_path'], base_model)
            if os.path.exists(defatult_path):
                base_model = defatult_path
        save_path = os.path.join(lazyllm.config['train_target_root'], target_path)
        target_path = os.path.join(save_path, model_type)
        os.system(f'mkdir -p {target_path}')
        super().__init__(
            base_model,
            target_path,
            launcher=launcher,
        )
        if model_type == 'rerank':
            self.kw = copy.deepcopy(self.defatult_rerank_kw)
            self.store_true_kw = copy.deepcopy(self.store_true_rerank_kw)
            self.module_run_path = 'FlagEmbedding.finetune.reranker.encoder_only.base'
        else:
            self.kw = copy.deepcopy(self.defatult_embed_kw)
            self.store_true_kw = copy.deepcopy(self.store_true_embed_kw)
            self.module_run_path = 'FlagEmbedding.finetune.embedder.encoder_only.base'
        self.kw.check_and_update(kw)
        if not self.kw['deepspeed']:
            folder_path = os.path.dirname(os.path.abspath(__file__))
            deepspeed_config_path = os.path.join(folder_path, 'flag_embedding', 'ds_stage0.json')
            self.kw['deepspeed'] = deepspeed_config_path
        self.nproc_per_node = launcher.ngpus

    def cmd(self, trainset, valset=None) -> str:
        thirdparty.check_packages(['flagembedding'])
        self.kw['train_data'] = trainset

        formatted_date = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
        self.log_file_path = f'{self.target_path}/train_log_{formatted_date}_{random.randint(1000, 9999)}.log'
        cache_path = os.path.join(os.path.expanduser(lazyllm.config['home']), 'fintune', 'embeding')
        cache_model_path = os.path.join(cache_path, 'model')
        cache_data_path = os.path.join(cache_path, 'data')
        os.system(f'mkdir -p {cache_model_path} {cache_data_path}')

        cmds = (f'export WANDB_MODE=disabled && torchrun --nproc_per_node {self.nproc_per_node} '
                f'-m {self.module_run_path} '
                f'--model_name_or_path {self.base_model} '
                f'--output_dir {self.target_path} '
                f'--cache_dir {cache_model_path} '
                f'--cache_path {cache_data_path} '
            )
        for key in self.store_true_kw:
            cmds += f'--{key} ' if self.kw.pop(key) else ''
        cmds += self.kw.parse_kwargs()
        cmds += f' 2>&1 | tee {self.log_file_path}'
        return cmds

lazyllm.components.auto.AutoFinetune

Bases: LazyLLMFinetuneBase

Source code in lazyllm\components\auto\autofinetune.py

class AutoFinetune(LazyLLMFinetuneBase):

    def __new__(cls, base_model: str, target_path: str, source: Optional[str] = None,
                merge_path: Optional[str] = None, batch_size: int = 32, lora_r: int = 8,
                model_type: Optional[str] = None, launcher: Optional[LazyLLMLaunchersBase] = None, **kw):
        base_model = ModelManager(source).download(base_model) or ''
        LOG.info(f'[AutoFinetune] Using base model from: {base_model}')
        model_name = get_model_name(base_model)
        if not model_type:
            model_type = ModelManager.get_model_type(model_name)
            LOG.info(f'[AutoFinetune] Infer type of model {model_name} is {model_type}')
        if model_type in ['tts', 'stt', 'sd', 'ocr', 'cross_modal_embed']:
            raise RuntimeError(f'Fine-tuning of the {model_type} model is not currently supported.')

        if model_type in ['embed', 'rerank']:
            LOG.info(f'[AutoFinetune] Finetune {model_name} with FlagEmbedding.')
            return finetune.flagembedding(base_model, target_path, **kw)

        params = {'gradient_step': 1, 'micro_batch_size': 32}
        if not launcher:
            match = re.search(r'(\d+)[bB]', model_name)
            model_size = int(match.group(1)) if match else 0
            gs, mbs, ngpus = estimate_finetune_plan(
                gpu_mem_gb=config['gpu_memory'], model_size_b=model_size, batch_size=batch_size, lora_r=lora_r)
            params.update({'gradient_step': gs, 'micro_batch_size': mbs})
            LOG.info(f'[AutoFinetune] Infer model_size: {model_size} B, '
                     f'gradient_step: {gs}, micro_batch_size: {mbs}, ngpus: {ngpus}')
            launcher = launchers.remote(ngpus=ngpus, sync=True)

        candidates = ['llamafactory', 'alpacalora']
        candidates = dict(llm=['llamafactory', 'alpacalora'], vlm=['llamafactory'])
        for finetune_cls_name in candidates[model_type]:
            if check_requirements(requirements[finetune_cls_name]):
                finetune_cls = getattr(finetune, finetune_cls_name)
                for key, value in finetune_cls.auto_map.items():
                    if value and value not in kw:
                        kw[value] = params[key]
                LOG.info(f'[AutoFinetune] Use {finetune_cls_name} to finetune.')
                if finetune_cls_name == 'llamafactory':
                    return finetune_cls(base_model, target_path, lora_r=lora_r, launcher=launcher, **kw)
                return finetune_cls(base_model, target_path, merge_path, cp_files='tokeniz*',
                                    batch_size=batch_size, lora_r=lora_r, launcher=launcher, **kw)
        raise RuntimeError('No valid framework found, candidates are '
                           f'{[c.framework.lower() for c in candidates[model_type]]}.')

lazyllm.components.finetune.base.DummyFinetune

Bases: LazyLLMFinetuneBase

Source code in lazyllm\components\finetune\base.py

class DummyFinetune(LazyLLMFinetuneBase):
    def __init__(self, base_model='base', target_path='target', *, launcher=launchers.remote(), **kw):  # noqa B008
        super().__init__(base_model, target_path, launcher=launchers.empty)
        self.kw = kw

    def cmd(self, *args, **kw) -> str:
        return f'echo \'dummy finetune!, and init-args is {self.kw}\''

lazyllm.components.finetune.LazyLLMFinetuneBase

Bases: ComponentBase

Source code in lazyllm\components\finetune\base.py

class LazyLLMFinetuneBase(ComponentBase):
    __reg_overwrite__ = 'cmd'

    def __init__(self, base_model, target_path, *, launcher=launchers.remote()):  # noqa B008
        super().__init__(launcher=launcher)
        self.base_model = base_model
        self.target_path = target_path
        self.merge_path = None

    def __call__(self, *args, **kw):
        super().__call__(*args, **kw)
        if self.merge_path:
            return self.merge_path
        else:
            return self.target_path

---

Deploy

lazyllm.components.deploy.Lightllm

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\lightllm.py

class Lightllm(LazyLLMDeployBase):
    keys_name_handle = {
        'inputs': 'inputs',
        'stop': 'stop_sequences'
    }
    default_headers = {'Content-Type': 'application/json'}
    message_format = {
        'inputs': 'Who are you ?',
        'parameters': {
            'do_sample': False,
            'presence_penalty': 0.0,
            'frequency_penalty': 0.0,
            'repetition_penalty': 1.0,
            'temperature': 1.0,
            'top_p': 1,
            'top_k': -1,  # -1 is for all
            'ignore_eos': False,
            'max_new_tokens': 8192,
            'stop_sequences': None,
        }
    }
    auto_map = {}
    stream_url_suffix = '_stream'
    stream_parse_parameters = {'delimiter': b'\n\n'}

    def __init__(self, trust_remote_code=True, launcher=launchers.remote(ngpus=1), log_path=None, # noqa B008
                 openai_api: Optional[bool] = None, **kw):
        super().__init__(launcher=launcher)
        self.kw = ArgsDict({
            'tp': 1,
            'max_total_token_num': 64000,
            'eos_id': 2,
            'port': None,
            'host': '0.0.0.0',
            'nccl_port': None,
            'tokenizer_mode': 'auto',
            'running_max_req_size': 256,
            'data_type': 'float16',
            'max_req_total_len': 64000,
            'max_req_input_len': 4096,
            'long_truncation_mode': 'head',
        })
        self.options_keys = kw.pop('options_keys', [])
        if trust_remote_code and 'trust_remote_code' not in self.options_keys:
            self.options_keys.append('trust_remote_code')
        self.kw.check_and_update(kw)
        self.random_port = False if 'port' in kw and kw['port'] else True
        self.random_nccl_port = False if 'nccl_port' in kw and kw['nccl_port'] else True
        self.temp_folder = make_log_dir(log_path, 'lightllm') if log_path else None

    def cmd(self, finetuned_model=None, base_model=None):
        if not os.path.exists(finetuned_model) or \
            not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                    for filename in os.listdir(finetuned_model)):
            if not finetuned_model:
                LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                            f'base_model({base_model}) will be used')
            finetuned_model = base_model

        def impl():
            if self.random_port:
                self.kw['port'] = random.randint(30000, 40000)
            if self.random_nccl_port:
                self.kw['nccl_port'] = random.randint(20000, 30000)
            cmd = f'python -m lightllm.server.api_server --model_dir {finetuned_model} '
            cmd += self.kw.parse_kwargs()
            cmd += ' ' + parse_options_keys(self.options_keys)
            if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
            return cmd

        return LazyLLMCMD(cmd=impl, return_value=self.geturl, checkf=verify_fastapi_func)

    def geturl(self, job=None):
        if job is None:
            job = self.job
        if lazyllm.config['mode'] == lazyllm.Mode.Display:
            return 'http://{ip}:{port}/generate'
        else:
            return f'http://{job.get_jobip()}:{self.kw["port"]}/generate'

    @staticmethod
    def extract_result(x, inputs):
        try:
            if x.startswith('data:'): return json.loads(x[len('data:'):])['token']['text']
            else: return json.loads(x)['generated_text'][0]
        except Exception as e:
            LOG.warning(f'JSONDecodeError on load {x}')
            raise e

cmd(finetuned_model=None, base_model=None)

Source code in lazyllm\components\deploy\lightllm.py

def cmd(self, finetuned_model=None, base_model=None):
    if not os.path.exists(finetuned_model) or \
        not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                for filename in os.listdir(finetuned_model)):
        if not finetuned_model:
            LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                        f'base_model({base_model}) will be used')
        finetuned_model = base_model

    def impl():
        if self.random_port:
            self.kw['port'] = random.randint(30000, 40000)
        if self.random_nccl_port:
            self.kw['nccl_port'] = random.randint(20000, 30000)
        cmd = f'python -m lightllm.server.api_server --model_dir {finetuned_model} '
        cmd += self.kw.parse_kwargs()
        cmd += ' ' + parse_options_keys(self.options_keys)
        if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
        return cmd

    return LazyLLMCMD(cmd=impl, return_value=self.geturl, checkf=verify_fastapi_func)

geturl(job=None)

Source code in lazyllm\components\deploy\lightllm.py

def geturl(self, job=None):
    if job is None:
        job = self.job
    if lazyllm.config['mode'] == lazyllm.Mode.Display:
        return 'http://{ip}:{port}/generate'
    else:
        return f'http://{job.get_jobip()}:{self.kw["port"]}/generate'

extract_result(x, inputs) staticmethod

Source code in lazyllm\components\deploy\lightllm.py

@staticmethod
def extract_result(x, inputs):
    try:
        if x.startswith('data:'): return json.loads(x[len('data:'):])['token']['text']
        else: return json.loads(x)['generated_text'][0]
    except Exception as e:
        LOG.warning(f'JSONDecodeError on load {x}')
        raise e

lazyllm.components.deploy.Vllm

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\vllm.py

class Vllm(LazyLLMDeployBase, metaclass=_VllmStreamParseParametersMeta):
    # keys_name_handle/default_headers/message_format will lose efficacy when openai_api is True
    keys_name_handle = {'inputs': 'prompt', 'stop': 'stop'}
    default_headers = {'Content-Type': 'application/json'}
    message_format = {
        'prompt': 'Who are you ?',
        'stream': False,
        'stop': ['<|im_end|>', '<|im_start|>', '</s>', '<|assistant|>', '<|user|>', '<|system|>', '<eos>'],
        'skip_special_tokens': False,
        'temperature': 0.6,
        'top_p': 0.8,
        'max_tokens': 4096
    }
    auto_map = {
        'tp': 'tensor_parallel_size'
    }  # from cli to vllm
    optional_keys = set([
        'max_model_len',
        'gpu_memory_utilization',
        'task',
        'dtype',
        'kv_cache_dtype',
        'tokenizer_mode',
        'block_size',
        'max_num_seqs',
        'pipeline_parallel_size',
        'tensor_parallel_size',
        'seed',
        'port',
        'max_num_batched_tokens',
        'tool_call_parser',
        'swap_space',
        'mm_processor_kwargs',
        'limit_mm_per_prompt',
        'hf_overrides'])

    # TODO(wangzhihong): change default value for `openai_api` argument to True
    def __init__(self, trust_remote_code: bool = True, launcher: LazyLLMLaunchersBase = launchers.remote(ngpus=1),  # noqa B008
                 log_path: str = None, openai_api: Optional[bool] = None, **kw):
        self.launcher_list, launcher = reallocate_launcher(launcher)
        super().__init__(launcher=launcher)
        self.kw = ArgsDict({
            'host': '0.0.0.0',
            'max_model_len': 10240,
        })
        if openai_api is None: openai_api = lazyllm.config['openai_api']
        self._vllm_cmd = 'vllm.entrypoints.openai.api_server' if openai_api else 'vllm.entrypoints.api_server'
        self._openai_api = openai_api
        self.options_keys = kw.pop('options_keys', [])
        if trust_remote_code and 'trust_remote_code' not in self.options_keys:
            self.options_keys.append('trust_remote_code')
        self.kw.update(**{key: kw[key] for key in self.optional_keys if key in kw})
        self.kw.check_and_update(kw)
        self.random_port = False if 'port' in kw and kw['port'] and kw['port'] != 'auto' else True
        self.temp_folder = make_log_dir(log_path, 'vllm') if log_path else None
        if self.launcher_list:
            ray_launcher = [Distributed(launcher=launcher) for launcher in self.launcher_list]
            parall_launcher = [lazyllm.pipeline(sleep_moment, launcher) for launcher in ray_launcher[1:]]
            self._prepare_deploy = lazyllm.pipeline(
                ray_launcher[0], post_action=(lazyllm.parallel(*parall_launcher) if len(parall_launcher) else None))

    def cmd(self, finetuned_model=None, base_model=None, master_ip=None):
        if finetuned_model:
            LOG.info(f'Using finetuned model from {finetuned_model} to deploy.')
        if not finetuned_model:
            LOG.info(f'Using model {base_model} to deploy.')
            finetuned_model = base_model
        elif not os.path.exists(finetuned_model):
            LOG.warning(f'Warning! The finetuned_model path does not exist: {finetuned_model}. '
                        f'Using base_model({base_model}) instead.')
            finetuned_model = base_model
        elif not any(filename.endswith(('.bin', '.safetensors', '.pt'))
                     for filename in os.listdir(finetuned_model)):
            LOG.warning(f'Warning! No valid model files (.bin, .safetensors or .pt) found in: {finetuned_model}. '
                        f'Using base_model({base_model}) instead.')
            finetuned_model = base_model

        def impl():
            if self.random_port:
                self.kw['port'] = random.randint(30000, 40000)

            cmd = ''
            if self.launcher_list:
                cmd += f'ray start --address="{master_ip}" && '
            cmd += f'{sys.executable} -m {self._vllm_cmd} --model {finetuned_model} '
            if self._openai_api: cmd += '--served-model-name lazyllm '
            cmd += self.kw.parse_kwargs()
            cmd += ' ' + parse_options_keys(self.options_keys)
            if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
            return cmd

        return LazyLLMCMD(cmd=impl, return_value=self.geturl,
                          checkf=(verify_vllm_openai_func if self._openai_api else verify_fastapi_func))

    def geturl(self, job=None):
        if job is None:
            job = self.job
        if lazyllm.config['mode'] == lazyllm.Mode.Display:
            return 'http://{ip}:{port}/generate'
        else:
            return f'http://{job.get_jobip()}:{self.kw["port"]}' + (
                '/v1/' if self._openai_api else '/generate')

    @staticmethod
    def extract_result(x, inputs):
        return json.loads(x)['text'][0]

cmd(finetuned_model=None, base_model=None, master_ip=None)

Source code in lazyllm\components\deploy\vllm.py

def cmd(self, finetuned_model=None, base_model=None, master_ip=None):
    if finetuned_model:
        LOG.info(f'Using finetuned model from {finetuned_model} to deploy.')
    if not finetuned_model:
        LOG.info(f'Using model {base_model} to deploy.')
        finetuned_model = base_model
    elif not os.path.exists(finetuned_model):
        LOG.warning(f'Warning! The finetuned_model path does not exist: {finetuned_model}. '
                    f'Using base_model({base_model}) instead.')
        finetuned_model = base_model
    elif not any(filename.endswith(('.bin', '.safetensors', '.pt'))
                 for filename in os.listdir(finetuned_model)):
        LOG.warning(f'Warning! No valid model files (.bin, .safetensors or .pt) found in: {finetuned_model}. '
                    f'Using base_model({base_model}) instead.')
        finetuned_model = base_model

    def impl():
        if self.random_port:
            self.kw['port'] = random.randint(30000, 40000)

        cmd = ''
        if self.launcher_list:
            cmd += f'ray start --address="{master_ip}" && '
        cmd += f'{sys.executable} -m {self._vllm_cmd} --model {finetuned_model} '
        if self._openai_api: cmd += '--served-model-name lazyllm '
        cmd += self.kw.parse_kwargs()
        cmd += ' ' + parse_options_keys(self.options_keys)
        if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
        return cmd

    return LazyLLMCMD(cmd=impl, return_value=self.geturl,
                      checkf=(verify_vllm_openai_func if self._openai_api else verify_fastapi_func))

geturl(job=None)

Source code in lazyllm\components\deploy\vllm.py

def geturl(self, job=None):
    if job is None:
        job = self.job
    if lazyllm.config['mode'] == lazyllm.Mode.Display:
        return 'http://{ip}:{port}/generate'
    else:
        return f'http://{job.get_jobip()}:{self.kw["port"]}' + (
            '/v1/' if self._openai_api else '/generate')

extract_result(x, inputs) staticmethod

Source code in lazyllm\components\deploy\vllm.py

@staticmethod
def extract_result(x, inputs):
    return json.loads(x)['text'][0]

lazyllm.components.deploy.LMDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\lmdeploy.py

class LMDeploy(LazyLLMDeployBase):
    keys_name_handle = {
        'inputs': 'prompt',
        'stop': 'stop',
    }
    default_headers = {'Content-Type': 'application/json'}
    message_format = {
        'prompt': 'Who are you ?',
        'stream': False,
        'stop': None,
        'top_p': 0.8,
        'temperature': 0.8,
        'skip_special_tokens': True,
    }
    auto_map = {
        'port': 'server-port',
        'host': 'server-name',
        'max_batch_size': 'max-batch-size',
    }
    stream_parse_parameters = {'delimiter': b'\n'}

    def __init__(self, launcher=launchers.remote(ngpus=1), trust_remote_code=True, log_path=None, **kw):  # noqa B008
        super().__init__(launcher=launcher)
        self.kw = ArgsDict({
            'server-name': '0.0.0.0',
            'server-port': None,
            'tp': 1,
            'max-batch-size': 128,
        })
        self.options_keys = kw.pop('options_keys', [])
        self.kw.check_and_update(kw)
        self._trust_remote_code = trust_remote_code
        self.random_port = False if 'server-port' in kw and kw['server-port'] else True
        self.temp_folder = make_log_dir(log_path, 'lmdeploy') if log_path else None

    def cmd(self, finetuned_model=None, base_model=None):
        if not os.path.exists(finetuned_model) or \
            not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                    for filename in os.listdir(finetuned_model)):
            if not finetuned_model:
                LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                            f'base_model({base_model}) will be used')
            finetuned_model = base_model

        def impl():
            if self.random_port:
                self.kw['server-port'] = random.randint(30000, 40000)
            cmd = f'lmdeploy serve api_server {finetuned_model} --model-name lazyllm '

            if importlib.util.find_spec('torch_npu') is not None: cmd += '--device ascend '
            if config['lmdeploy_eager_mode']: cmd += '--eager-mode '
            cmd += self.kw.parse_kwargs()
            cmd += ' ' + parse_options_keys(self.options_keys)
            if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
            return cmd

        return LazyLLMCMD(cmd=impl, return_value=self.geturl, checkf=verify_fastapi_func)

    def geturl(self, job=None):
        if job is None:
            job = self.job
        if lazyllm.config['mode'] == lazyllm.Mode.Display:
            return 'http://{ip}:{port}/v1/'
        else:
            return f'http://{job.get_jobip()}:{self.kw["server-port"]}/v1/'

    @staticmethod
    def extract_result(x, inputs):
        return json.loads(x)['text']

cmd(finetuned_model=None, base_model=None)

Source code in lazyllm\components\deploy\lmdeploy.py

def cmd(self, finetuned_model=None, base_model=None):
    if not os.path.exists(finetuned_model) or \
        not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                for filename in os.listdir(finetuned_model)):
        if not finetuned_model:
            LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                        f'base_model({base_model}) will be used')
        finetuned_model = base_model

    def impl():
        if self.random_port:
            self.kw['server-port'] = random.randint(30000, 40000)
        cmd = f'lmdeploy serve api_server {finetuned_model} --model-name lazyllm '

        if importlib.util.find_spec('torch_npu') is not None: cmd += '--device ascend '
        if config['lmdeploy_eager_mode']: cmd += '--eager-mode '
        cmd += self.kw.parse_kwargs()
        cmd += ' ' + parse_options_keys(self.options_keys)
        if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
        return cmd

    return LazyLLMCMD(cmd=impl, return_value=self.geturl, checkf=verify_fastapi_func)

geturl(job=None)

Source code in lazyllm\components\deploy\lmdeploy.py

def geturl(self, job=None):
    if job is None:
        job = self.job
    if lazyllm.config['mode'] == lazyllm.Mode.Display:
        return 'http://{ip}:{port}/v1/'
    else:
        return f'http://{job.get_jobip()}:{self.kw["server-port"]}/v1/'

extract_result(x, inputs) staticmethod

Source code in lazyllm\components\deploy\lmdeploy.py

@staticmethod
def extract_result(x, inputs):
    return json.loads(x)['text']

lazyllm.components.deploy.base.DummyDeploy

Bases: LazyLLMDeployBase, Pipeline

Source code in lazyllm\components\deploy\base.py

class DummyDeploy(LazyLLMDeployBase, flows.Pipeline):
    keys_name_handle = {'inputs': 'inputs'}
    message_format = {
        'inputs': '',
        'parameters': {
            'do_sample': False,
            'temperature': 0.1,
        }
    }

    def __init__(self, launcher=launchers.remote(sync=False), *, stream=False, **kw):  # noqa B008
        super().__init__(launcher=launcher)

        def func():

            def impl(x):
                LOG.info(f'input is {x["inputs"]}, parameters is {x["parameters"]}')
                return f'reply for {x["inputs"]}, and parameters is {x["parameters"]}'

            def impl_stream(x):
                for s in ['reply', ' for', f' {x["inputs"]}', ', and',
                          ' parameters', ' is', f' {x["parameters"]}']:
                    yield s
                    time.sleep(0.2)
            return impl_stream if stream else impl
        flows.Pipeline.__init__(self, func,
                                lazyllm.deploy.RelayServer(port=random.randint(30000, 40000), launcher=launcher))

    def __call__(self, *args):
        url = flows.Pipeline.__call__(self)
        LOG.info(f'dummy deploy url is : {url}')
        return url

    def __repr__(self):
        return flows.Pipeline.__repr__(self)

lazyllm.components.auto.AutoDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\auto\autodeploy.py

class AutoDeploy(LazyLLMDeployBase):
    @staticmethod
    def _get_embed_deployer(launcher, type, kw):
        launcher = launcher or launchers.remote(ngpus=1)
        kw['model_type'] = type
        if lazyllm.config['default_embedding_engine'].lower() in ('transformers', 'flagembedding') \
            or kw.get('embed_type')=='sparse' or not check_requirements('infinity_emb'):
            return deploy.Rerank if type == 'rerank' else deploy.Embedding, launcher, kw
        else:
            return deploy.InfinityRerank if type == 'rerank' else deploy.Infinity, launcher, kw

    @classmethod
    def get_deployer(cls, base_model: str, source: Optional[str] = None, trust_remote_code: bool = True,
                     launcher: Optional[LazyLLMLaunchersBase] = None, type: Optional[str] = None,
                     log_path: Optional[str] = None, **kw):
        model_name = get_model_name(base_model)
        kw['log_path'], kw['trust_remote_code'] = log_path, trust_remote_code
        if not type:
            type = ModelManager.get_model_type(model_name)

        if type in ('embed', 'cross_modal_embed', 'rerank'):
            return AutoDeploy._get_embed_deployer(launcher, type, kw)
        elif type == 'sd':
            return StableDiffusionDeploy, launcher or launchers.remote(ngpus=1), kw
        elif type == 'stt':
            return SenseVoiceDeploy, launcher or launchers.remote(ngpus=1), kw
        elif type == 'tts':
            return TTSDeploy.get_deploy_cls(model_name), launcher or launchers.remote(ngpus=1), kw
        elif type == 'vlm':
            return deploy.LMDeploy, launcher or launchers.remote(ngpus=1), kw
        elif type == 'ocr':
            return OCRDeploy, launcher or launchers.remote(ngpus=1), kw

        if not launcher:
            match = re.search(r'(\d+)[bB]', model_name)
            size = int(match.group(1)) if match else 0
            size = (size * 2) if 'awq' not in model_name.lower() else (size / 1.5)
            ngpus = (1 << (math.ceil(size * 2 * 0.6 / config['gpu_memory']) - 1).bit_length())
            launcher = launchers.remote(ngpus = ngpus)

        for deploy_cls in ['vllm', 'lightllm', 'lmdeploy', 'mindie']:
            if check_cmd(deploy_cls) or check_requirements(requirements.get(deploy_cls)):
                deploy_cls = getattr(deploy, deploy_cls)
                return deploy_cls, launcher, kw
        return deploy.auto, launcher, kw

    def __new__(cls, base_model, source=lazyllm.config['model_source'], trust_remote_code=True,
                launcher=None, type=None, log_path=None, **kw):
        deploy_cls, launcher, kw = __class__.get_deployer(
            base_model=base_model, source=source, trust_remote_code=trust_remote_code,
            launcher=launcher, type=type, log_path=log_path, **kw)
        return deploy_cls(launcher=launcher, **kw)

lazyllm.components.deploy.embed.AbstractEmbedding

Bases: ABC

Source code in lazyllm\components\deploy\embed.py

class AbstractEmbedding(ABC):
    def __init__(self, base_embed, source=None, init=False):
        from ..utils.downloader import ModelManager
        self._source = source or lazyllm.config['model_source']
        self._base_embed = ModelManager(self._source).download(base_embed) or ''
        self._embed = None
        self._init = lazyllm.once_flag()
        if init:
            lazyllm.call_once(self._init, self.load_embed)

    @abstractmethod
    def load_embed(self) -> None:
        pass

    @abstractmethod
    def _call(self, data: Dict[str, Union[str, List[str]]]) -> str:
        pass

    def __call__(self, data: Dict[str, Union[str, List[str]]]) -> str:
        lazyllm.call_once(self._init, self.load_embed)
        return self._call(data)

    def __reduce__(self):
        init = bool(os.getenv('LAZYLLM_ON_CLOUDPICKLE', None) == 'ON' or self._init)
        return self.__class__, (self._base_embed, self._source, init)

lazyllm.components.deploy.EmbeddingDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\embed.py

class EmbeddingDeploy(LazyLLMDeployBase):
    message_format = {
        'text': 'text',  # str,
        'images': []  # Union[str, List[str]]
    }
    keys_name_handle = {
        'inputs': 'text',
        'image': 'images'
    }
    default_headers = {'Content-Type': 'application/json'}

    def __init__(self, launcher: LazyLLMLaunchersBase = None, model_type: str = 'embed', log_path: Optional[str] = None,
                 embed_type: Optional[str] = 'dense', trust_remote_code: bool = True, port: Optional[int] = None, **kw):
        super().__init__(launcher=launcher)
        self._launcher = launcher
        self._port = port
        self._model_type = model_type
        self._log_path = log_path
        self._sparse_embed = True if embed_type == 'sparse' else False
        self._trust_remote_code = trust_remote_code
        self._port = port

    def _get_model_path(self, finetuned_model=None, base_model=None):
        if not os.path.exists(finetuned_model) or \
            not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                    for filename in os.listdir(finetuned_model)):
            if not finetuned_model:
                LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                            f'base_model({base_model}) will be used')
            finetuned_model = base_model
        return finetuned_model

    def __call__(self, finetuned_model=None, base_model=None):
        finetuned_model = self._get_model_path(finetuned_model, base_model)
        if self._sparse_embed or lazyllm.config['default_embedding_engine'] == 'flagEmbedding':
            return lazyllm.deploy.RelayServer(port=self._port, func=LazyFlagEmbedding(
                finetuned_model, sparse=self._sparse_embed),
                launcher=self._launcher, log_path=self._log_path, cls='embedding')()
        else:
            return lazyllm.deploy.RelayServer(port=self._port, func=HuggingFaceEmbedding(finetuned_model),
                                              launcher=self._launcher, log_path=self._log_path, cls='embedding')()

lazyllm.components.deploy.embed.RerankDeploy

Bases: EmbeddingDeploy

Source code in lazyllm\components\deploy\embed.py

class RerankDeploy(EmbeddingDeploy):
    message_format = {'query': 'query', 'documents': ['string'], 'top_n': 1}
    keys_name_handle = {'inputs': 'query', 'documents': 'documents', 'top_n': 'top_n'}
    default_headers = {'Content-Type': 'application/json'}

    def __call__(self, finetuned_model=None, base_model=None):
        finetuned_model = self._get_model_path(finetuned_model, base_model)
        return lazyllm.deploy.RelayServer(port=self._port, func=LazyHuggingFaceRerank(
            finetuned_model), launcher=self._launcher, log_path=self._log_path, cls='embedding')()

lazyllm.components.deploy.embed.LazyHuggingFaceRerank

Bases: object

Source code in lazyllm\components\deploy\embed.py

class LazyHuggingFaceRerank(object):
    def __init__(self, base_rerank, source=None, init=False):
        from ..utils.downloader import ModelManager
        source = lazyllm.config['model_source'] if not source else source
        self.base_rerank = ModelManager(source).download(base_rerank) or ''
        self.reranker = None
        self.init_flag = lazyllm.once_flag()
        if init:
            lazyllm.call_once(self.init_flag, self.load_reranker)

    def load_reranker(self):
        self.reranker = sentence_transformers.CrossEncoder(self.base_rerank)

    def __call__(self, inps):
        lazyllm.call_once(self.init_flag, self.load_reranker)
        query, documents, top_n = inps['query'], inps['documents'], inps['top_n']
        query_pairs = [(query, doc) for doc in documents]
        scores = self.reranker.predict(query_pairs)
        sorted_indices = [(index, scores[index]) for index in np.argsort(scores)[::-1]]
        if top_n > 0:
            sorted_indices = sorted_indices[:top_n]
        return sorted_indices

    @classmethod
    def rebuild(cls, base_rerank, init):
        return cls(base_rerank, init)

    def __reduce__(self):
        init = bool(os.getenv('LAZYLLM_ON_CLOUDPICKLE', None) == 'ON' or self.init_flag)
        return LazyHuggingFaceRerank.rebuild, (self.base_rerank, init)

load_reranker()

Source code in lazyllm\components\deploy\embed.py

def load_reranker(self):
    self.reranker = sentence_transformers.CrossEncoder(self.base_rerank)

rebuild(base_rerank, init) classmethod

Source code in lazyllm\components\deploy\embed.py

@classmethod
def rebuild(cls, base_rerank, init):
    return cls(base_rerank, init)

lazyllm.components.deploy.embed.HuggingFaceEmbedding

Source code in lazyllm\components\deploy\embed.py

class HuggingFaceEmbedding:
    _model_id_mapping = {}

    @classmethod
    def get_emb_cls(cls, model_name: str):
        model_id = model_name.split('/')[-1].lower()
        return cls._model_id_mapping.get(model_id, LazyHuggingFaceDefaultEmbedding)

    @classmethod
    def register(cls, model_ids: List[str]):
        def decorator(target_class):
            for ele in model_ids:
                cls._model_id_mapping[ele.lower()] = target_class
            return target_class
        return decorator

    def __init__(self, base_embed, source=None):
        self._embed = self.__class__.get_emb_cls(base_embed)(base_embed, source)

    def load_embed(self):
        self._embed.load_embed()

    def __call__(self, *args, **kwargs):
        try:
            args[0]['images'] = [_base64_to_file(image) if _is_base64_with_mime(image) else image
                                 for image in args[0]['images']]
        except Exception as e:
            LOG.error(f'Error converting base64 to image: {e}')
        return self._embed(*args, **kwargs)

get_emb_cls(model_name) classmethod

Source code in lazyllm\components\deploy\embed.py

@classmethod
def get_emb_cls(cls, model_name: str):
    model_id = model_name.split('/')[-1].lower()
    return cls._model_id_mapping.get(model_id, LazyHuggingFaceDefaultEmbedding)

register(model_ids) classmethod

Source code in lazyllm\components\deploy\embed.py

@classmethod
def register(cls, model_ids: List[str]):
    def decorator(target_class):
        for ele in model_ids:
            cls._model_id_mapping[ele.lower()] = target_class
        return target_class
    return decorator

load_embed()

Source code in lazyllm\components\deploy\embed.py

def load_embed(self):
    self._embed.load_embed()

lazyllm.components.deploy.embed.LazyFlagEmbedding

Bases: object

Source code in lazyllm\components\deploy\embed.py

class LazyFlagEmbedding(object):
    def __init__(self, base_embed, sparse=False, source=None, init=False):
        from ..utils.downloader import ModelManager
        source = lazyllm.config['model_source'] if not source else source
        self.base_embed = ModelManager(source).download(base_embed) or ''
        self.embed = None
        self.device = 'cpu'
        self.sparse = sparse
        self.init_flag = lazyllm.once_flag()
        if init:
            lazyllm.call_once(self.init_flag, self.load_embed)

    def load_embed(self):
        self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
        self.embed = fe.FlagAutoModel.from_finetuned(self.base_embed, use_fp16=False, devices=[self.device])

    def __call__(self, data: Dict[str, Union[str, List[str]]]):
        lazyllm.call_once(self.init_flag, self.load_embed)
        string, _ = data['text'], data['images']
        with torch.no_grad():
            model_output = self.embed.encode(string, return_sparse=self.sparse)
        if self.sparse:
            embeddings = model_output['lexical_weights']
            if isinstance(string, list):
                res = [dict(embedding) for embedding in embeddings]
            else:
                res = dict(embeddings)
        else:
            res = model_output['dense_vecs'].tolist()

        if type(string) is list and type(res) is dict:
            return json.dumps([res], default=lambda x: float(x))
        else:
            return json.dumps(res, default=lambda x: float(x))

    @classmethod
    def rebuild(cls, base_embed, sparse, init):
        return cls(base_embed, sparse, init=init)

    def __reduce__(self):
        init = bool(os.getenv('LAZYLLM_ON_CLOUDPICKLE', None) == 'ON' or self.init_flag)
        return LazyFlagEmbedding.rebuild, (self.base_embed, self.sparse, init)

load_embed()

Source code in lazyllm\components\deploy\embed.py

def load_embed(self):
    self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
    self.embed = fe.FlagAutoModel.from_finetuned(self.base_embed, use_fp16=False, devices=[self.device])

rebuild(base_embed, sparse, init) classmethod

Source code in lazyllm\components\deploy\embed.py

@classmethod
def rebuild(cls, base_embed, sparse, init):
    return cls(base_embed, sparse, init=init)

lazyllm.components.deploy.Mindie

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\mindie.py

class Mindie(LazyLLMDeployBase):
    keys_name_handle = {
        'inputs': 'prompt',
    }
    default_headers = {'Content-Type': 'application/json'}
    message_format = {
        'prompt': 'Who are you ?',
        'stream': False,
        'max_tokens': 4096,
        'presence_penalty': 1.03,
        'frequency_penalty': 1.0,
        'temperature': 0.5,
        'top_p': 0.95
    }
    auto_map = {
        'port': int,
        'tp': ('worldSize', int),
        'max_input_token_len': ('maxInputTokenLen', int),
        'max_prefill_tokens': ('maxPrefillTokens', int),
        'max_seq_len': ('maxSeqLen', int)
    }

    def __init__(self, trust_remote_code=True, launcher=launchers.remote(), log_path=None, **kw):  # noqa B008
        super().__init__(launcher=launcher)
        assert lazyllm.config['mindie_home'], 'Ensure you have installed MindIE and \
                                  "export LAZYLLM_MINDIE_HOME=/path/to/mindie/latest"'
        self.mindie_home = lazyllm.config['mindie_home']
        self.mindie_config_path = os.path.join(self.mindie_home, 'mindie-service/conf/config.json')
        self.backup_path = self.mindie_config_path + '.backup'
        self.custom_config = kw.pop('config', None)
        self.kw = ArgsDict({
            'npuDeviceIds': [[0]],
            'worldSize': 1,
            'port': 'auto',
            'host': '0.0.0.0',
            'maxSeqLen': 64000,
            'maxInputTokenLen': 4096,
            'maxPrefillTokens': 8192,
        })
        self.trust_remote_code = trust_remote_code
        self.options_keys = kw.pop('options_keys', [])
        assert len(self.options_keys) == 0, 'options_keys is not supported'
        kw.update({'skip_check': False})
        self.kw.check_and_update(kw)
        self.kw['npuDeviceIds'] = [[i for i in range(self.kw.get('worldSize', 1))]]
        self.random_port = False if 'port' in kw and kw['port'] and kw['port'] != 'auto' else True
        self.temp_folder = make_log_dir(log_path, 'mindie') if log_path else None

        if self.custom_config:
            self.config_dict = (ArgsDict(self.load_config(self.custom_config))
                                if isinstance(self.custom_config, str) else ArgsDict(self.custom_config))
            self.kw['host'] = self.config_dict['ServerConfig']['ipAddress']
            self.kw['port'] = self.config_dict['ServerConfig']['port']
        else:
            default_config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'mindie', 'config.json')
            self.config_dict = ArgsDict(self.load_config(default_config_path))

    def __del__(self):
        if hasattr(self, 'backup_path') and os.path.isfile(self.backup_path):
            shutil.copy2(self.backup_path, self.mindie_config_path)

    def load_config(self, config_path):
        with open(config_path, 'r') as file:
            config_dict = json.load(file)
        return config_dict

    def save_config(self):
        if os.path.isfile(self.mindie_config_path):
            shutil.copy2(self.mindie_config_path, self.backup_path)

        with open(self.mindie_config_path, 'w') as file:
            json.dump(self.config_dict, file)

    def update_config(self):
        backend_config = self.config_dict['BackendConfig']
        backend_config['npuDeviceIds'] = self.kw['npuDeviceIds']
        model_config = {
            'modelName': self.finetuned_model.split('/')[-1],
            'modelWeightPath': self.finetuned_model,
            'worldSize': self.kw['worldSize'],
            'trust_remote_code': self.trust_remote_code
        }
        backend_config['ModelDeployConfig']['ModelConfig'][0].update(model_config)
        backend_config['ModelDeployConfig']['maxSeqLen'] = self.kw['maxSeqLen']
        backend_config['ModelDeployConfig']['maxInputTokenLen'] = self.kw['maxInputTokenLen']
        backend_config['ScheduleConfig']['maxPrefillTokens'] = self.kw['maxPrefillTokens']
        self.config_dict['BackendConfig'] = backend_config
        if self.kw['host'] != '0.0.0.0':
            self.config_dict['ServerConfig']['ipAddress'] = self.kw['host']
        self.config_dict['ServerConfig']['port'] = self.kw['port']

    def cmd(self, finetuned_model=None, base_model=None, master_ip=None):
        if self.custom_config is None:
            self.finetuned_model = finetuned_model
            if finetuned_model or base_model:
                if not os.path.exists(finetuned_model) or \
                    not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                            for filename in os.listdir(finetuned_model)):
                    if not finetuned_model:
                        LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                                    f'base_model({base_model}) will be used')
                    self.finetuned_model = base_model

            if self.random_port:
                self.kw['port'] = random.randint(30000, 40000)

            self.update_config()

        self.save_config()

        def impl():
            cmd = f'{os.path.join(self.mindie_home, "mindie-service/bin/mindieservice_daemon")}'
            if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
            return cmd

        return LazyLLMCMD(cmd=impl, return_value=self.geturl, checkf=verify_fastapi_func)

    def geturl(self, job=None):
        if job is None:
            job = self.job
        if lazyllm.config['mode'] == lazyllm.Mode.Display:
            return f'http://{job.get_jobip()}:{self.kw["port"]}/generate'
        else:
            LOG.info(f'MindIE Server running on http://{job.get_jobip()}:{self.kw["port"]}')
            return f'http://{job.get_jobip()}:{self.kw["port"]}/generate'

    @staticmethod
    def extract_result(x, inputs):
        return json.loads(x)['text'][0]

lazyllm.components.deploy.OCRDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\ocr\pp_ocr.py

class OCRDeploy(LazyLLMDeployBase):
    keys_name_handle = {
        'inputs': 'inputs',
        'ocr_files': 'inputs',
    }
    message_format = {'inputs': '/path/to/pdf'}
    default_headers = {'Content-Type': 'application/json'}

    def __init__(self, launcher=None, log_path=None, trust_remote_code=True, port=None, **kw):
        super().__init__(launcher=launcher)
        self._log_path = log_path
        self._trust_remote_code = trust_remote_code
        self._port = port

    def __call__(self, finetuned_model=None, base_model=None):
        if not finetuned_model:
            finetuned_model = base_model
        return lazyllm.deploy.RelayServer(
            port=self._port, func=_OCR(finetuned_model), launcher=self._launcher, log_path=self._log_path, cls='ocr')()

lazyllm.components.deploy.Infinity

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\infinity.py

class Infinity(LazyLLMDeployBase):
    keys_name_handle = {
        'inputs': 'input',
    }
    message_format = {
        'input': 'who are you ?',
    }
    default_headers = {'Content-Type': 'application/json'}
    target_name = 'embeddings'

    def __init__(self, launcher=launchers.remote(ngpus=1), model_type='embed', log_path=None, **kw):  # noqa B008
        super().__init__(launcher=launcher)
        self.kw = ArgsDict({
            'host': '0.0.0.0',
            'port': None,
            'batch-size': 256,
        })
        self._model_type = model_type
        kw.pop('stream', '')
        # Infinity (embedding model) doesn't support tensor parallel, ignore 'tp' parameter
        kw.pop('tp', None)
        self.options_keys = kw.pop('options_keys', [])
        self.kw.check_and_update(kw)
        self.random_port = False if 'port' in kw and kw['port'] else True
        self.temp_folder = make_log_dir(log_path, 'lmdeploy') if log_path else None

    def cmd(self, finetuned_model=None, base_model=None):
        if not os.path.exists(finetuned_model) or \
            not any(filename.endswith('.bin') or filename.endswith('.safetensors')
                    for filename in os.listdir(finetuned_model)):
            if not finetuned_model:
                LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                            f'base_model({base_model}) will be used')
            finetuned_model = base_model

        def impl():
            if self.random_port:
                self.kw['port'] = random.randint(30000, 40000)
            cmd = f'infinity_emb v2 --model-id {finetuned_model} --no-bettertransformer '
            if isinstance(self._launcher, launchers.EmptyLauncher) and self._launcher.ngpus:
                available_gpus = self._launcher._get_idle_gpus()
                required_count = self._launcher.ngpus
                if required_count <= len(available_gpus):
                    try:
                        use_cuda_visible = lazyllm.config['cuda_visible']
                    except (KeyError, AttributeError):
                        use_cuda_visible = False
                    if use_cuda_visible:
                        # Use logical GPU IDs (0, 1, 2...) when CUDA_VISIBLE_DEVICES is set
                        gpu_ids = ','.join(map(str, range(required_count)))
                    else:
                        # Use physical GPU IDs when CUDA_VISIBLE_DEVICES is not set
                        gpu_ids = ','.join(map(str, available_gpus[:required_count]))
                    cmd += f'--device-id={gpu_ids} '
                else:
                    raise RuntimeError(
                        f'Insufficient GPUs available (required: {required_count}, '
                        f'available: {len(available_gpus)})'
                    )
            cmd += self.kw.parse_kwargs()
            cmd += ' ' + parse_options_keys(self.options_keys)
            if self.temp_folder: cmd += f' 2>&1 | tee {get_log_path(self.temp_folder)}'
            return cmd

        return LazyLLMCMD(cmd=impl, return_value=self.geturl, checkf=verify_fastapi_func)

    def geturl(self, job=None):
        if job is None:
            job = self.job
        if lazyllm.config['mode'] == lazyllm.Mode.Display:
            return f'http://<ip>:<port>/{self.target_name}'
        else:
            return f'http://{job.get_jobip()}:{self.kw["port"]}/{self.target_name}'

    @staticmethod
    def extract_result(x, inputs):
        try:
            res_object = json.loads(x)
        except Exception as e:
            LOG.warning(f'JSONDecodeError on load {x}')
            raise e
        assert 'object' in res_object
        object_type = res_object['object']
        if object_type == 'list':  # for infinity >= 0.0.64
            object_type = res_object['data'][0]['object']
        if object_type == 'embedding':
            res_list = [item['embedding'] for item in res_object['data']]
            if len(res_list) == 1 and type(inputs['input']) is str:
                res_list = res_list[0]
            return json.dumps(res_list)
        elif object_type == 'rerank':
            return [(x['index'], x['relevance_score']) for x in res_object['results']]

lazyllm.components.deploy.BertDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\bert.py

class BertDeploy(LazyLLMDeployBase):

    message_format = {
        'text_a': '',
        'text_b': '',
    }
    keys_name_handle = {'inputs': 'text_a'}
    default_headers = {'Content-Type': 'application/json'}
    stream_url_suffix = ''
    stream_parse_parameters: dict = {}

    def __init__(
        self,
        launcher: Optional[LazyLLMLaunchersBase] = None,
        log_path: Optional[str] = None,
        trust_remote_code: bool = True,
        port: Optional[int] = None,
        max_length: int = 512,
        device: Optional[str] = None,
        **kw,
    ):
        super().__init__(launcher=launcher)
        self._log_path = log_path
        self._trust_remote_code = trust_remote_code
        self._port = port
        self._max_length = max_length
        self._device = device
        if kw:
            LOG.warning(f'Bert deploy: ignoring unknown kwargs: {sorted(kw.keys())}')

    def __call__(self, finetuned_model=None, base_model=None):
        finetuned_model = finetuned_model or ''
        base_model = base_model or ''
        if not finetuned_model:
            model_path = base_model
        else:
            valid_local = (
                os.path.isdir(finetuned_model)
                and any(
                    f.endswith(('.bin', '.safetensors', '.pt'))
                    for f in os.listdir(finetuned_model)
                )
            )
            if valid_local:
                model_path = finetuned_model
            elif base_model:
                LOG.warning(
                    f'Note! finetuned_model({finetuned_model}) is not a local checkpoint with weights; '
                    f'using base_model({base_model}).'
                )
                model_path = base_model
            else:
                LOG.info(
                    f'Bert deploy: finetuned_model({finetuned_model}) is not a valid local '
                    f'checkpoint and no base_model provided; treating it as a remote model id.'
                )
                model_path = finetuned_model

        if not model_path:
            raise ValueError('Bert deploy: finetuned_model and base_model are both empty or invalid.')

        func = _BertSequenceClassificationService(
            model_path,
            trust_remote_code=self._trust_remote_code,
            max_length=self._max_length,
            device=self._device,
        )
        return RelayServer(
            port=self._port,
            func=func,
            launcher=self._launcher,
            log_path=self._log_path,
            cls='bert',
        )()

    @staticmethod
    def extract_result(output: str, inputs) -> str:
        return output

lazyllm.components.deploy.relay.base.RelayServer

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\relay\base.py

class RelayServer(LazyLLMDeployBase):
    keys_name_handle = None
    default_headers = {'Content-Type': 'application/json'}
    message_format = None

    def __init__(self, port=None, *, func=None, pre_func=None, post_func=None, pythonpath=None,
                 log_path=None, cls=None, launcher=launchers.remote(sync=False), num_replicas: int = 1,  # noqa B008
                 security_key: Optional[str] = None, defined_pos: Optional[str] = None,
                 pass_args_by_file: Optional[bool] = None):
        # func must dump in __call__ to wait for dependancies.
        self._func = func
        self._pre = dump_obj(pre_func)
        self._post = dump_obj(post_func)
        self._port, self._real_port = port, None
        self._pythonpath = pythonpath
        self._num_replicas = num_replicas
        self._security_key = security_key
        self._defined_pos = defined_pos
        # None means fall back to global config at call time.
        self._pass_args_by_file = pass_args_by_file
        super().__init__(launcher=launcher)
        self.temp_folder = make_log_dir(log_path, cls or 'relay') if log_path else None

    @property
    def _file_args_forced(self) -> bool:
        if self._pass_args_by_file is not None:
            return self._pass_args_by_file
        return config['pass_args_by_file']

    def _prepare_obj_arg(self, serialised: Optional[str], *, force: bool = False) -> Optional[str]:
        '''Return *serialised* as-is when small, or spill it to a temp file and
        return a ``@file:<path>`` reference when it exceeds the threshold.
        Pass ``force=True`` to always write to a file regardless of size.'''
        if serialised is None:
            return None
        if not force and len(serialised) <= _CMD_ARG_SIZE_THRESHOLD:
            return serialised
        temp_dir = os.path.abspath(config['temp_dir'])
        os.makedirs(temp_dir, exist_ok=True)
        fd, path = tempfile.mkstemp(suffix='.pkl', prefix='lazyllm_relay_', dir=temp_dir)
        raw = base64.b64decode(serialised.encode('utf-8'))
        with os.fdopen(fd, 'wb') as fp:
            fp.write(raw)
        return f'@file:{path}'

    @staticmethod
    def _join_command(args) -> str:
        if os.name == 'nt':
            return subprocess.list2cmdline([str(arg) for arg in args])
        return shlex.join([str(arg) for arg in args])

    def cmd(self, func=None):
        FastapiApp.update()
        self._func = dump_obj(func or self._func)
        folder_path = os.path.dirname(os.path.abspath(__file__))
        run_file_path = os.path.join(folder_path, 'server.py')

        def impl():
            self._real_port = self._port if self._port else random.randint(30000, 40000)
            force = self._file_args_forced
            func_arg = self._prepare_obj_arg(self._func, force=force)
            args = [
                sys.executable,
                run_file_path,
                f'--open_port={self._real_port}',
                f'--function={func_arg}',
            ]
            if self._pre:
                args.append(f'--before_function={self._prepare_obj_arg(self._pre, force=force)}')
            if self._post:
                args.append(f'--after_function={self._prepare_obj_arg(self._post, force=force)}')
            if self._pythonpath:
                args.append(f'--pythonpath={self._pythonpath}')
            if self._num_replicas > 1 and config['use_ray']:
                args.append(f'--num_replicas={self._num_replicas}')
            if self._security_key:
                args.append(f'--security_key={self._security_key}')
            if self._defined_pos:
                defined_pos = dump_obj(self._defined_pos.replace('"', r'\"'))
                args.append(f'--defined_pos={self._prepare_obj_arg(defined_pos, force=force)}')
            cmd = self._join_command(args)
            if self.temp_folder:
                log_path = get_log_path(self.temp_folder)
                if os.name == 'nt':
                    cmd += f' > {subprocess.list2cmdline([log_path])} 2>&1'
                else:
                    cmd += f' 2>&1 | tee {shlex.quote(log_path)}'
            return cmd

        return LazyLLMCMD(cmd=impl, return_value=self.geturl,
                          checkf=verify_ray_func if config['use_ray'] else verify_fastapi_func,
                          no_displays=['function', 'before_function', 'after_function', 'security_key', 'defined_pos'])

    def geturl(self, job=None):
        if job is None:
            job = self.job
        return f'http://{job.get_jobip()}:{self._real_port}/generate'

lazyllm.components.deploy.text_to_speech.utils.TTSBase

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\text_to_speech\utils.py

class TTSBase(LazyLLMDeployBase):
    func = None

    def __init__(self, launcher: LazyLLMLaunchersBase = None,
                 log_path: Optional[str] = None, port: Optional[int] = None, **kw):
        super().__init__(launcher=launcher)
        self._log_path = log_path
        self._port = port

    def __call__(self, finetuned_model=None, base_model=None):
        if not finetuned_model:
            finetuned_model = base_model
        elif not os.path.exists(finetuned_model) or \
            not any(file.endswith(('.bin', '.safetensors'))
                    for _, _, filenames in os.walk(finetuned_model) for file in filenames):
            LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                        f'base_model({base_model}) will be used')
            finetuned_model = base_model
        return lazyllm.deploy.RelayServer(port=self._port, func=self.__class__.func(finetuned_model),
                                          launcher=self._launcher, log_path=self._log_path, cls='tts')()

lazyllm.components.deploy.relay.base.FastapiApp

Bases: object

Source code in lazyllm\components\deploy\relay\base.py

class FastapiApp(object):
    __relay_services__ = []

    @staticmethod
    def _server(method, path, **kw):
        def impl(f):
            FastapiApp.__relay_services__.append([f, method, path, kw])
            return f
        return impl

    @staticmethod
    def get(path, **kw):
        return FastapiApp._server('get', path, **kw)

    @staticmethod
    def post(path, **kw):
        return FastapiApp._server('post', path, **kw)

    @staticmethod
    def list(path, **kw):
        return FastapiApp._server('list', path, **kw)

    @staticmethod
    def delete(path, **kw):
        return FastapiApp._server('delete', path, **kw)

    @staticmethod
    def update():
        for f, method, path, kw in FastapiApp.__relay_services__:
            cls = inspect._findclass(f)
            if '__relay_services__' not in cls.__dict__:
                cls.__relay_services__ = dict()
            cls.__relay_services__[method, path] = ([f.__name__, kw])
        FastapiApp.__relay_services__.clear()

    @staticmethod
    def _prepare_openapi(fastapi_app):
        '''Prepare dynamically registered routes before OpenAPI generation.'''
        for route in fastapi_app.routes:
            body_field = getattr(route, 'body_field', None)
            annotation = getattr(getattr(body_field, 'field_info', None), 'annotation', None)
            if not hasattr(annotation, 'model_rebuild'):
                continue
            endpoint = getattr(route, 'endpoint', None)
            globalns = getattr(endpoint, '__globals__', None)
            if globalns is None:
                globalns = getattr(getattr(endpoint, '__func__', None), '__globals__', {})
            annotation.model_rebuild(force=True, _types_namespace=globalns)

---

Prompter

lazyllm.components.prompter.LazyLLMPrompterBase

Source code in lazyllm\components\prompter\builtinPrompt.py

class LazyLLMPrompterBase(metaclass=LazyLLMRegisterMetaClass):
    ISA = '<!lazyllm-spliter!>'
    ISE = '</!lazyllm-spliter!>'

    def __init__(self, show=False, tools=None, skills=None, history=None, *, enable_system: bool = True):
        self._set_model_configs(system=_DEFULT_CONFIG if not config['disable_system_prompt'] else '',
                                sos='', soh='', soa='', eos='', eoh='', eoa='')
        self._show = show
        self._tools = _DynamicValue(tools) if callable(tools) else tools
        self._skills = _DynamicValue(skills) if callable(skills) else (skills or '')
        self._pre_hook = None
        self._history = history or []
        self._enable_system = enable_system

    def _init_prompt(self, template: str, instruction_template: str, split: Union[None, str] = None):
        self._template = template
        self._instruction_template = instruction_template
        if split:
            assert not hasattr(self, '_split')
            self._split = split

    @staticmethod
    def _get_extro_key_template(extra_keys, prefix='Here are some extra messages you can referred to:\n\n'):
        if extra_keys:
            if isinstance(extra_keys, str): extra_keys = [extra_keys]
            assert isinstance(extra_keys, (tuple, list)), 'Only str, tuple[str], list[str] are supported'
            return prefix + ''.join([f'### {k}:\n{{{k}}}\n\n' for k in extra_keys])
        return ''

    def _handle_tool_call_instruction(self, instruction, tools):
        tool_dict = {}
        for key in ['tool_start_token', 'tool_args_token', 'tool_end_token']:
            if getattr(self, f'_{key}', None) and key in instruction:
                tool_dict[key] = getattr(self, f'_{key}')
        if 'tool_names' in instruction: tool_dict['tool_names'] = self._get_tools_name(tools)
        return reduce(lambda s, kv: s.replace(f'{{{kv[0]}}}', kv[1]), tool_dict.items(), instruction)

    def _set_model_configs(self, system: str = None, sos: Union[None, str] = None, soh: Union[None, str] = None,
                           soa: Union[None, str] = None, eos: Union[None, str] = None,
                           eoh: Union[None, str] = None, eoa: Union[None, str] = None,
                           soe: Union[None, str] = None, eoe: Union[None, str] = None,
                           separator: Union[None, str] = None, plugin: Union[None, str] = None,
                           interpreter: Union[None, str] = None, stop_words: Union[None, List[str]] = None,
                           tool_start_token: Union[None, str] = None, tool_end_token: Union[None, str] = None,
                           tool_args_token: Union[None, str] = None):

        local = locals()
        for name in ['system', 'sos', 'soh', 'soa', 'eos', 'eoh', 'eoa', 'soe', 'eoe', 'tool_start_token',
                     'tool_end_token', 'tool_args_token']:
            if local[name] is not None: setattr(self, f'_{name}', local[name])

    def _resolve_dynamic(self, value, input, history):
        return value.resolve(input, history) if isinstance(value, _DynamicValue) else value

    def _get_tools(self, tools, *, for_chat_api: bool):
        return tools if for_chat_api else '### Function-call Tools. \n\n' +\
            f'{json.dumps(tools, ensure_ascii=False)}\n\n' if tools else ''

    def _get_tools_name(self, tools):
        return json.dumps([t['function']['name'] for t in tools], ensure_ascii=False) if tools else ''

    def _get_histories(self, history, *, for_chat_api: bool):  # noqa: C901
        if not self._history and not history: return ''
        if for_chat_api:
            content = []
            for item in self._history + (history or []):
                if isinstance(item, list):
                    assert len(item) <= 2, 'history item length cannot be greater than 2'
                    if len(item) > 0: content.append({'role': 'user', 'content': item[0]})
                    if len(item) > 1: content.append({'role': 'assistant', 'content': item[1]})
                elif isinstance(item, dict):
                    content.append(item)
                else:
                    LOG.error(f'history: {history}')
                    raise ValueError('history must be a list of list or dict')
            return content
        else:
            ret = ''.join([f'{self._soh}{h}{self._eoh}{self._soa}{a}{self._eoa}' for h, a in self._history])
            if not history: return ret
            if isinstance(history[0], list):
                return ret + ''.join([f'{self._soh}{h}{self._eoh}{self._soa}{a}{self._eoa}' for h, a in history])
            elif isinstance(history[0], dict):
                for item in history:
                    if item['role'] == 'user':
                        ret += f'{self._soh}{item["content"]}{self._eoh}'
                    elif item['role'] == 'assistant':
                        ret += f'{self._soa}'
                        ret += f'{item.get("content", "")}'
                        for idx in range(len(item.get('tool_calls', []))):
                            tool = item['tool_calls'][idx]['function']
                            if getattr(self, '_tool_args_token', None):
                                tool = tool['name'] + self._tool_args_token + \
                                    json.dumps(tool['arguments'], ensure_ascii=False)
                            ret += (f'{getattr(self, "_tool_start_token", "")}' + '\n'
                                    f'{tool}'
                                    f'{getattr(self, "_tool_end_token", "")}' + '\n')
                        ret += f'{self._eoa}'
                    elif item['role'] == 'tool':
                        try:
                            content = json.loads(item['content'].strip())
                        except Exception:
                            content = item['content']
                        ret += f'{getattr(self, "_soe", "")}{content}{getattr(self, "_eoe", "")}'

                return ret
            else:
                raise NotImplementedError('Cannot transform json history to {type(history[0])} now')

    def _get_instruction_and_input(self, input, *, for_chat_api: bool = False, tools=None):
        instruction = self._instruction_template
        fc_prompt = '' if for_chat_api or not tools else FC_PROMPT
        if fc_prompt and FC_PROMPT_PLACEHOLDER not in instruction:
            instruction = f'{instruction}\n\n{fc_prompt}'
        instruction = instruction.replace(FC_PROMPT_PLACEHOLDER, fc_prompt)
        instruction = self._handle_tool_call_instruction(instruction, tools)
        prompt_keys = list(set(re.findall(r'\{(\w+)\}', instruction)))
        if isinstance(input, (str, int)):
            if len(prompt_keys) == 1:
                return instruction.format(**{prompt_keys[0]: input}), ''
            else:
                assert len(prompt_keys) == 0
                return instruction, input
        assert isinstance(input, dict), f'expected types are str, int and dict, bug get {type(input)}(`{input})`'
        kwargs = {k: input.pop(k) for k in prompt_keys}
        assert len(input) <= 1, f'Unexpected keys found in input: {list(input.keys())}'
        return (reduce(lambda s, kv: s.replace(f'{{{kv[0]}}}', kv[1]),
                       kwargs.items(),
                       instruction)
                if len(kwargs) > 0 else instruction,
                list(input.values())[0] if input else '')

    def _check_values(self, instruction, input, history, tools): pass

    # Used for TrainableModule(local deployed)
    def _generate_prompt_impl(self, instruction, input, user, history, tools, label, skills):
        is_tool = False
        if isinstance(input, dict):
            input = input.get('content', '')
            is_tool = input.get('role') == 'tool'
        elif isinstance(input, list):
            is_tool = any(item.get('role') == 'tool' for item in input)
            input = '\n'.join([item.get('content', '') for item in input])
        params = dict(system=self._system, instruction=instruction, input=input, user=user, history=history, tools=tools,
                      skills=skills, sos=self._sos, eos=self._eos, soh=self._soh, eoh=self._eoh, soa=self._soa,
                      eoa=self._eoa)
        if is_tool:
            params['soh'] = getattr(self, '_soe', self._soh)
            params['eoh'] = getattr(self, '_eoe', self._eoh)
        return (self._template.format(**params) + (label if label else '')).lstrip('\n')

    # Used for OnlineChatModule
    def _generate_prompt_dict_impl(self, instruction, input, user, history, tools, label, skills):
        if not history: history = []
        if isinstance(input, str):
            history.append({'role': 'user', 'content': input})
        elif isinstance(input, dict):
            history.append(input)
        elif isinstance(input, list) and all(isinstance(ele, dict) for ele in input):
            history.extend(input)
        elif isinstance(input, tuple) and len(input) == 1:
            # Note tuple size 1 with one single string is not expected
            history.append({'role': 'user', 'content': input[0]})
        else:
            raise TypeError('input must be a string or a dict')

        if user:
            history[-1]['content'] = user + history[-1]['content']

        if self._enable_system:
            system_content = '\n'.join(p for p in (self._system, instruction, skills) if p)
            history.insert(0, {'role': 'system', 'content': system_content})
        return dict(messages=history, tools=tools) if tools else dict(messages=history)

    # Used for OnlineChatModule with Anthropic-format API
    def _generate_prompt_anthropic_impl(self, instruction, input, user, history, tools, label, skills):
        result = self._generate_prompt_dict_impl(instruction, input, user, history, tools, label, skills)
        messages = result.get('messages', [])
        system_text = None
        non_system = []
        for msg in messages:
            if msg.get('role') == 'system':
                system_text = msg['content']
            else:
                non_system.append(msg)
        out = dict(messages=non_system)
        if system_text is not None:
            out['system'] = system_text
        if tools:
            out['tools'] = result['tools']
        return out

    def pre_hook(self, func: Optional[Callable] = None):
        self._pre_hook = func
        return self

    def _split_instruction(self, instruction: str):
        system_instruction = instruction
        user_instruction = ''
        if LazyLLMPrompterBase.ISA in instruction and LazyLLMPrompterBase.ISE in instruction:
            # The instruction includes system prompts and/or user prompts
            pattern = re.compile(r'%s(.*)%s' % (LazyLLMPrompterBase.ISA, LazyLLMPrompterBase.ISE), re.DOTALL)
            ret = re.split(pattern, instruction)
            system_instruction = ret[0]
            user_instruction = ret[1]

        return system_instruction, user_instruction

    def generate_prompt(self, input: Union[str, List, Dict[str, str], None] = None,
                        history: List[Union[List[str], Dict[str, Any]]] = None,
                        tools: Union[List[Dict[str, Any]], None] = None,
                        label: Union[str, None] = None,
                        *, show: bool = False, return_dict: bool = False,
                        format: Optional[str] = None) -> Union[str, Dict]:
        if return_dict and format is None:
            LOG.log_once('return_dict is deprecated, use format="openai" instead.', level='warning')
            format = 'openai'
        input = copy.deepcopy(input)
        if self._pre_hook:
            input, history, tools, label = self._pre_hook(input, history, tools, label)
        tools = self._resolve_dynamic(tools or self._tools, input, history)
        skills = self._resolve_dynamic(self._skills, input, history)
        for_chat_api = bool(format)
        instruction, input = self._get_instruction_and_input(input, for_chat_api=for_chat_api, tools=tools)
        history = self._get_histories(history, for_chat_api=for_chat_api)
        tools = self._get_tools(tools, for_chat_api=for_chat_api)
        self._check_values(instruction, input, history, tools)
        instruction, user_instruction = self._split_instruction(instruction)
        if format == 'anthropic':
            func = self._generate_prompt_anthropic_impl
        elif format == 'openai':
            func = self._generate_prompt_dict_impl
        else:
            func = self._generate_prompt_impl
        result = func(instruction, input, user_instruction, history, tools, label, skills)
        if self._show or show: LOG.info(result)
        return result

    def get_response(self, output: str, input: Union[str, None] = None) -> str:
        if input and output.startswith(input):
            return output[len(input):]
        return output if getattr(self, '_split', None) is None else output.split(self._split)[-1]

lazyllm.components.prompter.EmptyPrompter

Bases: LazyLLMPrompterBase

Source code in lazyllm\components\prompter\builtinPrompt.py

class EmptyPrompter(LazyLLMPrompterBase):

    def generate_prompt(self, input, history=None, tools=None, label=None, *, show=False,
                        return_dict: bool = False, format=None):
        if return_dict and format is None:
            LOG.log_once('return_dict is deprecated, use format="openai" instead.', level='warning')
            format = 'openai'
        if format:
            return {'messages': [{'role': 'user', 'content': input}]}
        if self._show or show: LOG.info(input)
        return input

generate_prompt(input, history=None, tools=None, label=None, *, show=False, return_dict=False, format=None)

Source code in lazyllm\components\prompter\builtinPrompt.py

def generate_prompt(self, input, history=None, tools=None, label=None, *, show=False,
                    return_dict: bool = False, format=None):
    if return_dict and format is None:
        LOG.log_once('return_dict is deprecated, use format="openai" instead.', level='warning')
        format = 'openai'
    if format:
        return {'messages': [{'role': 'user', 'content': input}]}
    if self._show or show: LOG.info(input)
    return input

lazyllm.components.Prompter

Bases: object

Source code in lazyllm\components\prompter\prompter.py

class Prompter(object):
    def __init__(self, prompt=None, response_split=None, *, chat_prompt=None,
                 history_symbol='llm_chat_history', eoa=None, eoh=None, show=False):
        self._prompt, self._response_split = prompt, response_split
        self._chat_prompt = chat_prompt
        self._history_symbol, self._eoa, self._eoh = history_symbol, eoa, eoh
        self._show = show
        self._prompt_keys = list(set(re.findall(r'\{(\w+)\}', self._prompt))) if prompt else []
        if chat_prompt is not None:
            chat_keys = set(re.findall(r'\{(\w+)\}', self._chat_prompt))
            assert set(self._prompt_keys).issubset(chat_keys)
            assert chat_keys - set(self._prompt_keys) == set([self._history_symbol])
            self.use_history = True
        else:
            self.use_history = history_symbol in self._prompt_keys
            if self.use_history:
                self._prompt_keys.pop(self._prompt_keys.index(history_symbol))
                self._chat_prompt = self._prompt

    @classmethod
    def from_dict(cls, prompt, *, show=False):
        assert isinstance(prompt, dict)
        return cls(**prompt, show=show)

    @classmethod
    def from_template(cls, template_name, *, show=False):
        return cls.from_dict(templates[template_name], show=show)

    @classmethod
    def from_file(cls, fname, *, show=False):
        with open(fname) as fp:
            return cls.from_dict(json.load(fp), show=show)

    @classmethod
    def empty(cls):
        return cls()

    def _is_empty(self):
        return self._prompt is None

    def generate_prompt(self, input, history=None, tools=None, label=None, show=False):
        if not self._is_empty():
            assert tools is None
            # datasets.formatting.formatting.LazyDict is used in transformers
            if not isinstance(input, collections.abc.Mapping):
                assert len(self._prompt_keys) == 1, (
                    f'invalid prompt `{self._prompt}` for <{type(input)}> input `{input}`')
                input = {self._prompt_keys[0]: input}
            try:
                if self.use_history and isinstance(history, list) and len(history) > 0:
                    assert isinstance(history[0], list), 'history must be list of list'
                    input[self._history_symbol] = self._eoa.join([self._eoh.join(h) for h in history])
                    input = self._chat_prompt.format(**input)
                else:
                    if self.use_history: input[self._history_symbol] = ''
                    input = self._prompt.format(**input)
            except Exception:
                raise RuntimeError(f'Generate prompt failed, and prompt is {self._prompt}; chat-prompt'
                                   f' is {self._chat_prompt}; input is {input}; history is {history}')
            if label: input += label
        if self._show or show: LOG.info(input)
        return input

    def get_response(self, response, input=None):
        if input and response.startswith(input):
            return response[len(input):]
        return response if self._response_split is None else response.split(self._response_split)[-1]

from_dict(prompt, *, show=False) classmethod

Source code in lazyllm\components\prompter\prompter.py

@classmethod
def from_dict(cls, prompt, *, show=False):
    assert isinstance(prompt, dict)
    return cls(**prompt, show=show)

from_template(template_name, *, show=False) classmethod

Source code in lazyllm\components\prompter\prompter.py

@classmethod
def from_template(cls, template_name, *, show=False):
    return cls.from_dict(templates[template_name], show=show)

from_file(fname, *, show=False) classmethod

Source code in lazyllm\components\prompter\prompter.py

@classmethod
def from_file(cls, fname, *, show=False):
    with open(fname) as fp:
        return cls.from_dict(json.load(fp), show=show)

empty() classmethod

Source code in lazyllm\components\prompter\prompter.py

@classmethod
def empty(cls):
    return cls()

generate_prompt(input, history=None, tools=None, label=None, show=False)

Source code in lazyllm\components\prompter\prompter.py

def generate_prompt(self, input, history=None, tools=None, label=None, show=False):
    if not self._is_empty():
        assert tools is None
        # datasets.formatting.formatting.LazyDict is used in transformers
        if not isinstance(input, collections.abc.Mapping):
            assert len(self._prompt_keys) == 1, (
                f'invalid prompt `{self._prompt}` for <{type(input)}> input `{input}`')
            input = {self._prompt_keys[0]: input}
        try:
            if self.use_history and isinstance(history, list) and len(history) > 0:
                assert isinstance(history[0], list), 'history must be list of list'
                input[self._history_symbol] = self._eoa.join([self._eoh.join(h) for h in history])
                input = self._chat_prompt.format(**input)
            else:
                if self.use_history: input[self._history_symbol] = ''
                input = self._prompt.format(**input)
        except Exception:
            raise RuntimeError(f'Generate prompt failed, and prompt is {self._prompt}; chat-prompt'
                               f' is {self._chat_prompt}; input is {input}; history is {history}')
        if label: input += label
    if self._show or show: LOG.info(input)
    return input

get_response(response, input=None)

Source code in lazyllm\components\prompter\prompter.py

def get_response(self, response, input=None):
    if input and response.startswith(input):
        return response[len(input):]
    return response if self._response_split is None else response.split(self._response_split)[-1]

lazyllm.components.AlpacaPrompter

Bases: LazyLLMPrompterBase

Source code in lazyllm\components\prompter\alpacaPrompter.py

class AlpacaPrompter(LazyLLMPrompterBase):
    def __init__(self, instruction: Union[None, str, Dict[str, str]] = None, extra_keys: Union[None, List[str]] = None,
                 show: bool = False, tools: Optional[List] = None):
        super(__class__, self).__init__(show, tools=tools)
        extra_keys_template = LazyLLMPrompterBase._get_extro_key_template(extra_keys)
        _prefix = ('Below is an instruction that describes a task, paired with extra messages such as '
                   'input that provides further context if possible. Write a response that appropriately '
                   'completes the request.\n\n### Instruction:\n')
        if isinstance(instruction, dict):
            user_suffix = ('\n\n' + extra_keys_template) if extra_keys_template else ''
            splice_struction = instruction.get('system', '') + \
                AlpacaPrompter.ISA + instruction.get('user', '') + user_suffix + AlpacaPrompter.ISE
            instruction = splice_struction
            instruction_template = _prefix + instruction
        else:
            instruction_template = (_prefix + (instruction if instruction else '')
                                    + '\n\n' + extra_keys_template)
        self._init_prompt('{system}\n{instruction}\n{tools}\n{user}### Response:\n',
                          instruction_template,
                          '### Response:\n')

    def _check_values(self, instruction, input, history, tools):
        assert not history, f'Chat history is not supported in {__class__}.'
        assert not input, 'All keys should in instruction or extra-keys'

generate_prompt(input=None, history=None, tools=None, label=None, *, show=False, return_dict=False, format=None)

Source code in lazyllm\components\prompter\builtinPrompt.py

def generate_prompt(self, input: Union[str, List, Dict[str, str], None] = None,
                    history: List[Union[List[str], Dict[str, Any]]] = None,
                    tools: Union[List[Dict[str, Any]], None] = None,
                    label: Union[str, None] = None,
                    *, show: bool = False, return_dict: bool = False,
                    format: Optional[str] = None) -> Union[str, Dict]:
    if return_dict and format is None:
        LOG.log_once('return_dict is deprecated, use format="openai" instead.', level='warning')
        format = 'openai'
    input = copy.deepcopy(input)
    if self._pre_hook:
        input, history, tools, label = self._pre_hook(input, history, tools, label)
    tools = self._resolve_dynamic(tools or self._tools, input, history)
    skills = self._resolve_dynamic(self._skills, input, history)
    for_chat_api = bool(format)
    instruction, input = self._get_instruction_and_input(input, for_chat_api=for_chat_api, tools=tools)
    history = self._get_histories(history, for_chat_api=for_chat_api)
    tools = self._get_tools(tools, for_chat_api=for_chat_api)
    self._check_values(instruction, input, history, tools)
    instruction, user_instruction = self._split_instruction(instruction)
    if format == 'anthropic':
        func = self._generate_prompt_anthropic_impl
    elif format == 'openai':
        func = self._generate_prompt_dict_impl
    else:
        func = self._generate_prompt_impl
    result = func(instruction, input, user_instruction, history, tools, label, skills)
    if self._show or show: LOG.info(result)
    return result

get_response(output, input=None)

Source code in lazyllm\components\prompter\builtinPrompt.py

def get_response(self, output: str, input: Union[str, None] = None) -> str:
    if input and output.startswith(input):
        return output[len(input):]
    return output if getattr(self, '_split', None) is None else output.split(self._split)[-1]

lazyllm.components.ChatPrompter

Bases: LazyLLMPrompterBase

Source code in lazyllm\components\prompter\chatPrompter.py

class ChatPrompter(LazyLLMPrompterBase):
    def __init__(self, instruction: Union[None, str, Dict[str, str]] = None, extra_keys: Union[None, List[str]] = None,
                 show: bool = False, tools: Optional[List] = None, skills: Optional[List] = None,
                 history: Optional[List[List[str]]] = None, *, enable_system: bool = True):
        super(__class__, self).__init__(show, tools=tools, skills=skills, history=history, enable_system=enable_system)
        extra_keys_template = LazyLLMPrompterBase._get_extro_key_template(extra_keys)
        if isinstance(instruction, dict):
            splice_instruction = instruction.get('system', '') + \
                ChatPrompter.ISA + instruction.get('user', '') + extra_keys_template + ChatPrompter.ISE
            instruction = splice_instruction
            instruction_template = f'{instruction}\n' if instruction else ''
        else:
            instruction_template = f'{instruction}\n{extra_keys_template}\n' if instruction else ''
        self._init_prompt(
            '{sos}{system}{instruction}{skills}{tools}{eos}\n\n{history}\n{soh}\n{user}{input}\n{eoh}{soa}\n',
            instruction_template)

    @property
    def _split(self): return f'{self._soa}\n' if self._soa else None

generate_prompt(input=None, history=None, tools=None, label=None, *, show=False, return_dict=False, format=None)

Source code in lazyllm\components\prompter\builtinPrompt.py

def generate_prompt(self, input: Union[str, List, Dict[str, str], None] = None,
                    history: List[Union[List[str], Dict[str, Any]]] = None,
                    tools: Union[List[Dict[str, Any]], None] = None,
                    label: Union[str, None] = None,
                    *, show: bool = False, return_dict: bool = False,
                    format: Optional[str] = None) -> Union[str, Dict]:
    if return_dict and format is None:
        LOG.log_once('return_dict is deprecated, use format="openai" instead.', level='warning')
        format = 'openai'
    input = copy.deepcopy(input)
    if self._pre_hook:
        input, history, tools, label = self._pre_hook(input, history, tools, label)
    tools = self._resolve_dynamic(tools or self._tools, input, history)
    skills = self._resolve_dynamic(self._skills, input, history)
    for_chat_api = bool(format)
    instruction, input = self._get_instruction_and_input(input, for_chat_api=for_chat_api, tools=tools)
    history = self._get_histories(history, for_chat_api=for_chat_api)
    tools = self._get_tools(tools, for_chat_api=for_chat_api)
    self._check_values(instruction, input, history, tools)
    instruction, user_instruction = self._split_instruction(instruction)
    if format == 'anthropic':
        func = self._generate_prompt_anthropic_impl
    elif format == 'openai':
        func = self._generate_prompt_dict_impl
    else:
        func = self._generate_prompt_impl
    result = func(instruction, input, user_instruction, history, tools, label, skills)
    if self._show or show: LOG.info(result)
    return result

get_response(output, input=None)

Source code in lazyllm\components\prompter\builtinPrompt.py

def get_response(self, output: str, input: Union[str, None] = None) -> str:
    if input and output.startswith(input):
        return output[len(input):]
    return output if getattr(self, '_split', None) is None else output.split(self._split)[-1]

---

MultiModal

Text to Image

lazyllm.components.StableDiffusionDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\stable_diffusion\stable_diffusion3.py

class StableDiffusionDeploy(LazyLLMDeployBase):
    message_format = None
    keys_name_handle = None
    default_headers = {'Content-Type': 'application/json'}

    def __init__(self, launcher: Optional[LazyLLMLaunchersBase] = None,
                 log_path: Optional[str] = None, trust_remote_code: bool = True, port: Optional[int] = None, **kw):
        super().__init__(launcher=launcher)
        self._log_path = log_path
        self._trust_remote_code = trust_remote_code
        self._port = port

    def __call__(self, finetuned_model=None, base_model=None):
        if not finetuned_model:
            finetuned_model = base_model
        elif not os.path.exists(finetuned_model) or \
            not any(file.endswith(('.bin', '.safetensors'))
                    for _, _, filenames in os.walk(finetuned_model) for file in filenames):
            LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                        f'base_model({base_model}) will be used')
            finetuned_model = base_model
        return lazyllm.deploy.RelayServer(port=self._port, func=_StableDiffusion3(finetuned_model),
                                          launcher=self._launcher, log_path=self._log_path, cls='stable_diffusion')()

Visual Question Answering

Reference LMDeploy, which supports the Visual Question Answering model.

Text to Sound

lazyllm.components.TTSDeploy

Source code in lazyllm\components\deploy\text_to_speech\__init__.py

class TTSDeploy:

    def __new__(cls, name, **kwarg):
        return cls.get_deploy_cls(name)(**kwarg)

    @classmethod
    def get_deploy_cls(cls, name):
        name = name.lower()
        if name == 'bark':
            return BarkDeploy
        elif name in ('chattts', 'chattts-new'):
            raise RuntimeError('ChatTTS is deprecated and no longer supported.')
            return ChatTTSDeploy
        elif name.startswith('musicgen'):
            return MusicGenDeploy
        else:
            raise RuntimeError(f'Not support model: {name}')

lazyllm.components.ChatTTSDeploy

Bases: TTSBase

Source code in lazyllm\components\deploy\text_to_speech\chattts.py

class ChatTTSDeploy(TTSBase):
    keys_name_handle = {
        'inputs': 'inputs',
    }
    message_format = {
        'inputs': 'Who are you ?',
        'refinetext': {
            'prompt': '[oral_2][laugh_0][break_6]',
            'top_P': 0.7,
            'top_K': 20,
            'temperature': 0.7,
            'repetition_penalty': 1.0,
            'max_new_token': 384,
            'min_new_token': 0,
            'show_tqdm': True,
            'ensure_non_empty': True,
        },
        'infercode': {
            'prompt': '[speed_5]',
            'spk_emb': None,
            'temperature': 0.3,
            'repetition_penalty': 1.05,
            'max_new_token': 2048,
        }

    }
    default_headers = {'Content-Type': 'application/json'}
    func = _ChatTTSModule

lazyllm.components.BarkDeploy

Bases: TTSBase

Source code in lazyllm\components\deploy\text_to_speech\bark.py

class BarkDeploy(TTSBase):
    keys_name_handle = {
        'inputs': 'inputs',
    }
    message_format = {
        'inputs': 'Who are you ?',
        'voice_preset': None,
    }
    default_headers = {'Content-Type': 'application/json'}

    func = _Bark

lazyllm.components.MusicGenDeploy

Bases: TTSBase

Source code in lazyllm\components\deploy\text_to_speech\musicgen.py

class MusicGenDeploy(TTSBase):
    message_format = None
    keys_name_handle = None
    default_headers = {'Content-Type': 'application/json'}
    func = _MusicGen

Speech to Text

lazyllm.components.SenseVoiceDeploy

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\speech_to_text\sense_voice.py

class SenseVoiceDeploy(LazyLLMDeployBase):
    keys_name_handle = {
        'inputs': 'inputs',
        'audio': 'audio',
    }
    message_format = {
        'inputs': 'Who are you ?',
        'audio': None,
    }
    default_headers = {'Content-Type': 'application/json'}

    def __init__(self, launcher: Optional[LazyLLMLaunchersBase] = None,
                 log_path: Optional[str] = None, trust_remote_code: bool = True, port: Optional[int] = None, **kw):
        super().__init__(launcher=launcher)
        self._log_path = log_path
        self._trust_remote_code = trust_remote_code
        self._port = port

    def __call__(self, finetuned_model=None, base_model=None):
        if not finetuned_model:
            finetuned_model = base_model
        elif not os.path.exists(finetuned_model) or \
            not any(file.endswith(('.pt', '.bin', '.safetensors'))
                    for _, _, filenames in os.walk(finetuned_model) for file in filenames):
            LOG.warning(f'Note! That finetuned_model({finetuned_model}) is an invalid path, '
                        f'base_model({base_model}) will be used')
            finetuned_model = base_model
        return lazyllm.deploy.RelayServer(port=self._port, func=SenseVoice(finetuned_model), launcher=self._launcher,
                                          log_path=self._log_path, cls='sensevoice')()

lazyllm.components.deploy.speech_to_text.sense_voice.SenseVoice

Bases: object

Source code in lazyllm\components\deploy\speech_to_text\sense_voice.py

class SenseVoice(object):
    def __init__(self, base_path, source=None, init=False):
        source = lazyllm.config['model_source'] if not source else source
        self.base_path = ModelManager(source).download(base_path) or ''
        self.model = None
        self.init_flag = lazyllm.once_flag()
        if init:
            lazyllm.call_once(self.init_flag, self.load_stt)

    def load_stt(self):
        if importlib.util.find_spec('torch_npu') is not None:
            import torch_npu  # noqa F401
            from torch_npu.contrib import transfer_to_npu  # noqa F401

        self.model = funasr.AutoModel(
            model=self.base_path,
            trust_remote_code=False,
            vad_model='fsmn-vad',
            vad_kwargs={'max_single_segment_time': 30000},
            device='cuda:0',
        )

    def __call__(self, string):
        lazyllm.call_once(self.init_flag, self.load_stt)
        if isinstance(string, dict):
            if string['audio']:
                string = string['audio'][-1] if isinstance(string['audio'], list) else string['audio']
            else:
                string = string['inputs']
        assert isinstance(string, str)
        string = string.strip()
        try:
            string = _base64_to_file(string) if _is_base64_with_mime(string) else string
        except Exception as e:
            LOG.error(f'Error processing base64 encoding: {e}')
            return f'Error processing base64 encoding {e}'
        if not string.endswith(supported_formats):
            return f'Only {", ".join(supported_formats)} formats in the form of file paths or URLs are supported.'
        if not is_valid_path(string) and not is_valid_url(string):
            return f'This {string} is not a valid URL or file path. Please check.'
        res = self.model.generate(
            input=string,
            cache={},
            language='auto',  # 'zn', 'en', 'yue', 'ja', 'ko', 'nospeech'
            use_itn=True,
            batch_size_s=60,
            merge_vad=True,
            merge_length_s=15,
        )
        text = funasr.utils.postprocess_utils.rich_transcription_postprocess(res[0]['text'])
        return text

    @classmethod
    def rebuild(cls, base_path, init):
        return cls(base_path, init=init)

    def __reduce__(self):
        init = bool(os.getenv('LAZYLLM_ON_CLOUDPICKLE', None) == 'ON' or self.init_flag)
        return SenseVoice.rebuild, (self.base_path, init)

load_stt()

Source code in lazyllm\components\deploy\speech_to_text\sense_voice.py

def load_stt(self):
    if importlib.util.find_spec('torch_npu') is not None:
        import torch_npu  # noqa F401
        from torch_npu.contrib import transfer_to_npu  # noqa F401

    self.model = funasr.AutoModel(
        model=self.base_path,
        trust_remote_code=False,
        vad_model='fsmn-vad',
        vad_kwargs={'max_single_segment_time': 30000},
        device='cuda:0',
    )

rebuild(base_path, init) classmethod

Source code in lazyllm\components\deploy\speech_to_text\sense_voice.py

@classmethod
def rebuild(cls, base_path, init):
    return cls(base_path, init=init)

---

ModelManager

lazyllm.components.ModelManager

Source code in lazyllm\components\utils\downloader\model_downloader.py

class ModelManager():
    def __init__(self, model_source, token=lazyllm.config['model_source_token'],
                 cache_dir=lazyllm.config['model_cache_dir'], model_path=lazyllm.config['model_path']):
        self.model_source = model_source or lazyllm.config['model_source']
        self.token = token or None
        self.cache_dir = cache_dir
        self.model_paths = model_path.split(':') if len(model_path) > 0 else []
        if self.model_source == 'huggingface':
            self.hub_downloader = _HuggingfaceDownloader(token=self.token)
        else:
            self.hub_downloader = _ModelscopeDownloader(token=self.token)
            if self.model_source != 'modelscope':
                lazyllm.LOG.error('Only support Huggingface and Modelscope currently. '
                                  f'Unsupported model source: {self.model_source}. Forcing use of Modelscope.')

    @staticmethod
    @functools.lru_cache
    def get_model_type(model) -> str:
        assert isinstance(model, str) and len(model) > 0, f'model name should be a non-empty string, get {model}'
        __class__._try_add_mapping(model)
        for name, info in model_name_mapping.items():
            if 'type' not in info: continue

            model_name_set = {name.casefold()}
            for source in info['source']:
                model_name_set.add(info['source'][source].split('/')[-1].casefold())

            if model.split(os.sep)[-1].casefold() in model_name_set:
                return info['type']
        return infer_model_type(model)

    @staticmethod
    @functools.lru_cache
    def _get_model_name(model) -> str:
        search_string = os.path.basename(model)
        __class__._try_add_mapping(search_string)
        for model_name, sources in model_name_mapping.items():
            if model_name.lower() == search_string.lower() or any(
                    os.path.basename(source_file).lower() == search_string.lower()
                    for source_file in sources['source'].values()):
                return model_name
        return ''

    @staticmethod
    @functools.lru_cache
    def get_model_prompt_keys(model) -> dict:
        model_name = __class__._get_model_name(model)
        __class__._try_add_mapping(model_name)
        if model_name and 'prompt_keys' in model_name_mapping[model_name.lower()]:
            return model_name_mapping[model_name.lower()]['prompt_keys']
        else:
            return dict()

    @staticmethod
    def validate_model_path(model_path):
        extensions = {'.pt', '.bin', '.safetensors'}
        for _, _, files in os.walk(model_path):
            for file in files:
                if any(file.endswith(ext) for ext in extensions):
                    return True
        return False

    @staticmethod
    def _try_add_mapping(model):
        model_base = os.path.basename(model)
        model = model_base.lower()
        if model in model_name_mapping.keys():
            return
        matched_model_prefix = next((key for key in model_provider if model.startswith(key)), None)
        if matched_model_prefix:
            matching_keys = [key for key in model_groups.keys() if key in model]
            if matching_keys:
                matched_groups = max(matching_keys, key=len)
                model_name_mapping[model] = {
                    'prompt_keys': model_groups[matched_groups]['prompt_keys'],
                    'source': {k: v + '/' + model_base for k, v in model_provider[matched_model_prefix].items()}
                }

    def download(self, model='', call_back=None):
        assert isinstance(model, str), 'model name should be a string.'
        if len(model) == 0 or model[0] in (os.sep, '.', '~') or os.path.isabs(model): return model
        if (model_at_path := self._model_exists_at_path(model)): return model_at_path
        if self.model_source == '' or self.model_source not in ('huggingface', 'modelscope'):
            lazyllm.LOG.error('model automatic downloads only support Huggingface and Modelscope currently.')
            return model

        self._try_add_mapping(model)
        if model_name_mapping.get(model.lower(), {}).get('download_by_other'): return model

        if model.lower() in model_name_mapping.keys() and \
                self.model_source in model_name_mapping[model.lower()]['source'].keys():
            full_model_dir = os.path.join(self.cache_dir, model)

            mapped_model_name = model_name_mapping[model.lower()]['source'][self.model_source]
            model_save_dir = self._do_download(mapped_model_name, call_back)
            if model_save_dir:
                # The code safely creates a symbolic link by removing any existing target.
                if os.path.exists(full_model_dir):
                    os.remove(full_model_dir)
                if os.path.islink(full_model_dir):
                    os.unlink(full_model_dir)
                os.symlink(model_save_dir, full_model_dir, target_is_directory=True)
                return full_model_dir
            return model_save_dir  # return False
        else:
            model_name_for_download = model

            if '/' not in model_name_for_download:
                # Try to figure out a possible model provider
                matched_model_prefix = next((key for key in model_provider if model.lower().startswith(key)), None)
                if matched_model_prefix and self.model_source in model_provider[matched_model_prefix]:
                    model_name_for_download = model_provider[matched_model_prefix][self.model_source] + '/' + model

            model_save_dir = self._do_download(model_name_for_download, call_back)
            return model_save_dir

    def _validate_token(self):
        return self.hub_downloader.verify_hub_token()

    def _validate_model_id(self, model_id):
        return self.hub_downloader._verify_model_id(model_id)

    def _model_exists_at_path(self, model_name):
        if len(self.model_paths) == 0:
            return None
        model_dirs = []

        # For short model name, get all possible names from the mapping.
        if model_name.lower() in model_name_mapping.keys():
            for source in ('huggingface', 'modelscope'):
                if source in model_name_mapping[model_name.lower()]['source'].keys():
                    model_dirs.append(model_name_mapping[model_name.lower()]['source'][source].replace('/', os.sep))
        model_dirs.append(model_name.replace('/', os.sep))

        for model_path in self.model_paths:
            if len(model_path) == 0: continue
            if model_path[0] != os.sep:
                lazyllm.LOG.warning(f'skipping path {model_path} as only absolute paths is accepted.')
                continue
            for model_dir in model_dirs:
                full_model_dir = os.path.join(model_path, model_dir)
                if self._is_model_valid(full_model_dir):
                    return full_model_dir
        return None

    def _is_model_valid(self, model_dir):
        if not os.path.isdir(model_dir):
            return False
        return any((True for _ in os.scandir(model_dir)))

    def _do_download(self, model='', call_back=None):
        model_dir = model.replace('/', os.sep)
        full_model_dir = os.path.join(self.cache_dir, self.model_source, model_dir)

        try:
            return self.hub_downloader.download(model, full_model_dir, call_back)
        # Use `BaseException` to capture `KeyboardInterrupt` and normal `Exceptioin`.
        except BaseException as e:  # noqa B036
            lazyllm.LOG.warning(f'Download encountered an error: {e}')
            if '401' in str(e) or 'Client Error' in str(e):
                raise RuntimeError('Authentication failed (401 Error). Please check your access token and '
                                   'permissions.  And set the token with the environment variable '
                                   'LAZYLLM_MODEL_SOURCE_TOKEN.')
            if not self.token and 'Permission denied' not in str(e):
                lazyllm.LOG.warning('Token is empty, which may prevent private models from being downloaded, '
                                    'as indicated by "the model does not exist." Please set the token with the '
                                    'environment variable LAZYLLM_MODEL_SOURCE_TOKEN to download private models.')
            raise RuntimeError(f'Model download failed for model: {model}, with error: {e}')
        return False

get_model_type(model) cached staticmethod

Source code in lazyllm\components\utils\downloader\model_downloader.py

@staticmethod
@functools.lru_cache
def get_model_type(model) -> str:
    assert isinstance(model, str) and len(model) > 0, f'model name should be a non-empty string, get {model}'
    __class__._try_add_mapping(model)
    for name, info in model_name_mapping.items():
        if 'type' not in info: continue

        model_name_set = {name.casefold()}
        for source in info['source']:
            model_name_set.add(info['source'][source].split('/')[-1].casefold())

        if model.split(os.sep)[-1].casefold() in model_name_set:
            return info['type']
    return infer_model_type(model)

get_model_prompt_keys(model) cached staticmethod

Source code in lazyllm\components\utils\downloader\model_downloader.py

@staticmethod
@functools.lru_cache
def get_model_prompt_keys(model) -> dict:
    model_name = __class__._get_model_name(model)
    __class__._try_add_mapping(model_name)
    if model_name and 'prompt_keys' in model_name_mapping[model_name.lower()]:
        return model_name_mapping[model_name.lower()]['prompt_keys']
    else:
        return dict()

validate_model_path(model_path) staticmethod

Source code in lazyllm\components\utils\downloader\model_downloader.py

@staticmethod
def validate_model_path(model_path):
    extensions = {'.pt', '.bin', '.safetensors'}
    for _, _, files in os.walk(model_path):
        for file in files:
            if any(file.endswith(ext) for ext in extensions):
                return True
    return False

download(model='', call_back=None)

Source code in lazyllm\components\utils\downloader\model_downloader.py

def download(self, model='', call_back=None):
    assert isinstance(model, str), 'model name should be a string.'
    if len(model) == 0 or model[0] in (os.sep, '.', '~') or os.path.isabs(model): return model
    if (model_at_path := self._model_exists_at_path(model)): return model_at_path
    if self.model_source == '' or self.model_source not in ('huggingface', 'modelscope'):
        lazyllm.LOG.error('model automatic downloads only support Huggingface and Modelscope currently.')
        return model

    self._try_add_mapping(model)
    if model_name_mapping.get(model.lower(), {}).get('download_by_other'): return model

    if model.lower() in model_name_mapping.keys() and \
            self.model_source in model_name_mapping[model.lower()]['source'].keys():
        full_model_dir = os.path.join(self.cache_dir, model)

        mapped_model_name = model_name_mapping[model.lower()]['source'][self.model_source]
        model_save_dir = self._do_download(mapped_model_name, call_back)
        if model_save_dir:
            # The code safely creates a symbolic link by removing any existing target.
            if os.path.exists(full_model_dir):
                os.remove(full_model_dir)
            if os.path.islink(full_model_dir):
                os.unlink(full_model_dir)
            os.symlink(model_save_dir, full_model_dir, target_is_directory=True)
            return full_model_dir
        return model_save_dir  # return False
    else:
        model_name_for_download = model

        if '/' not in model_name_for_download:
            # Try to figure out a possible model provider
            matched_model_prefix = next((key for key in model_provider if model.lower().startswith(key)), None)
            if matched_model_prefix and self.model_source in model_provider[matched_model_prefix]:
                model_name_for_download = model_provider[matched_model_prefix][self.model_source] + '/' + model

        model_save_dir = self._do_download(model_name_for_download, call_back)
        return model_save_dir

---

Formatter

lazyllm.components.formatter.LazyLLMFormatterBase

Source code in lazyllm\components\formatter\formatterbase.py

class LazyLLMFormatterBase(metaclass=LazyLLMRegisterMetaClass):
    def _load(self, msg: str):
        return msg

    def _parse_py_data_by_formatter(self, py_data):
        raise NotImplementedError('This data parse function is not implemented.')

    def format(self, msg):
        if _is_chat_message(msg):
            if msg.get('tool_calls'):
                # The base formatter collapses chat messages to plain text, which drops
                # `tool_calls` and `reasoning_content`. Tool-calling models must be
                # configured with `FunctionCallFormatter`, which keeps those structured
                # fields so a provider-compatible multi-turn history can be replayed
                # (e.g. DeepSeek rejects a tool-calling request whose previous assistant
                # turn omits `reasoning_content`). Warn rather than drop them silently.
                lazyllm.LOG.warning(
                    f'{type(self).__name__}: formatting an assistant message that '
                    'contains `tool_calls`; `tool_calls` and `reasoning_content` will be '
                    'dropped. Configure the model with `FunctionCallFormatter` for tool '
                    'calling.'
                )
            if reasoning_content := msg.get('reasoning_content'):
                msg = f'<think>{reasoning_content}</think>' + msg['content']
            else:
                msg = msg['content']
        if isinstance(msg, str): msg = self._load(msg)
        return self._parse_py_data_by_formatter(msg)

    def __call__(self, *msg):
        return self.format(msg[0] if len(msg) == 1 else package(msg))

    def __or__(self, other):
        if not isinstance(other, LazyLLMFormatterBase):
            return NotImplemented
        return PipelineFormatter(other.__ror__(self))

    def __ror__(self, f: Callable) -> Pipeline:
        if isinstance(f, Pipeline):
            if not f._capture:
                _ = Finalizer(lambda: setattr(f, '_capture', True), lambda: setattr(f, '_capture', False))
            f._add(str(uuid.uuid4().hex) if len(f._item_names) else None, self)
            return f
        return Pipeline(f, self)

lazyllm.components.formatter.formatterbase.JsonLikeFormatter

Bases: LazyLLMFormatterBase

Source code in lazyllm\components\formatter\formatterbase.py

class JsonLikeFormatter(LazyLLMFormatterBase):
    class _ListIdxes(tuple): pass
    class _DictKeys(tuple): pass

    def __init__(self, formatter: Optional[str] = None):
        if formatter and formatter.startswith('*['):
            self._return_package = True
            self._formatter = formatter.strip('*')
        else:
            self._return_package = False
            self._formatter = formatter

        if self._formatter:
            assert '*' not in self._formatter, '`*` can only be used before `[` in the beginning'
            self._formatter = self._formatter.strip().replace('{', '[{').replace('}', '}]')
            self._parse_formatter()
        else:
            self._slices = None

    def _parse_formatter(self):
        # Remove the surrounding brackets
        assert self._formatter.startswith('[') and self._formatter.endswith(']')
        slice_str = self._formatter.strip()[1:-1]
        dimensions = slice_str.split('][')
        slices = []

        for dim in dimensions:
            if '{' in dim:
                slices.append(__class__._DictKeys(d.strip() for d in dim[1:-1].split(',') if d.strip()))
            elif ':' in dim:
                assert ',' not in dim, '[a, b:c] is not supported'
                parts = dim.split(':')
                start = int(parts[0]) if _is_number(parts[0]) else None
                end = int(parts[1]) if len(parts) > 1 and _is_number(parts[1]) else None
                step = int(parts[2]) if len(parts) > 2 and _is_number(parts[2]) else None
                slices.append(slice(start, end, step))
            elif ',' in dim:
                slices.append(__class__._ListIdxes(d.strip() for d in dim.split(',') if d.strip()))
            else:
                slices.append(dim.strip())
        self._slices = slices

    def _parse_py_data_by_formatter(self, data, *, slices=None):  # noqa C901
        def _impl(data, slice):
            if isinstance(data, (tuple, list)) and isinstance(slice, str):
                return data[int(slice)]
            if isinstance(slice, __class__._ListIdxes):
                if isinstance(data, dict): return [data[k] for k in slice]
                elif isinstance(data, (tuple, list)): return type(data)(data[int(k)] for k in slice)
                else: raise RuntimeError('Only tuple/list/dict is supported for [a,b,c]')
            if isinstance(slice, __class__._DictKeys):
                assert isinstance(data, dict)
                if len(slice) == 1 and slice[0] == ':': return data
                return {k: data[k] for k in slice}
            return data[slice]

        if slices is None: slices = self._slices
        if not slices: return data
        curr_slice = slices[0]
        if isinstance(curr_slice, slice):
            if isinstance(data, dict):
                assert curr_slice.start is None and curr_slice.stop is None and curr_slice.step is None, (
                    'Only {:} and [:] is supported in dict slice')
                curr_slice = __class__._ListIdxes(data.keys())
            elif isinstance(data, (tuple, list)):
                return type(data)(self._parse_py_data_by_formatter(d, slices=slices[1:])
                                  for d in _impl(data, curr_slice))
        if isinstance(curr_slice, __class__._DictKeys):
            return {k: self._parse_py_data_by_formatter(v, slices=slices[1:])
                    for k, v in _impl(data, curr_slice).items()}
        elif isinstance(curr_slice, __class__._ListIdxes):
            tp = package if self._return_package else list if isinstance(data, dict) else type(data)
            return tp(self._parse_py_data_by_formatter(r, slices=slices[1:]) for r in _impl(data, curr_slice))
        else: return self._parse_py_data_by_formatter(_impl(data, curr_slice), slices=slices[1:])

lazyllm.components.formatter.formatterbase.PythonFormatter

Bases: JsonLikeFormatter

Source code in lazyllm\components\formatter\formatterbase.py

class PythonFormatter(JsonLikeFormatter): pass

lazyllm.components.formatter.FileFormatter

Bases: LazyLLMFormatterBase

Source code in lazyllm\components\formatter\formatterbase.py

class FileFormatter(LazyLLMFormatterBase):

    def __init__(self, formatter: str = 'decode'):
        self._mode = formatter.strip().lower()
        assert self._mode in ('decode', 'encode', 'merge')

    def _parse_py_data_by_formatter(self, py_data):
        if self._mode == 'merge':
            if isinstance(py_data, str):
                return py_data
            assert isinstance(py_data, package)
            return lazyllm_merge_query(*py_data)

        if isinstance(py_data, package):
            res = []
            for i_data in py_data:
                res.append(self._parse_py_data_by_formatter(i_data))
            return package(res)
        elif isinstance(py_data, (str, dict)):
            return self._decode_one_data(py_data)
        else:
            return py_data

    def _decode_one_data(self, py_data):
        if self._mode == 'decode':
            if isinstance(py_data, str):
                return decode_query_with_filepaths(py_data)
            else:
                return py_data
        else:
            if isinstance(py_data, dict) and 'query' in py_data and 'files' in py_data:
                return encode_query_with_filepaths(**py_data)
            else:
                return py_data

lazyllm.components.formatter.YamlFormatter

Bases: JsonLikeFormatter

Source code in lazyllm\components\formatter\yamlformatter.py

class YamlFormatter(JsonLikeFormatter):
    def _load(self, msg: str):
        try:
            return yaml.load(msg, Loader=yaml.SafeLoader)
        except Exception as e:
            lazyllm.LOG.info(f'Error: {e}')
            return ''

lazyllm.components.formatter.encode_query_with_filepaths(query=None, files=None)

Source code in lazyllm\components\formatter\formatterbase.py

def encode_query_with_filepaths(query: str = None, files: Union[str, List[str]] = None) -> str:
    query = query if query else ''
    if files:
        if isinstance(files, str): files = [files]
        assert isinstance(files, list), 'files must be a list.'
        assert all(isinstance(item, str) for item in files), 'All items in files must be strings'
        return LAZYLLM_QUERY_PREFIX + json.dumps({'query': query, 'files': files})
    else:
        return query

lazyllm.components.formatter.decode_query_with_filepaths(query_files)

Source code in lazyllm\components\formatter\formatterbase.py

def decode_query_with_filepaths(query_files: str) -> Union[dict, str]:
    assert isinstance(query_files, str), 'query_files must be a str.'
    query_files = query_files.strip()
    if query_files.startswith(LAZYLLM_QUERY_PREFIX):
        try:
            obj = json.loads(query_files[len(LAZYLLM_QUERY_PREFIX):])
            return obj
        except json.JSONDecodeError as e:
            raise ValueError(f'JSON parsing failed: {e}')
    else:
        return query_files

lazyllm.components.formatter.lazyllm_merge_query(*args)

Source code in lazyllm\components\formatter\formatterbase.py

def lazyllm_merge_query(*args: str) -> str:
    if len(args) == 1:
        return args[0]
    for item in args:
        assert isinstance(item, str), 'Merge object must be str!'
    querys = ''
    files = []
    for item in args:
        decode = decode_query_with_filepaths(item)
        if isinstance(decode, dict):
            querys += decode['query']
            files.extend(decode['files'])
        else:
            querys += decode
    return encode_query_with_filepaths(querys, files)

lazyllm.components.JsonFormatter

Bases: JsonLikeFormatter

Source code in lazyllm\components\formatter\jsonformatter.py

class JsonFormatter(JsonLikeFormatter):
    def _extract_json_from_string(self, mixed_str: str):  # noqa: C901
        json_objects = []
        brace_level = 0
        current_json = ''
        in_string = False

        for char in mixed_str:
            if char == '"' and (len(current_json) == 0 or current_json[-1] != '\\'):
                in_string = not in_string

            if not in_string:
                if char in '{[':
                    if brace_level == 0:
                        current_json = ''
                    brace_level += 1
                elif char in '}]':
                    brace_level -= 1

            if brace_level > 0 or (brace_level == 0 and char in '}]'):
                current_json += char

            if brace_level == 0 and current_json:
                try:
                    json_objects.append(json.loads(current_json))
                    current_json = ''
                except json.JSONDecodeError:
                    try:
                        repaired_obj = json_repair.loads(current_json)
                        json_objects.append(repaired_obj)
                        current_json = ''
                    except (json.JSONDecodeError, ValueError):
                        continue

        return json_objects

    def _load(self, msg: str):
        # Convert str to json format
        assert msg.count('{') == msg.count('}'), f'{msg} is not a valid json string.'
        try:
            json_objects = self._extract_json_from_string(msg)
            if len(json_objects) == 0:
                raise TypeError(f'{msg} is not a valid json string.')
            return json_objects if len(json_objects) > 1 else json_objects[0]
        except Exception as e:
            lazyllm.LOG.info(f'Error: {e}')
            return ''

lazyllm.components.EmptyFormatter

Bases: LazyLLMFormatterBase

Source code in lazyllm\components\formatter\formatterbase.py

class EmptyFormatter(LazyLLMFormatterBase):
    def _parse_py_data_by_formatter(self, msg: str):
        return msg

lazyllm.components.FunctionCallFormatter

Bases: LazyLLMFormatterBase

Source code in lazyllm\components\formatter\formatterbase.py

class FunctionCallFormatter(LazyLLMFormatterBase):
    def format(self, msg):
        assert isinstance(msg, dict), 'FunctionCallFormatter only supports dict input.'
        ret = {k: msg[k] for k in ('role', 'content', 'tool_calls', 'reasoning_content') if k in msg}
        if isinstance(ret.get('tool_calls'), list):
            ret['tool_calls'] = [{k: v for k, v in tool_call.items() if k != 'index'}
                                 for tool_call in ret['tool_calls']]
        return ret

lazyllm.components.formatter.formatterbase.PipelineFormatter

Bases: LazyLLMFormatterBase

Source code in lazyllm\components\formatter\formatterbase.py

class PipelineFormatter(LazyLLMFormatterBase):
    def __init__(self, formatter: Pipeline):
        self._formatter = formatter

    def _parse_py_data_by_formatter(self, py_data):
        return self._formatter(py_data)

    def __or__(self, other):
        if isinstance(other, LazyLLMFormatterBase):
            if isinstance(other, PipelineFormatter): other = other._formatter
            return PipelineFormatter(self._formatter | other)
        return NotImplemented

---

ComponentBase

lazyllm.components.core.ComponentBase

Bases: object

Source code in lazyllm\components\core.py

class ComponentBase(object, metaclass=LazyLLMRegisterMetaClass):
    def __init__(self, *, launcher=launchers.empty()):  # noqa B008
        self._llm_name = None
        self.job = ReadOnlyWrapper()
        if isinstance(launcher, LazyLLMLaunchersBase):
            self._launcher = launcher
        elif isinstance(launcher, type) and issubclass(launcher, LazyLLMLaunchersBase):
            self._launcher = launcher()
        else:
            raise RuntimeError('Invalid launcher given:', launcher)

    def apply():
        raise NotImplementedError('please implement function \'apply\'')

    def cmd(self, *args, **kw) -> Union[str, tuple, list]:
        raise NotImplementedError('please implement function \'cmd\'')

    @property
    def name(self): return self._llm_name
    @name.setter
    def name(self, name): self._llm_name = name

    @property
    def launcher(self): return self._launcher

    def _get_job_with_cmd(self, *args, **kw):
        cmd = self.cmd(*args, **kw)
        cmd = cmd if isinstance(cmd, LazyLLMCMD) else LazyLLMCMD(cmd)
        return self._launcher.makejob(cmd=cmd)

    def _overwrote(self, f):
        return getattr(self.__class__, f) is not getattr(__class__, f) or \
            getattr(self.__class__, '__reg_overwrite__', None) == f

    def __call__(self, *args, **kw):
        if self._overwrote('apply'):
            assert not self._overwrote('cmd'), (
                'Cannot overwrite \'cmd\' and \'apply\' in the same class')
            assert isinstance(self._launcher, launchers.Empty), 'Please use EmptyLauncher instead.'
            return self._launcher.launch(self.apply, *args, **kw)
        else:
            job = self._get_job_with_cmd(*args, **kw)
            self.job.set(job)
            return self._launcher.launch(job)

    def __repr__(self):
        return lazyllm.make_repr('lazyllm.llm.' + self.__class__._lazy_llm_group,
                                 self.__class__.__name__, name=self.name)

apply()

Source code in lazyllm\components\core.py

def apply():
    raise NotImplementedError('please implement function \'apply\'')

cmd(*args, **kw)

Source code in lazyllm\components\core.py

def cmd(self, *args, **kw) -> Union[str, tuple, list]:
    raise NotImplementedError('please implement function \'cmd\'')

lazyllm.components.deploy.ray.Distributed

Bases: LazyLLMDeployBase

Source code in lazyllm\components\deploy\ray.py

class Distributed(LazyLLMDeployBase):

    def __init__(self, launcher=launchers.remote(ngpus=1), port=None):  # noqa B008
        super().__init__(launcher=launcher)
        self.port = port or random.randint(30000, 40000)
        self.finetuned_model = None
        self.base_model = None
        self.master_ip = None

    def cmd(self, finetuned_model=None, base_model=None, master_ip=None):
        self.finetuned_model = finetuned_model
        self.base_model = base_model
        self.master_ip = master_ip
        if not self.master_ip:
            cmd = f'ray start --block --head --port={self.port} && sleep 365d'
        else:
            cmd = f'ray start --address={self.master_ip} && sleep 365d'
        return LazyLLMCMD(cmd=cmd, return_value=self.geturl)

    def geturl(self, job=None):
        time.sleep(5)
        if job is None:
            job = self.job
        if lazyllm.config['mode'] == lazyllm.Mode.Display:
            return lazyllm.package(self.finetuned_model, self.base_model, None)
        else:
            if self.master_ip:
                return lazyllm.package(self.finetuned_model, self.base_model, self.master_ip)
            return lazyllm.package(self.finetuned_model, self.base_model, f'{job.get_jobip()}:{self.port}')

cmd(finetuned_model=None, base_model=None, master_ip=None)

Source code in lazyllm\components\deploy\ray.py

def cmd(self, finetuned_model=None, base_model=None, master_ip=None):
    self.finetuned_model = finetuned_model
    self.base_model = base_model
    self.master_ip = master_ip
    if not self.master_ip:
        cmd = f'ray start --block --head --port={self.port} && sleep 365d'
    else:
        cmd = f'ray start --address={self.master_ip} && sleep 365d'
    return LazyLLMCMD(cmd=cmd, return_value=self.geturl)