llm_eval_script/gemini_google_nxcloud_choice.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
"""
https://cloud.google.com/vertex-ai/generative-ai/docs/partner-models/claude?hl=zh-cn
https://cloud.google.com/vertex-ai/generative-ai/docs/partner-models/claude/use-claude?hl=zh-cn


Llama

https://cloud.google.com/vertex-ai/generative-ai/docs/partner-models/llama/use-llama?hl=zh-cn
https://cloud.google.com/vertex-ai/generative-ai/docs/partner-models/llama/use-llama?hl=zh-cn#regions-quotas

Model Name
llama-4-maverick-17b-128e-instruct-maas
llama-4-scout-17b-16e-instruct-maas

区域选择 us-east5



Model Name

gemini-2.5-pro
The model does not support setting thinking_budget to 0.
Unable to submit request because thinking_budget is out of range; supported values are integers from 128 to 32768.


"""
import argparse
from datetime import datetime
import json
import os
from pathlib import Path
import sys
import time
import tempfile
from zoneinfo import ZoneInfo  # Python 3.9+ 自带，无需安装

pwd = os.path.abspath(os.path.dirname(__file__))
sys.path.append(os.path.join(pwd, "../"))

from google import genai
from google.genai import types

from project_settings import environment, project_path


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--model_name",
        # default="gemini-2.5-pro",   # The model does not support setting thinking_budget to 0.
        default="gemini-2.5-flash",
        # default="gemini-2.5-flash-lite-preview-06-17",
        # default="llama-4-maverick-17b-128e-instruct-maas",
        # default="llama-4-scout-17b-16e-instruct-maas",
        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="google_potent_veld_462405_t3",
        default="google_nxcloud_312303",
        type=str
    )
    parser.add_argument(
        "--create_time_str",
        default="null",
        # default="20250731_162116",
        type=str
    )
    parser.add_argument(
        "--interval",
        default=1,
        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()

    service = environment.get(args.service, dtype=json.loads)
    project_id = service["project_id"]

    google_application_credentials = Path(tempfile.gettempdir()) / f"llm_eval_system/{project_id}.json"
    google_application_credentials.parent.mkdir(parents=True, exist_ok=True)

    with open(google_application_credentials.as_posix(), "w", encoding="utf-8") as f:
        content = json.dumps(service, ensure_ascii=False, indent=4)
        f.write(f"{content}\n")

    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = google_application_credentials.as_posix()

    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 = "20250729-interval-5"
    else:
        create_time_str = args.create_time_str

    eval_dataset = eval_dataset_dir / args.eval_dataset_name

    output_file = eval_data_dir / f"gemini_google_nxcloud_choice/google/{args.model_name}/{args.client}/{args.service}/{create_time_str}/{args.eval_dataset_name}"
    output_file.parent.mkdir(parents=True, exist_ok=True)

    client = genai.Client(
        vertexai=True,
        project=project_id,
        location="global",
        # location="us-east5",
    )
    generate_content_config = types.GenerateContentConfig(
        top_p=0.95,
        temperature=0.6,
        max_output_tokens=1,
        response_modalities=["TEXT"],
        thinking_config=types.ThinkingConfig(
            thinking_budget=0
        )
    )

    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)}")
            contents = [
                types.Content(
                    role="user" if messages[0]["role"] == "user" else "model",
                    parts=[
                        types.Part.from_text(text=messages[0]["content"])
                    ]
                ),
                types.Content(
                    role=messages[1]["role"],
                    parts=[
                        types.Part.from_text(text=messages[1]["content"])
                    ]
                )
            ]

            time.sleep(args.interval)
            print(f"sleep: {args.interval}")
            time_begin = time.time()
            llm_response: types.GenerateContentResponse = client.models.generate_content(
                model=args.model_name,
                contents=contents,
                config=generate_content_config,
            )
            time_cost = time.time() - time_begin
            print(f"time_cost: {time_cost}")
            try:
                prediction = llm_response.candidates[0].content.parts[0].text
            except TypeError as e:
                print(f"request failed, error type: {type(e)}, error text: {str(e)}")
                continue
            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")

    return


if __name__ == "__main__":
    main()