赛题连接

https://challenge.xfyun.cn/topic/info?type=role-element-extraction&option=phb

Datawhale Al夏令营 零基础入门大模型技术竞赛

数据集预处理

由于赛题官方限定使用了星火大模型，所以只能调用星火大模型的API或者使用零代码微调

首先训练数据很少是有129条，其中只有<code>chat_text和infos两个属性，chat_text是聊天文本，infos就是提取的信息也是训练集标签，他的平均长度有6000左右对于星火对于信息提取任务已经很长了，而且最长的将近30000，如果使用星火大模型进行询问肯定是要被截断的，而且微调上传的数据也是有最大长度的，我门需要对数据进行处理。

数据简单清洗

简单的导包

<code>from dataclasses import dataclass

from sparkai.llm.llm import ChatSparkLLM, ChunkPrintHandler

from sparkai.core.messages import ChatMessage

import pandas as pd

import os

import json

import re

import matplotlib.pyplot as plt

from tqdm import tqdm

from math import ceil

import numpy as np

from copy import deepcopy

import random

tqdm.pandas()

plt.rcParams['font.family'] = ['STFangsong']

plt.rcParams['axes.unicode_minus'] = False

加载数据

data_dir = "./data"

train_file = "train.json"

test_file = "test_data.json"

train_data = pd.read_json(os.path.join(data_dir, train_file))

test_data = pd.read_json(os.path.join(data_dir, test_file))

首先我们发现数据集中有许多[图片]和超链接，这些对数据提取作用不大，我们可以将其去掉，

# 删除表情图片、超链接

train_data['chat_text'] = train_data['chat_text'].str.replace(r"\[[^\[\]]{2,10}\]", "", regex=True)

train_data['chat_text'] = train_data['chat_text'].str.replace("https?://\S+", "", regex=True)

test_data['chat_text'] = test_data['chat_text'].str.replace(r"\[[^\[\]]{2,10}\]", "", regex=True)

test_data['chat_text'] = test_data['chat_text'].str.replace("https?://\S+", "", regex=True)

对于一个人连续的对话我们可以哦将其合并成一个对话

def get_names_phones_and_emails(example):

names = re.findall(r"(?:\n)?([\u4e00-\u9fa5]+\d+)：", example["chat_text"])

names += re.findall(r"@([\u4e00-\u9fa5]+)\s", example["chat_text"])

emails = re.findall(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}", example["chat_text"])

# phones = re.findall(r"1[356789]\d{9}", example["chat_text"]) # 文本中的手机号并不是标准手机号

phones = re.findall(r"\d{3}\s*\d{4}\s*\d{4}", example["chat_text"])

return pd.Series([set(names), set(phones), set(emails)], index=['names', 'phones', 'emails'])

def merge_chat(example):

for name in example['names']:

example["chat_text"] = example["chat_text"].replace(f"\n{ name}：", f"<|sep|>{ name}：")

chats = example["chat_text"].split("<|sep|>")

last_name = "UNKNOWN"

new_chats = []

for chat in chats:

if chat.startswith(last_name):

chat = chat.strip("\n")

chat = "".join(chat.split("：")[1:])

new_chats[-1] += " " + chat

else:

new_chats.append(chat)

last_name = chat.split("：")[0]

return pd.Series(["\n".join(new_chats), new_chats], index=["chats", "chat_list"])

# 使用正则表达式获得'names', 'phones', 'emails'

train_data[['names', 'phones', 'emails']] = train_data.apply(get_names_phones_and_emails, axis=1)

test_data[['names', 'phones', 'emails']] = test_data.apply(get_names_phones_and_emails, axis=1)

# 分割聊天记录， 合并连续相同人的聊天

train_data[["chats", "chat_list"]] = train_data.apply(merge_chat, axis=1)

test_data[["chats", "chat_list"]] = test_data.apply(merge_chat, axis=1)

补充

补充：后面我们发现数据中<code>chat_text中有许多是重复多编的，我们需要把重复的也给去除掉，这样处理后的数据就会大大减小，使用暴力匹配去除

<code>def process(excemple):

chat_list = excemple["chat_text"].split("\n")

res = []

s = 0

while s < len(chat_list):

i, j = s, s+1

start_j = j

while i < len(chat_list) and j < len(chat_list):

if chat_list[i] == chat_list[j]:

i += 1

else:

if i != s:

if j - start_j >10:

res += list(range(start_j, j))

i = s

start_j = j

j += 1

s += 1

texts = []

for i in range(len(chat_list)):

if i not in res:

texts.append(chat_list[i])

return "\n".join(texts)

train_data["chat_text"] = train_data.apply(process, axis = 1)

test_data["chat_text"] = test_data.apply(process, axis = 1)

构造训练集

处理之后其实有些还是很长，我们可以有两种简单粗暴的方法

截断分块

对于构造训练数据，我们使用了第一种截断的方法，但这两种方法都有一定的缺点

我们需要查看讯飞官方微调需要的训练集格式，这里我选择使用JSONL格式，并且其每一行是一个JSON字符串,格式为

{ "input":"", "target":""}

训练时我选用了讯飞的<code>spark pro进行训练，其要求训练数据不少于1500条，每一个input+target长度不能大于8000

def process(x):

# 提示词，我们交代清楚大模型的角色、目标、注意事项，然后提供背景信息，输出格式就可以了

prompt = f"""Instruction:

你是一个信息要素提取工作人员，你需要从给定的`ChatText`中提取出**客户**的`Infos`中相关信息，将提取的信息填到`Infos`中，

注意事项：

1. 没有的信息无需填写

2. 保持`Infos`的JSON格式不变，没有的信息项也要保留！！！

4. 姓名可以是聊天昵称

5. 注意是客户的信息，不是客服的信息

6. 可以有多个客户信息

ChatText:

{ x["chat_text"]}

"""

# 要求的输出格式

infos = """"

Infos:

infos": [{

"基本信息-姓名": "",

"基本信息-手机号码": "",

"基本信息-邮箱": "",

"基本信息-地区": "",

"基本信息-详细地址": "",

"基本信息-性别": "",

"基本信息-年龄": "",

"基本信息-生日": "",

"咨询类型": [],

"意向产品": [],

"购买异议点": [],

"客户预算-预算是否充足": "",

"客户预算-总体预算金额": "",

"客户预算-预算明细": "",

"竞品信息": "",

"客户是否有意向": "",

"客户是否有卡点": "",

"客户购买阶段": "",

"下一步跟进计划-参与人": [],

"下一步跟进计划-时间点": "",

"下一步跟进计划-具体事项": ""

}]

"""

# prompt+infos是文件中的input，answer是文件中的target

answer = f"""{ x["infos"]}""" #target

total= len(prompt + infos + answer)

if total > 8000:

prompt = prompt[:8000-len(infos + answer)]

return pd.Series([prompt, answer], index=["input", "target"])

data = train_data.apply(process, axis=1)

# 测试集中的target并没有用可以忽略

data = test_data.apply(process, axis=1)

#保存数据

with open(os.path.join(data_dir, "my_train.jsonl"), "w", encoding="utf-8") as f:code>

f.write("\n".join([json.dumps(i, ensure_ascii=False) for i in list(data.transpose().to_dict().values())]))

f.close()

with open(os.path.join(data_dir, "my_test.jsonl"), "w", encoding="utf-8") as f:code>

f.write("\n".join([json.dumps(i, ensure_ascii=False) for i in list(data.transpose().to_dict().values())]))

f.close()

对于训练数据不少于1500条的要求，我直接将训练集进行了多次复制，只要不少于1500条就可以训练。训练我只训练了两轮。

使用官方零代码微调

<code>from sparkai.llm.llm import ChatSparkLLM, ChunkPrintHandler

from sparkai.core.messages import ChatMessage

import pandas as pd

import os

from tqdm import tqdm

import json

spark = ChatSparkLLM(

spark_api_url="wss://spark-api-n.xf-yun.com/v3.1/chat",#spark pro微调的urlcode>

spark_app_id="",code>

spark_api_key="",code>

spark_api_secret="",code>

spark_llm_domain="patchv3", #spark pro微调的版本code>

streaming=False,

)

def save_result(data):

with open("./data/result1.json", "w") as f:

file = data.to_json(orient='records', index=False, force_ascii=False)code>

f.write(file)

f.close()

for j in range(0, 10):

res = []

for i in tqdm(range(len(data)), desc=f"正在询问第{ j}轮"):

messages = [ChatMessage(

role="user",code>

content=data.iloc[i]["input"]

)]

while True:

try:

handler = ChunkPrintHandler()

a = spark.generate([messages], callbacks=[handler])

a = json.loads(a.generations[0][0].text.replace("'", "\""))

except:

print("出错了")

continue

res.append(a)

break

multi_res.append(res)

test_data[f"infos_{ j}"] = res

save_result(test_data)

多轮投票

from typing import Counter, defaultdict

template_infos = {

"基本信息-姓名": "",

"基本信息-手机号码": "",

"基本信息-邮箱": "",

"基本信息-地区": "",

"基本信息-详细地址": "",

"基本信息-性别": "",

"基本信息-年龄": "",

"基本信息-生日": "",

"咨询类型": [],

"意向产品": [],

"购买异议点": [],

"客户预算-预算是否充足": "",

"客户预算-总体预算金额": "",

"客户预算-预算明细": "",

"竞品信息": "",

"客户是否有意向": "",

"客户是否有卡点": "",

"客户购买阶段": "",

"下一步跟进计划-参与人": [],

"下一步跟进计划-时间点": "",

"下一步跟进计划-具体事项": ""

}

result_Infos = []

## 这里的代码已经不是我最初始的代码了，可能会影响到效果，最初我是不管有结果个用户，只投出一个用户，其他信息也是直接全部投票，没有使用根据'基本信息-姓名'进行分开投票，可以自行尝试，投票还是可以提升一点分数的

for multi_infos in zip(*multi_res):

names_info_dict = defaultdict(list)

for infos in multi_infos:

for info in infos:

names_info_dict[info['基本信息-姓名']].append(info)

res_infos = []

for name in names_info_dict:

l = len(names_info_dict[name])

print(l)

if l < 5:

continue

infos = template_infos.copy()

for attr in template_infos:

if isinstance(template_infos[attr], str):

val_freq = Counter([multi_info.get(attr, "") for multi_info in names_info_dict[name]])

top_2 = val_freq.most_common(2)

if len(top_2) == 1:

val = top_2[0][0]

else:

if top_2[0][0] == "" and top_2[1][1] < l/2:

val = ""

elif top_2[0][0] == "":

val = top_2[1][0]

else:

val = top_2[0][0]

else:

val_freq = []

for multi_info in names_info_dict[name]:

val_freq.extend((multi_info.get(attr, [])))

val_freq = Counter(val_freq)

val =[val for val, freq in val_freq.most_common(10) if freq > l/2]

infos[attr] = val

res_infos.append(infos)

# if len(res_infos) >= 2:

# print(len(names_info_dict[name]),res_infos)

result_Infos.append(res_infos)

test_data["infos"] = result_Infos

save_result(test_data[["chat_text", "infos"]])

总结