大模型微調實戰之基于星火大模型的群聊對話分角色要素提取挑戰賽：Task01：跑通Baseline

0 背景

Datawhale AI夏令營第二期開始啦，去年有幸參與過第一期，收獲很多，這次也立馬參與了第二期，這一期主要是關于大模型微調實戰的，之前一直想接觸大模型，但是忙于畢業一直沒有行動，抓住這次機會行動起來！

在當今數字化時代，企業積累了豐富的對話數據，這些數據不僅是客戶與企業之間交流的記錄，更是隱藏著寶貴信息的寶庫。在這個背景下，群聊對話分角色要素提取成為了企業營銷和服務的一項重要策略。
群聊對話分角色要素提取的理念是基于企業對話數據的深度分析和挖掘。通過對群聊對話數據進行分析，企業可以更好地理解客戶的需求、興趣和行為模式，從而精準地把握客戶的需求和心理，提供更加個性化和優質的服務。這不僅有助于企業更好地滿足客戶的需求，提升客戶滿意度，還可以為企業帶來更多的商業價值和競爭優勢。
群聊對話分角色要素提取的研究，將企業對話數據轉化為可用的信息和智能的洞察，為企業營銷和服務提供了新的思路和方法。通過挖掘對話數據中隱藏的客戶行為特征和趨勢，企業可以更加精準地進行客戶定位、推廣營銷和產品服務，實現營銷效果的最大化和客戶價值的最大化。這將為企業帶來更廣闊的發展空間和更持續的競爭優勢。

相關鏈接：
基于星火大模型的群聊對話分角色要素提取挑戰賽
零基礎入門大模型技術競賽-速通學習手冊

1 環境配置

1.1 下載包

本項目是基于windows環境，pycharm編譯器，星火認知大模型Spark3.5 Max，首先安裝軟件包

pip install --upgrade spark_ai_python  # 這里相較于baseline版本去掉-q

國內使用：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple spark_ai_python

如果清華源版本不可用,請使用一下命令升級到最新版本:

pip install -i  https://repo.model.xfyun.cn/api/packages/administrator/pypi/simple  spark_ai_python --upgrade

或者，開那個解決。（最好用）

我在安裝的時候出現了報錯，經分析是網絡問題，開那個解決了。
在這里插入圖片描述

注：項目僅支持 Python3.8+

1.2 配置密鑰

這里密鑰從訊飛開發者平臺申請，Datawhale還幫我們申請了200w的token，隨便花！

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'

1.3 測試模型

baseline提供了一個函數用于測試環境以及API是否配置合理，直接運行即可

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 = "你是誰？"
print(get_completions(text))  # 注意這里要添加一個print函數

注意：相較于baseline，最后一行添加了print函數，因為百度在線平臺會直接打印以及輸出圖片。

直接調用了ChatMessage()用于獲取用戶輸入的字符串，其中role參數：system用于設置對話背景，user表示是用戶的問題，assistant表示AI的回復。content是用戶和AI的對話內容。
ChatSparkLLM()構造了星火模型，其中streaming參數指的是采用一次性返回結果（非流式）還是采用流式返回結果。這里簡單測試，采用False。

詳細的SDK說明可以在星火的Github查看。

2 解決問題

2.1 獲取數據

def read_json(json_file_path):"""讀取json文件"""with open(json_file_path, 'r', encoding='utf-8') as f:data = json.load(f)return datadef write_json(json_file_path, data):"""寫入json文件"""with open(json_file_path, 'w', encoding='utf-8') 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")

這里就是很普通的實現了Json文件的讀取函數和寫入函數。注意win環境下在讀取和寫入的時候要添加, encoding='utf-8'，否則會讀取失敗。在許多中文Windows系統中，默認編碼是gbk，而不是utf-8。

2.2 設計Prompt

baseline提供的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格式輸出。
不要添加任何澄清信息。
輸出必須遵循上面的模式。
不要添加任何沒有出現在模式中的附加字段。
不要隨意刪除字段。**輸出：**
'''
[{{"基本信息-姓名": "姓名","基本信息-手機號碼": "手機號碼","基本信息-郵箱": "郵箱","基本信息-地區": "地區","基本信息-詳細地址": "詳細地址","基本信息-性別": "性別","基本信息-年齡": "年齡","基本信息-生日": "生日","咨詢類型": ["咨詢類型"],"意向產品": ["意向產品"],"購買異議點": ["購買異議點"],"客戶預算-預算是否充足": "充足或不充足","客戶預算-總體預算金額": "總體預算金額","客戶預算-預算明細": "預算明細","競品信息": "競品信息","客戶是否有意向": "有意向或無意向","客戶是否有卡點": "有卡點或無卡點","客戶購買階段": "購買階段","下一步跟進計劃-參與人": ["跟進計劃參與人"],"下一步跟進計劃-時間點": "跟進計劃時間點","下一步跟進計劃-具體事項": "跟進計劃具體事項"
}}, ...]
'''
"""

2.2 設計處理函數

import jsonclass JsonFormatError(Exception):def __init__(self, message):self.message = messagesuper().__init__(self.message)def convert_all_json_in_text_to_dict(text):"""提取LLM輸出文本中的json字符串"""dicts, stack = [], []for i in range(len(text)):if text[i] == '{':stack.append(i)elif text[i] == '}':begin = stack.pop()if not stack:dicts.append(json.loads(text[begin:i+1]))return dicts# 查看對話標簽
def print_json_format(data):"""格式化輸出json格式"""print(json.dumps(data, indent=4, ensure_ascii=False))def check_and_complete_json_format(data):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):raise JsonFormatError("Data is not a list")for item in data: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

JsonFormatError 類：這是一個自定義異常類，繼承自Python內置的Exception類。當遇到JSON格式錯誤時，這個異常會被拋出。它接收一個消息參數，并將其存儲在message屬性中。
convert_all_json_in_text_to_dict 函數：這個函數接受一個字符串參數text，然后掃描這個字符串，尋找JSON對象，并將它們轉換為Python字典。它使用一個棧來處理嵌套的JSON對象，并只在找到匹配的括號對時才嘗試解析JSON。
print_json_format 函數：這個函數接受一個Python字典data作為參數，并將其轉換為格式化的JSON字符串，然后打印出來。它使用json.dumps函數來實現這個轉換，其中indent=4用于美化輸出，ensure_ascii=False允許打印非ASCII字符。
check_and_complete_json_format 函數：這個函數用于檢查一個列表中的每個字典是否包含一組特定的鍵，并且這些鍵對應的值的類型也是正確的。

2.3 開始提取

from tqdm import tqdmretry_count = 5 # 重試次數
result = []
error_data = []for index, data in tqdm(enumerate(test_data)):index += 1is_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": index})is_success = Truebreakexcept Exception as e:print("index:", index, ", error:", e)continueif not is_success:data["index"] = indexerror_data.append(data)
write_json("output.json", result)

附全流程代碼

配置好虛擬環境，下載兩個json文件到dataset文件夾下即可直接使用。

"""
==========================================
@author: Seaton
@Time: 2024/6/29:下午7:19
@IDE: PyCharm
@Summary：Task01:baseline實現
==========================================
"""
from sparkai.llm.llm import ChatSparkLLM, ChunkPrintHandler
from sparkai.core.messages import ChatMessage
import json
from tqdm import tqdm# 星火認知大模型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 = "你是誰？"
# print(get_completions(text))  # 注意這里要添加一個print函數def read_json(json_file_path):"""讀取json文件"""with open(json_file_path, 'r', encoding='utf-8') as f:data = json.load(f)return datadef write_json(json_file_path, data):"""寫入json文件"""with open(json_file_path, 'w', encoding='utf-8') 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")# 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格式輸出。
不要添加任何澄清信息。
輸出必須遵循上面的模式。
不要添加任何沒有出現在模式中的附加字段。
不要隨意刪除字段。**輸出：**
'''
[{{"基本信息-姓名": "姓名","基本信息-手機號碼": "手機號碼","基本信息-郵箱": "郵箱","基本信息-地區": "地區","基本信息-詳細地址": "詳細地址","基本信息-性別": "性別","基本信息-年齡": "年齡","基本信息-生日": "生日","咨詢類型": ["咨詢類型"],"意向產品": ["意向產品"],"購買異議點": ["購買異議點"],"客戶預算-預算是否充足": "充足或不充足","客戶預算-總體預算金額": "總體預算金額","客戶預算-預算明細": "預算明細","競品信息": "競品信息","客戶是否有意向": "有意向或無意向","客戶是否有卡點": "有卡點或無卡點","客戶購買階段": "購買階段","下一步跟進計劃-參與人": ["跟進計劃參與人"],"下一步跟進計劃-時間點": "跟進計劃時間點","下一步跟進計劃-具體事項": "跟進計劃具體事項"
}}, ...]
'''
"""class JsonFormatError(Exception):def __init__(self, message):self.message = messagesuper().__init__(self.message)def convert_all_json_in_text_to_dict(text):"""提取LLM輸出文本中的json字符串"""dicts, stack = [], []for i in range(len(text)):if text[i] == '{':stack.append(i)elif text[i] == '}':begin = stack.pop()if not stack:dicts.append(json.loads(text[begin:i+1]))return dicts# 查看對話標簽
def print_json_format(data):"""格式化輸出json格式"""print(json.dumps(data, indent=4, ensure_ascii=False))def check_and_complete_json_format(data):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):raise JsonFormatError("Data is not a list")for item in data: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 dataretry_count = 5  # 重試次數
result = []
error_data = []for index, data in tqdm(enumerate(test_data)):index += 1is_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": index})is_success = Truebreakexcept Exception as e:print("index:", index, ", error:", e)continueif not is_success:data["index"] = indexerror_data.append(data)write_json("output.json", result)