使用任务式建模精调自定义大模型

!pip install modelscope

from modelscope import snapshot_download
#qwen/Qwen2-0.5b-Instruct为需要下载的模型名称，cache_dir为下载模型保存的地址，这里'./'表示将下载模型保存在CFS挂载目录的根目录中
model_dir = snapshot_download('qwen/Qwen2-0.5b-Instruct', cache_dir='./')


git clone -b v0.9.2 https://github.com/hiyouga/LLaMA-Factory.git

scp -r -P <port> Qwen2-0.5B-Instruct root@host:/home/tione/notebook/workspace


# 预训练模型路径
model_name_or_path: /opt/ml/pretrain_model

# 训练数据
dataset：identity
template：qwen

# 训练输出路径
output_dir: /llm-private/output

# 训练轮数（为了快速验证可以适当调小）
num_train_epochs: 3.0

#!/bin/bash
TRAIN_CONFIG=$1
DISTRIBUTED_ARGS="
    --nproc_per_node ${NPROC_PER_NODE:-$(nvidia-smi -L | wc -l)} \\
    --nnodes ${WORLD_SIZE:-1} \\
    --node_rank ${RANK:-0} \\
    --master_addr ${MASTER_ADDR:-127.0.0.1} \\
    --master_port ${MASTER_PORT:-23456}
    "

set -ex
torchrun $DISTRIBUTED_ARGS src/train.py $TRAIN_CONFIG

cd /opt/ml/code
pip3 install -r requirements.txt -i https://mirrors.tencentyun.com/pypi/simple
bash start.sh examples/train_full/qwen2_full_sft.yaml

# -*- coding: UTF-8 -*-
import time
import requests
import argparse
import json

from concurrent.futures import ThreadPoolExecutor, as_completed


def generate(prompt, log_stream=False):
    start = time.time()
    header = {
        "content-type": "application/json",
    }

    # print(prompt,type(prompt))
    data = {
        "messages": [{"role":"user","content":prompt}],        "temperature": 0.7,
        "max_tokens": 256,
    }
    result = requests.post(f"{args.server}/v1/chat/completions", 
        headers=header, 
        json=data, 
        stream=True    )
    response = ""
    for part in result.iter_lines():
        if part:
            if "content" in part.decode("utf-8"):
                print(part.decode("utf-8"))
                content = json.loads(part.decode("utf-8"))["choices"][0]["message"]["content"]  # 字符串过滤 data: 之后转成 json 格式再取文本
                response += content
                if log_stream:
                    print(content, end = "", flush = True)
    if log_stream:
        print()
    end = time.time()
    return {"prompt": prompt, "generated_text": response, "time_cost": end - start}


def main():    
    # 准备测试集
    prompts = []
    with open(args.test_data, "r", encoding="utf8") as data:
        for line in data.readlines():
            line_dict = json.loads(line.strip())
            prompt = line_dict['question']
            prompts.append(prompt)
    # 或者使用几个简单的测试样例：
    prompts = ["你好！","你是谁？"]
    task_list = []
    t1 = time.time()    
    with ThreadPoolExecutor(max_workers=args.concurrency) as executor:
        for prompt in prompts:            
            task_list.append(executor.submit(generate, prompt, args.verbose))
        for item in as_completed(task_list):
            result = item.result()
            print(result)
    total_time = time.time() - t1
    req_per_sec = len(prompts) / total_time
    print(f"total_time={total_time:.4f}s, requests/s={req_per_sec:.4f}")
            
if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    # 调用地址换为在线服务的地址
    parser.add_argument("-s", "--server", type=str, default="https://ms-xxxxxxxx-yyyyyy.cs.com/ms-xxxxxxx", help="推理服务地址")
    parser.add_argument("-d", "--test-data", type=str, default="./test.jsonl", help="测试jsonl文件")
    parser.add_argument("-c", "--concurrency", type=int, default=4, help="请求并发数")
    parser.add_argument("-v", "--verbose", action="store_true", help="打印流式详情")
    args = parser.parse_args()
    main()


{'prompt': '你是谁？', 'generated_text': '您好，我是 xiatian，一个由 tencent 开发的人工智能助手，我可以帮人们解决各种语言相关的问题和任务。', 'time_cost': 1.1000895500183105}
total_time=1.1006s, requests/s=0.9086