基于星火大模型的群聊对话分角色要素提取挑战赛
赛事任务：从给定的<客服>与<客户>的群聊对话中, 提取出指定的字段信息。
其实就是企业收集了大量的客服与客户之间的对话记录，这些对话记录属于非结构化数据，而企业想通过大模型技术，将它们变为结构化数据，以用于后续的数据治理和数据挖掘，为企业带来更多价值。
输入:客服对话记录
输出:结构化json数据
下载相关库
!pip install --upgrade -q spark_ai_python
配置导入
from sparkai.llm.llm import ChatSparkLLM, ChunkPrintHandler
from sparkai.core.messages import ChatMessage
import json#星火认知大模型Spark3.5 Max的URL值，其他版本大模型URL值请前往文档（https://www.xfyun.cn/doc/spark/Web.html）查看
SPARKAI_URL = 'wss://spark-api.xf-yun.com/v3.5/chat'
#星火认知大模型调用秘钥信息，请前往讯飞开放平台控制台（https://console.xfyun.cn/services/bm35）查看
SPARKAI_APP_ID = ''
SPARKAI_API_SECRET = ''
SPARKAI_API_KEY = ''
#星火认知大模型Spark3.5 Max的domain值，其他版本大模型domain值请前往文档（https://www.xfyun.cn/doc/spark/Web.html）查看
SPARKAI_DOMAIN = 'generalv3.5'
模型测试
def get_completions(text):messages = [ChatMessage(role="user",content=text)]spark = ChatSparkLLM(spark_api_url=SPARKAI_URL,spark_app_id=SPARKAI_APP_ID,spark_api_key=SPARKAI_API_KEY,spark_api_secret=SPARKAI_API_SECRET,spark_llm_domain=SPARKAI_DOMAIN,streaming=False,)handler = ChunkPrintHandler()a = spark.generate([messages], callbacks=[handler])return a.generations[0][0].text# 测试模型配置是否正确
text = "你好"
get_completions(text)
#输出
'你好！有什么我可以帮助你的吗？'
#测试成功
数据读取
def read_json(json_file_path):"""读取json文件"""with open(json_file_path, 'r') as f:data = json.load(f)return datadef write_json(json_file_path, data):"""写入json文件"""with open(json_file_path, 'w') as f:json.dump(data, f, ensure_ascii=False, indent=4)# 读取数据
train_data = read_json("dataset/train.json")
test_data = read_json("dataset/test_data.json")
我们来看一看数据集里都有什么。
print(train_data[10]['chat_text'])
dictionary=train_data[10]['infos'][0]
dictionary
#输出
{'基本信息-姓名': '赵芳7','基本信息-手机号码': '14121650240','基本信息-邮箱': '','基本信息-地区': '河南省马鞍山市','基本信息-详细地址': '','基本信息-性别': '','基本信息-年龄': '','基本信息-生日': '','咨询类型': ['询价', '答疑'],'意向产品': ['高级版'],'购买异议点': ['产品功能'],'客户预算-预算是否充足': '','客户预算-总体预算金额': '','客户预算-预算明细': '','竞品信息': '','客户是否有意向': '无意向','客户是否有卡点': '有卡点','客户购买阶段': '项目搁置','下一步跟进计划-参与人': [],'下一步跟进计划-时间点': '','下一步跟进计划-具体事项': ''}
Prompt设计
prompt 设计
PROMPT_EXTRACT = “”"
你将获得一段群聊对话记录。你的任务是根据给定的表单格式从对话记录中提取结构化信息。在提取信息时，请确保它与类型信息完全匹配，不要添加任何没有出现在下面模式中的属性。
表单格式如下：
info: Array<Dict(
“基本信息-姓名”: string | “”, // 客户的姓名。
“基本信息-手机号码”: string | “”, // 客户的手机号码。
“基本信息-邮箱”: string | “”, // 客户的电子邮箱地址。
“基本信息-地区”: string | “”, // 客户所在的地区或城市。
“基本信息-详细地址”: string | “”, // 客户的详细地址。
“基本信息-性别”: string | “”, // 客户的性别。
“基本信息-年龄”: string | “”, // 客户的年龄。
“基本信息-生日”: string | “”, // 客户的生日。
“咨询类型”: string[] | [], // 客户的咨询类型，如询价、答疑等。
“意向产品”: string[] | [], // 客户感兴趣的产品。
“购买异议点”: string[] | [], // 客户在购买过程中提出的异议或问题。
“客户预算-预算是否充足”: string | “”, // 客户的预算是否充足。示例：充足, 不充足
“客户预算-总体预算金额”: string | “”, // 客户的总体预算金额。
“客户预算-预算明细”: string | “”, // 客户预算的具体明细。
“竞品信息”: string | “”, // 竞争对手的信息。
“客户是否有意向”: string | “”, // 客户是否有购买意向。示例：有意向, 无意向
“客户是否有卡点”: string | “”, // 客户在购买过程中是否遇到阻碍或卡点。示例：有卡点, 无卡点
“客户购买阶段”: string | “”, // 客户当前的购买阶段，如合同中、方案交流等。
“下一步跟进计划-参与人”: string[] | [], // 下一步跟进计划中涉及的人员（客服人员）。
“下一步跟进计划-时间点”: string | “”, // 下一步跟进的时间点。
“下一步跟进计划-具体事项”: string | “” // 下一步需要进行的具体事项。
)>
请分析以下群聊对话记录，并根据上述格式提取信息：
对话记录：
{content}
请将提取的信息以JSON格式输出。
不要添加任何澄清信息。
输出必须遵循上面的模式。
不要添加任何没有出现在模式中的附加字段。
不要随意删除字段。
输出：
[{{"基本信息-姓名": "姓名","基本信息-手机号码": "手机号码","基本信息-邮箱": "邮箱","基本信息-地区": "地区","基本信息-详细地址": "详细地址","基本信息-性别": "性别","基本信息-年龄": "年龄","基本信息-生日": "生日","咨询类型": ["咨询类型"],"意向产品": ["意向产品"],"购买异议点": ["购买异议点"],"客户预算-预算是否充足": "充足或不充足","客户预算-总体预算金额": "总体预算金额","客户预算-预算明细": "预算明细","竞品信息": "竞品信息","客户是否有意向": "有意向或无意向","客户是否有卡点": "有卡点或无卡点","客户购买阶段": "购买阶段","下一步跟进计划-参与人": ["跟进计划参与人"],"下一步跟进计划-时间点": "跟进计划时间点","下一步跟进计划-具体事项": "跟进计划具体事项"
}}, ...]
“”"
这块有点格式错误，详情请看baseline1。
我们先来将单独的一条数据交给大模型处理。
res=get_completions(PROMPT_EXTRACT.format(content=test_data[10]['chat_text']))
res
#输出
'```json\n[{\n "基本信息-姓名": "王勇7",\n "基本信息-手机号码": "19797589218",\n "咨询类型": ["企业微信认证费用及年审信息"],\n "意向产品": ["企业微信服务"],\n "购买异议点": ["报价中不包含企业微信数据迁移，需额外联系企微处理"],\n "竞品信息": "无明确提及",\n "客户是否有意向": "有意向",\n "下一步跟进计划-参与人": ["王勇7", "吴军6"],\n "下一步跟进计划-时间点": "待定，需再确认时间",\n "下一步跟进计划-具体事项": "讨论自来水项目，报备项目与腾讯，可能的会面安排"\n}, {\n "基本信息-姓名": "吴军6",\n "基本信息-手机号码": "18500132989",\n "咨询类型": ["企业微信运维费用", "企业微信数据迁移需求"],\n "意向产品": ["企业微信服务"],\n "购买异议点": ["报价中未包括数据迁移，需要额外处理"],\n "竞品信息": "无明确提及",\n "客户是否有意向": "有意向",\n "下一步跟进计划-参与人": ["吴军6", "王勇7"],\n "下一步跟进计划-时间点": "待定，因双方日程紧张可能需要重新安排",\n "下一步跟进计划-具体事项": "会面讨论合作详情，解决企业微信数据迁移问题"\n}]\n```'
我们可以看到，大模型成功地输出了我们想要的内容。
但是有一些问题需要注意，比如大模型输出的json格式还是比较混乱。
因此，我们需要一个格式检查和纠正的函数来处理大模型的输出。
格式检查和处理
class JsonFormatError(Exception):'''用于处理 JSON 格式错误相关的异常情况如果JSON 数据的格式正确，那么它不会被调用如果 JSON 数据的格式不正确，例如不符合 JSON 标准的字符串，解析过程会失败然后 parse_json 函数会抛出 JsonFormatError 异常，其中的错误消息可以指示出 JSON 格式错误的具体原因'''def __init__(self, message):self.message = messagesuper().__init__(self.message)
def convert_all_json_in_text_to_dict(text):"""提取LLM输出文本中的json字符串此函数接收的参数是包含json字符串的文本"""dicts,stack=[],[]#dicts是一个空列表，用于存储解析后的json字典#stack是一个空列表，用于辅助在文本中定位json字符串的起始和结束位置for i in range(len(text)):if text[i]=='{':stack.append(i) #如果当前字符是'{'，说明找到了JSON 字符串的开始，将其索引i放入stack中elif text[i]=='}':begin=stack.pop() #如果当前字符是'}'，说明找到了JSON字符串的结束，此时从stack中弹出上一个'{'的索引，存入beginif not stack:#如果stack现在为空(即没有未匹配的'{')#则说明begin到当前索引i+1之间的部分是一个完整的JSON字符串#使用json.loads()解析成字典，并将其添加到dicts列表中dicts.append(json.loads(text[begin:i+1]))return dicts
测试一下
convert_all_json_in_text_to_dict(res)
#输出
[{'基本信息-姓名': '王勇7','基本信息-手机号码': '19797589218','咨询类型': ['企业微信认证费用及年审信息'],'意向产品': ['企业微信服务'],'购买异议点': ['报价中不包含企业微信数据迁移，需额外联系企微处理'],'竞品信息': '无明确提及','客户是否有意向': '有意向','下一步跟进计划-参与人': ['王勇7', '吴军6'],'下一步跟进计划-时间点': '待定，需再确认时间','下一步跟进计划-具体事项': '讨论自来水项目，报备项目与腾讯，可能的会面安排'},{'基本信息-姓名': '吴军6','基本信息-手机号码': '18500132989','咨询类型': ['企业微信运维费用', '企业微信数据迁移需求'],'意向产品': ['企业微信服务'],'购买异议点': ['报价中未包括数据迁移，需要额外处理'],'竞品信息': '无明确提及','客户是否有意向': '有意向','下一步跟进计划-参与人': ['吴军6', '王勇7'],'下一步跟进计划-时间点': '待定，因双方日程紧张可能需要重新安排','下一步跟进计划-具体事项': '会面讨论合作详情，解决企业微信数据迁移问题'}]
def check_and_complete_json_format(data):'''这段代码的主要目的是确保输入的data符合预期的JSON数据结构如果有缺失或类型不匹配的情况，会进行相应的处理和异常抛出'''required_keys = {"基本信息-姓名": str,"基本信息-手机号码": str,"基本信息-邮箱": str,"基本信息-地区": str,"基本信息-详细地址": str,"基本信息-性别": str,"基本信息-年龄": str,"基本信息-生日": str,"咨询类型": list,"意向产品": list,"购买异议点": list,"客户预算-预算是否充足": str,"客户预算-总体预算金额": str,"客户预算-预算明细": str,"竞品信息": str,"客户是否有意向": str,"客户是否有卡点": str,"客户购买阶段": str,"下一步跟进计划-参与人": list,"下一步跟进计划-时间点": str,"下一步跟进计划-具体事项": str}#这个字典定义了所有必需的键及其对应的值类型。例如，"基本信息-姓名"应该是一个字符串，"咨询类型"应该是一个列表，等等if not isinstance(data, list):#首先检查输入的data是否是一个列表，如果不是，则抛出JsonFormatError异常。raise JsonFormatError("Data is not a list")for item in data:'''对于每个item(字典)在data(列表中):- 检查是否每个必需的键都存在，如果不存在，则根据值类型value_type添加一个空列表或空字符串- 验证每个键的值类型是否正确- 如果值类型是列表，并且列表中的元素不全是字符串，则抛出JsonFormatError异常'''if not isinstance(item, dict):raise JsonFormatError("Item is not a dictionary")for key, value_type in required_keys.items():if key not in item:item[key] = [] if value_type == list else ""if not isinstance(item[key], value_type):raise JsonFormatError(f"Key '{key}' is not of type {value_type.__name__}")if value_type == list and not all(isinstance(i, str) for i in item[key]):raise JsonFormatError(f"Key '{key}' does not contain all strings in the list")return data
我们还是以单条test_data的数据为例。
res=get_completions(PROMPT_EXTRACT.format(content=test_data[10]['chat_text']))
infos=convert_all_json_in_text_to_dict(res) #将文本字符串转化为字典
infos=check_and_complete_json_format(infos)infos
#输出
[{'基本信息-姓名': '王勇7','基本信息-手机号码': '19797589218','基本信息-邮箱': '','基本信息-地区': '浙江省马鞍山市','基本信息-详细地址': '','基本信息-性别': '','基本信息-年龄': '','基本信息-生日': '','咨询类型': ['企业微信认证审核费用', '年审的作用和费用'],'意向产品': ['企业微信认证'],'购买异议点': ['数据迁移不包含在报价内'],'客户预算-预算是否充足': '','客户预算-总体预算金额': '','客户预算-预算明细': '','竞品信息': '','客户是否有意向': '有意向','客户是否有卡点': '有卡点','客户购买阶段': '考虑中','下一步跟进计划-参与人': ['王勇7'],'下一步跟进计划-时间点': '未确定','下一步跟进计划-具体事项': '讨论自来水项目，解决数据迁移问题'},{'基本信息-姓名': '吴军6','基本信息-手机号码': '18500132989','基本信息-邮箱': '','基本信息-地区': '亚金','基本信息-详细地址': '','基本信息-性别': '','基本信息-年龄': '','基本信息-生日': '','咨询类型': ['企业微信认证审核费用', '年审的作用和费用'],'意向产品': ['企业微信认证'],'购买异议点': ['数据迁移不包含在报价内'],'客户预算-预算是否充足': '','客户预算-总体预算金额': '','客户预算-预算明细': '','竞品信息': '','客户是否有意向': '有意向','客户是否有卡点': '有卡点','客户购买阶段': '考虑中','下一步跟进计划-参与人': ['王勇7'],'下一步跟进计划-时间点': '未确定','下一步跟进计划-具体事项': '讨论自来水项目，解决数据迁移问题'}]
针对单条数据的处理差不多就是这样了。
循环遍历
我们接下来通过循环遍历每一条tast_data数据。
from tqdm import tqdmretry_count = 5 # 重试次数
result = [] # 存储成功获取数据的结果
error_data = [] # 存储失败的数据，以待后续处理for index, data in tqdm(enumerate(test_data)):index += 1is_success = False #一个flag标志，用于记录是否成功for i in range(retry_count):try:res = get_completions(PROMPT_EXTRACT.format(content=data["chat_text"]))infos = convert_all_json_in_text_to_dict(res)infos = check_and_complete_json_format(infos)result.append({"infos": infos,"index": index})is_success = True #如果成功获取数据，则将结果存储在result列表中的字典中，同时标记is_success为True并跳出循环breakexcept Exception as e:print("index:", index, ", error:", e)continueif not is_success:'''如果无法成功获取数据(即is_success为False)将当前data添加到error_data列表中，并在其字典中添加一个index键来记录该数据在原始列表中的位置'''data["index"] = indexerror_data.append(data)
故障数据处理
# 处理之前剩下的失败数据if error_data:retry_count = 10 # 重试次数error_data_temp = [] # 临时存储失败数据while True:# 循环处理失败数据，直到全部成功if error_data_temp:# 如果error_data_temp非空，将它赋值给error_data，将error_data_temp置为空# 后续处理的是error_dataerror_data = error_data_temperror_data_temp = []for data in tqdm(error_data):is_success = Falsefor i in range(retry_count):try:res = get_completions(PROMPT_EXTRACT.format(content=data["chat_text"]))infos = convert_all_json_in_text_to_dict(res)infos = check_and_complete_json_format(infos)result.append({"infos": infos,"index": data["index"]})is_success = Truebreakexcept Exception as e:print("index:", index, ", error:", e)continueif not is_success:error_data_temp.append(data)if not error_data_temp:# 如果error_data_temp为空了，则用break跳出循环break#当所有失败数据处理完毕后，将处理成功的result列表按照index键进行排序result = sorted(result, key=lambda x: x["index"])
生成提交文件
# 保存输出
write_json("output.json", result)
提交结果
下载output.json文件，去大赛官方提交结果。
