#!/usr/bin/python3 # -*- coding: utf-8 -*- import argparse from datetime import datetime import json import os from pathlib import Path import sys import time from zoneinfo import ZoneInfo # Python 3.9+ 自带，无需安装 pwd = os.path.abspath(os.path.dirname(__file__)) sys.path.append(os.path.join(pwd, "../")) import openai from openai import AzureOpenAI from project_settings import environment, project_path def get_args(): parser = argparse.ArgumentParser() parser.add_argument( "--model_name", # default="gpt-4o", default="gpt-4o-mini", type=str ) parser.add_argument( "--eval_dataset_name", default="agent-nxcloud-zh-375-choice.jsonl", type=str ) parser.add_argument( "--eval_dataset_dir", default=(project_path / "data/dataset").as_posix(), type=str ) parser.add_argument( "--eval_data_dir", default=(project_path / "data/eval_data").as_posix(), type=str ) parser.add_argument( "--client", default="shenzhen_sase", type=str ) parser.add_argument( "--service", default="west_us_chatgpt_openai_azure_com", type=str ) parser.add_argument( "--create_time_str", default="null", # default="20251209_151045", type=str ) parser.add_argument( "--interval", default=5, type=int ) args = parser.parse_args() return args def conversation_to_str(conversation: list): conversation_str = "" for turn in conversation: role = turn["role"] content = turn["content"] row_ = f"{role}: {content}\n" conversation_str += row_ return conversation_str system_prompt = """ 你是一位专业的电话对话分析专家，负责根据客服与客户之间的通话内容判断客户意图类别。 请仔细分析用户提供的完整对话，并严格按照以下规则进行分类： - **A**：客户**明确同意参加试听课**（如“好啊，安排一下”）。仅询问细节、模糊回应（如“嗯嗯”“好的”）不算。 - **B**：客户**投诉、辱骂、或明确要求停止拨打此类电话**（如“别再打了！”）。仅拒绝试听（如“不用了”）不属于 B。 - **C**：客户表示**当前时刻不方便通话，例如提到“在开车”、“不方便”等**。 - **D**：对话为**语音留言/自动应答**，或包含“留言”“voicemail”“message”“已录音”等关键词，或出现**逐字念出的数字串**（如“九零九五……”）。 - **E**：客服**完成两次独立推销后**，客户**两次都表达了明确拒绝，仅一次不算做E分类**。 - **F**：客户未表达明确意愿，或以上情况均不符合（默认类别）。 **输出要求：** - 仅输出一个大写字母（A、B、C、D、E 或 F）； - 不要任何解释、标点、空格、换行、JSON、引号或其他字符； - 输出必须且只能是单个字母。 """ def main(): args = get_args() eval_dataset_dir = Path(args.eval_dataset_dir) eval_dataset_dir.mkdir(parents=True, exist_ok=True) eval_data_dir = Path(args.eval_data_dir) eval_data_dir.mkdir(parents=True, exist_ok=True) if args.create_time_str == "null": tz = ZoneInfo("Asia/Shanghai") now = datetime.now(tz) create_time_str = now.strftime("%Y%m%d_%H%M%S") # create_time_str = "20250722_173400" else: create_time_str = args.create_time_str eval_dataset = eval_dataset_dir / args.eval_dataset_name output_file = eval_data_dir / f"azure_openai_nxcloud_choice/azure/{args.model_name}/{args.client}/{args.service}/{create_time_str}/{args.eval_dataset_name}" output_file.parent.mkdir(parents=True, exist_ok=True) service_params = environment.get(args.service, dtype=json.loads) client = AzureOpenAI( **service_params, # api_key="Dqt75blRABmhgrwhfcupd1rq44YqNuEgku8FcFFDrEljMq6gltf0JQQJ99BCACYeBjFXJ3w3AAABACOG2njW", # api_version="2025-01-01-preview", # azure_endpoint="https://west-us-chatgpt.openai.azure.com" ) total = 0 total_correct = 0 # finished finished_idx_set = set() if os.path.exists(output_file.as_posix()): with open(output_file.as_posix(), "r", encoding="utf-8") as f: for row in f: row = json.loads(row) idx = row["idx"] total = row["total"] total_correct = row["total_correct"] finished_idx_set.add(idx) print(f"finished count: {len(finished_idx_set)}") with open(eval_dataset.as_posix(), "r", encoding="utf-8") as fin, open(output_file.as_posix(), "a+", encoding="utf-8") as fout: for row in fin: row = json.loads(row) idx = row["idx"] # system_prompt = row["system_prompt"] conversation = row["conversation"] examples = row["examples"] choices = row["choices"] response = row["response"] if idx in finished_idx_set: continue finished_idx_set.add(idx) # conversation conversation_str = conversation_to_str(conversation) examples_str = "" for example in examples: conversation_ = example["conversation"] outputs = example["outputs"] output = outputs["output"] explanation = outputs["explanation"] examples_str += conversation_to_str(conversation_) # output_json = {"Explanation": explanation, "output": output} # output_json_str = json.dumps(output_json, ensure_ascii=False) # examples_str += f"\nOutput: {output_json_str}\n" examples_str += f"\nOutput: {output}\n\n" # print(examples_str) choices_str = "" for choice in choices: condition = choice["condition"] choice_letter = choice["choice_letter"] row_ = f"{condition}, output: {choice_letter}\n" choices_str += row_ # choices_str += "\nRemember to output ONLY the corresponding letter.\nYour output is:" # choices_str += "\nPlease use only 10-15 words to explain.\nOutput:" # prompt = f"{system_prompt}\n\n**Output**\n{choices_}\n**Examples**\n{examples_}" prompt1 = f"{system_prompt}\n\n**Examples**\n{examples_str}" prompt2 = f"**Conversation**\n{conversation_str}\n\nOutput:" # print(prompt1) # print(prompt2) messages = list() messages.append( {"role": "system", "content": prompt1}, ) messages.append( {"role": "user", "content": prompt2}, ) # print(f"messages: {json.dumps(messages, ensure_ascii=False, indent=4)}") try: time.sleep(args.interval) print(f"sleep: {args.interval}") time_begin = time.time() llm_response = client.chat.completions.create( model=args.model_name, messages=messages, stream=False, max_tokens=1, top_p=0.95, temperature=0.6, logit_bias={ 32: 100, 33: 100, 34: 100, 35: 100, 36: 100, 37: 100, 38: 100, 39: 100, } ) time_cost = time.time() - time_begin print(f"time_cost: {time_cost}") except openai.BadRequestError as e: print(f"request failed, error type: {type(e)}, error text: {str(e)}") continue prediction = llm_response.choices[0].message.content correct = 1 if prediction == response else 0 total += 1 total_correct += correct score = total_correct / total row_ = { "idx": idx, "messages": messages, "response": response, "prediction": prediction, "correct": correct, "total": total, "total_correct": total_correct, "score": score, "time_cost": time_cost, } row_ = json.dumps(row_, ensure_ascii=False) fout.write(f"{row_}\n") fout.flush() return if __name__ == "__main__": main()