import requests
import json
def callServer(prompt, model="deepseek-ai/DeepSeek-V3.2"):
url = "https://api.siliconflow.cn/v1/chat/completions"
headers = {
"Authorization": f"Bearer sk-envatqyeffxlrtalgexbrueftyzjcezorhhenlexsioqesox",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
#注意,这里的tempreture要调低,保证结果稳定性
"temperature": 0.1,
"max_tokens": 6000
}
success = True
msg = "OK"
think = ""
result = ""
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
result = response.json()['choices'][0]['message']['content']
except Exception as e:
success = False
msg = "API调用异常:" + str(e)
return success, msg, result
def loadRawDataSet(fileName):
dataList = []
with open(fileName, "r", encoding="utf-8") as file:
while True:
line = file.readline()
if not line: # 读取到文件末尾时退出
break
line = line.strip()
if len(line)>0:
dataList.append(line.split("\t",1))
return dataList
def writeData(dataList,fileName):
with open(fileName, 'w', encoding='utf-8') as f:
for data in dataList:
f.write(data + "\n")
def tagValid(dataList,promptModel):
correctList = []
for data in dataList[400:402]:
prompt = promptModel.format(data[0],data[1])
success = False
while not success:
success, msg, result = callServer(prompt)
print(data[0],data[1])
print(result)
print("*"*80)
if success and len(result) == 1 and result in {"是","否"}:
correctList += list(result)
else:
success = False
writeData(correctList, "valid.txt")
if __name__ == "__main__":
dataList = loadRawDataSet("total.txt")
promptModel = """你是一位电商评论文本的分类专家,在【分类体系】规定的类别中,【输入文本】是不是能且仅能分类到{}。只允许输出是或者否,不允许输出其他字符
【分类体系】
["电脑","水果","平板","衣服","酒店","洗浴","书籍","蒙牛","电器","手机"]
【输入文本】
{}"""
tagValid(dataList,promptModel)