中文rlhf數據集50w條數據解析
- 解析代碼
- 數據名
- 代碼解析
解析代碼
import jieba
from tqdm import tqdm
import re
import pandas as pd
import numpy as npdef find_non_english_text(text):pattern = re.compile(r'[^a-zA-Z]')return pattern.sub('', text)def find_chinese_text(text):pattern = re.compile(r'[^\u4e00-\u9fff]')return pattern.sub('', text)json_list = pd.read_parquet("E:/data_sets/train-rm-static-m2m100-zh-jianti.parquet")
data_list = []
basic_list = ["<|User|>", "<|Ash|>"] * 100for _, one_data in tqdm(json_list.iterrows()):ins = "".join(one_data["prompt"].split("\n\n")[1:-1])# inp=one_data["input"]out = one_data["response"]ins = np.hstack([i.split("Human:") for i in ins.split("Assistant:")])[1:].tolist()ins = [i for i in ins if len(i)>0]ins_len = len(ins)try:ins = np.hstack([[i, j] for i, j in zip(basic_list[:ins_len], ins)]).tolist()ins += [basic_list[ins_len], out]data_list.append(ins)except:continue
with open("data_set_five.txt", "a", encoding="utf-8") as f:voc_set = set()for one_data in tqdm(data_list):ins = []for one in one_data:if one in ["<|User|>", "<|Ash|>"]:ins += [one]else:one = jieba.lcut(one)try:one = np.hstack([list(i) if len(find_chinese_text(i)) > 0 else i for i in one]).tolist()except:breakins += onevoc_set |= set(ins)f.write(str(ins) + "\n")
數據名
在hf上搜方可
train-rm-static-m2m100-zh-jianti.parquet
數據地址
代碼解析
import jieba
: 導入分詞庫jieba。from tqdm import tqdm
: 導入tqdm庫,用于在循環中顯示進度條。import re
: 導入正則表達式庫。import pandas as pd
: 導入pandas庫,用于處理數據。import numpy as np
: 導入numpy庫,用于處理數組。def find_non_english_text(text)
: 定義一個函數,用于查找非英文文本。該函數輸入一個文本,使用正則表達式找到文本中的非英文字符,并用空字符替換。def find_chinese_text(text)
: 定義一個函數,用于查找中文文本。該函數輸入一個文本,使用正則表達式找到文本中的非中文字符,并用空字符替換。json_list = pd.read_parquet("E:/data_sets/train-rm-static-m2m100-zh-jianti.parquet")
: 讀取parquet格式的文件,并將數據賦值給json_list。data_list = []
: 定義一個空列表,用于存儲最終的數據。basic_list = ["<|User|>", "<|Ash|>"] * 100
: 定義一個列表,其中包含兩個字符串"<|User|>“和”<|Ash|>",并將其重復100次。for _, one_data in tqdm(json_list.iterrows()):
: 遍歷json_list,并使用tqdm在循環中顯示進度條。其中,_表示索引值,one_data表示每一行數據。ins = "".join(one_data["prompt"].split("\n\n")[1:-1])
: 從one_data中獲取prompt字段,并去除字段中的換行符。然后,取出prompt中除第一行和最后一行之外的所有內容,并將其賦值給ins。out = one_data["response"]
: 從one_data中獲取response字段,并將其賦值給out。ins = np.hstack([i.split("Human:") for i in ins.split("Assistant:")])[1:].tolist()
: 將ins根據"Assistant:"分割成多個子串,然后將子串中的"Human:"去除,并將子串合并為一個列表。最后,將列表中的第一個元素去除,并將其轉換為列表類型。ins_len = len(ins)
: 獲取ins列表的長度,并將其賦值給ins_len。ins = np.hstack([[i, j] for i, j in zip(basic_list[:ins_len], ins)]).tolist()
: 將basic_list中與ins長度相同的元素取出,然后將basic_list和ins中的元素一一配對組成元素為兩個字符串的列表,并將這些列表合并為一個列表。最后,將這個列表轉換為列表類型。data_list.append(ins)
: 將ins添加到data_list中。with open("data_set_five.txt", "a", encoding="utf-8") as f:
: 打開一個文件"data_set_five.txt",并將其賦值給變量f。打開文件時,使用"a"模式表示追加數據到文件末尾。voc_set = set()
: 定義一個空集合,用于存儲數據中的所有詞匯。for one_data in tqdm(data_list):
: 遍歷data_list,并使用tqdm在循環中顯示進度條。其中,one_data表示每一個元素。ins = []
: 定義一個空列表,用于存儲分詞后的文本。for one in one_data:
: 遍歷one_data中的每一個元素,并將其賦值給變量one。if one in ["<|User|>", "<|Ash|>"]:
: 判斷one是否等于"<|User|>“或”<|Ash|>"。ins += [one]
: 如果one等于"<|User|>“或”<|Ash|>",則將one加入ins中。else:
: 如果one不等于"<|User|>“或”<|Ash|>"。one = jieba.lcut(one)
: 使用jieba對one進行分詞。try:
: 開始異常處理。one = np.hstack([list(i) if len(find_chinese_text(i)) > 0 else i for i in one]).tolist()
: 對于one中的每個分詞,如果其包含中文字符,則將其拆分成單個字符,否則不做處理。然后,將處理后的結果合并為一個列表,并將這個列表轉換為列表類型。voc_set |= set(ins)
: 將ins中的所有元素加入voc_set中。f.write(str(ins) + "\n")
: 將ins轉換為字符串,并將其寫入文件中,同時在字符串末尾加入換行符。