import json import csv import os import pandas as pd import openai import time import requests from dotenv import load_dotenv from tqdm import tqdm # 加载环境变量（如果有.env文件） load_dotenv() # 配置SiliconFlow API SILICONFLOW_API_KEY = os.getenv("SILICONFLOW_API_KEY", "sk-ypjvmantsostdxrkirhidrtswohjpmlzuhyqojpudbreakwk") SILICONFLOW_API_BASE = os.getenv("SILICONFLOW_API_BASE", "https://api.siliconflow.cn/v1") # 保留OpenAI API配置（作为备选） OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your_openai_api_key_here") openai.api_key = OPENAI_API_KEY # 可以配置为Azure OpenAI #AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT", "") #AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY", "") #AZURE_DEPLOYMENT_NAME = os.getenv("AZURE_DEPLOYMENT_NAME", "") # 获取Azure配置参数 AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT", "") AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY", "") AZURE_DEPLOYMENT_NAME = os.getenv("AZURE_DEPLOYMENT_NAME", "") # 如果有Azure OpenAI配置，则使用Azure OpenAI if AZURE_OPENAI_ENDPOINT.strip() and AZURE_OPENAI_API_KEY.strip() and AZURE_DEPLOYMENT_NAME.strip(): openai.api_type = "azure" openai.api_base = AZURE_OPENAI_ENDPOINT openai.api_key = AZURE_OPENAI_API_KEY openai.api_version = "2023-05-15" # 可能需要根据实际情况调整 # 定义TruePositive的标准（根据mvp三类案例） def define_positive_sample_criteria(): """ 定义TruePositive的标准 根据搜索结果，TruePositive被定义为"mvp三类案例"，但没有找到具体定义 这里我们定义一些可能的标准，实际使用时可以根据需求调整 """ return """ 请判断以下消息是否属于TruePositive。TruePositive定义为与任务管理、待办事项、提醒、通知筛选相关的有用信息，具体包括： 1. 包含明确的任务、待办事项或需要完成的工作 2. 包含时间安排、截止日期或日程提醒 3. 包含项目进展、状态更新或工作报告 如果消息符合以上任一条件，则为TruePositive；否则为TrueNegative。 请只回答"TruePositive"或"TrueNegative"。 """ # 使用大模型API进行分类 def classify_with_llm(message, criteria, max_retries=3, retry_delay=2): """ 使用大模型API对消息进行分类 Args: message: 要分类的消息内容 criteria: 分类标准 max_retries: 最大重试次数 retry_delay: 重试延迟（秒） Returns: str: "TruePositive" 或 "TrueNegative" """ prompt = f"{criteria}\n\n消息内容: {message}" system_message = "你是一个专业的数据分类助手，根据给定标准判断消息是TruePositive还是TrueNegative。" for attempt in range(max_retries): try: # 使用SiliconFlow API headers = { "Content-Type": "application/json", "Authorization": f"Bearer {SILICONFLOW_API_KEY}" } payload = { "model": "deepseek-ai/DeepSeek-V3", "messages": [ {"role": "system", "content": system_message}, {"role": "user", "content": prompt} ], "stream": False, "max_tokens": 512, "temperature": 0.1, "top_p": 0.7, "top_k": 50, "frequency_penalty": 0.5, "n": 1 } response = requests.post( f"{SILICONFLOW_API_BASE}/chat/completions", headers=headers, json=payload ) # 检查响应状态 response.raise_for_status() response_data = response.json() # 解析响应 result = response_data["choices"][0]["message"]["content"].strip() # 标准化结果 if "TruePositive" in result: return "TruePositive" else: return "TrueNegative" except Exception as e: if attempt < max_retries - 1: print(f"API调用失败，{retry_delay}秒后重试: {e}\n响应状态码: {response.status_code if 'response' in locals() else 'N/A'}\n响应内容: {response.text if 'response' in locals() else 'N/A'}") time.sleep(retry_delay) else: print(f"API调用失败，达到最大重试次数: {e}\n最后响应状态码: {response.status_code if 'response' in locals() else 'N/A'}\n最后响应内容: {response.text if 'response' in locals() else 'N/A'}") return "分类失败" # 返回一个默认值 # 批量处理消息 def batch_process_messages(messages, batch_size=10, delay=1): """ 批量处理消息以避免API限制 Args: messages: 消息列表 batch_size: 每批处理的消息数量 delay: 批次间延迟（秒） Returns: list: 处理结果列表 """ results = [] criteria = define_positive_sample_criteria() for i in tqdm(range(0, len(messages), batch_size), desc="处理批次"): batch = messages[i:i+batch_size] batch_results = [] for msg in tqdm(batch, desc="处理消息", leave=False): # 只处理有实际内容的消息 if msg.get("content") and len(msg["content"]) > 5: # 忽略过短的消息 classification = classify_with_llm(msg["content"], criteria) msg["classification"] = classification else: msg["classification"] = "TrueNegative" # 默认短消息为TrueNegative batch_results.append(msg) results.extend(batch_results) if i + batch_size < len(messages): time.sleep(delay) # 批次间延迟 return results # 主函数 def main(): # 检查API密钥是否配置 if SILICONFLOW_API_KEY == "": print("警告: 未设置SiliconFlow API密钥。请设置环境变量SILICONFLOW_API_KEY或在代码中直接设置。") return # 确定输入文件 input_file = "Messages.json" # 默认使用JSON格式 if not os.path.exists(input_file): print(f"错误: 找不到JSON输入文件 {input_file}") return print(f"使用输入文件: {input_file}") # 读取数据 messages = [] if input_file.endswith(".json"): with open(input_file, "r", encoding="utf-8") as f: messages = json.load(f) print(f"读取了 {len(messages)} 条消息") # 询问用户是否要处理所有消息或仅处理一部分样本 sample_size = input("请输入要处理的消息数量（输入'all'处理所有消息，或输入一个数字如'100'处理部分消息）: ") if sample_size.lower() != "all": try: sample_size = int(sample_size) if sample_size < len(messages): print(f"将处理 {sample_size} 条消息作为样本") messages = messages[:sample_size] else: print(f"样本大小大于等于总消息数，将处理所有 {len(messages)} 条消息") except ValueError: print("无效输入，将处理所有消息") # 批量处理消息 print("开始处理消息...") classified_messages = batch_process_messages(messages) # 分离TruePositive from TrueNegative positive_samples = [msg for msg in classified_messages if msg.get("classification") == "TruePositive"] negative_samples = [msg for msg in classified_messages if msg.get("classification") == "TrueNegative"] print(f"分类完成: TruePositive {len(positive_samples)} 条, TrueNegative {len(negative_samples)} 条") # 保存结果 if input_file.endswith(".json"): # 保存JSON格式 with open("positive_samples.json", "w", encoding="utf-8") as f: json.dump(positive_samples, f, ensure_ascii=False, indent=2) with open("negative_samples.json", "w", encoding="utf-8") as f: json.dump(negative_samples, f, ensure_ascii=False, indent=2) print("结果已保存到 positive_samples.json/csv 和 negative_samples.json/csv") if __name__ == "__main__": main()