local-tool/analysis/call_rating.py

import json
from typing import TypedDict, Optional
from openai import OpenAI
from pydantic import BaseModel

from utils.file_util import ExcelFile


class ScoringRule(BaseModel):
    score: str
    rule: str


class ScoringRuleList(BaseModel):
    rules: list[ScoringRule]


class HeaderIndex(TypedDict, total=False):
    human_intent: int
    human_rating: int
    robot_intent: int
    robot_rating: int
    llm_intent: int
    llm_rating: int
    chat_history: int
    chat_history_simple: int
    scoring_criteria: int
    error_reasons: int


class RatingScore(BaseModel):
    score: str
    confidence_score: str
    scoring_criteria: str


class InferenceScoringRuleAdvanced(BaseModel):
    rule: str

# 金融
# no_intention = ['非目标人群', 'F非目标人群', '贷不了', 'E贷不了', '语音助手', 'F语音助手', '接通挂机', 'F接通挂机']
# Intended = ['问贷款信息', 'C问贷款信息', '成功结束', 'A成功结束', '在忙待跟进', 'D在忙待跟进', '待跟进', 'B待跟进']

# pos
no_intention = ['无意向', '不需要', '非目标人群', '已经有']
Intended = ['有意向', 'A邀约成功']


class CallRating:
    def __init__(self, file_path):
        self._file = ExcelFile(file_path)
        api_key = "sk-TL1U0f39NF5xOFnm8bB39e9b73E2474b9753C6BfCe95B5Db" # 计价测试
        # api_key = "sk-uN8GZmYwwu8bKeZyF605D111AcCa48738cD777B1332348D1" # 调试
        base_url = "https://newapi.gkscrm.com/v1"
        self.model = "gpt-4o"
        # self.model = "qwen2.5:7b"
        # base_url = "http://localhost:11434/v1"

        self._openai_client = OpenAI(api_key=api_key,
                                     base_url=base_url)

    # 评分方法
    def rating_score(self, scoring_rule: list[ScoringRule], general_rule: Optional[int] = 0, start_row: Optional[int] = 0, end_row: Optional[int] = 0):
        # 获取表头索引
        header_index = self._get_header_index()
        if start_row == 0 and end_row == 0:
            df_subset = self._file.file
        else:
            df_subset = self._file.file.iloc[start_row:end_row]
        for index, row in df_subset.iterrows():
            robot_intent = row.iloc[header_index["robot_intent"]]
            chat_history = row.iloc[header_index["chat_history"]]
            print(f"chat_history:{chat_history}")
            chat_history = self._format_chat_history(chat_history)
            print(f"chat_history:{chat_history}")
            completion = self._openai_client.beta.chat.completions.parse(
                model=self.model,
                messages=[
                    {"role": "system", "content": f"""# 任务
1. 首先，判断用户的第一句话是否说了：“你好，(任意内容)通话”，如果说了，则不用理会评级规则，直接强制分配为"语音助手"
2. 如果不属于“语音助手”，请根据评级规则，对聊天记录给出评级、置信度、评分依据（逐项分析不要遗漏）

# 细节说明
置信度从0到1，0为置信度最低，1为置信度最高。"""},
                    {
                        "role": "user",
                        "content": f"""# 评级规则：
{scoring_rule}

# 聊天记录
{chat_history}
"""
                    }],
                response_format=RatingScore
                # response_format={"type": "json_object"}
            )
            score = json.loads(completion.choices[0].message.content)["score"]
            confidence_score = json.loads(completion.choices[0].message.content)["confidence_score"]
            scoring_criteria = json.loads(completion.choices[0].message.content)["scoring_criteria"]

            if robot_intent in Intended:
                self._file.new_value(index, "机器人意向", "有意向")
            elif robot_intent in no_intention:
                self._file.new_value(index, "机器人意向", "无意向")
            else:
                self._file.new_value(index, "机器人意向", "不确定")
            self._file.new_value(index, "精简聊天记录", chat_history)
            self._file.new_value(index, "大模型评级", score)
            if score in Intended:
                self._file.new_value(index, "大模型意向", "有意向")
            elif score in no_intention:
                self._file.new_value(index, "大模型意向", "无意向")
            else:
                self._file.new_value(index, "大模型意向", "不确定")
            self._file.new_value(index, "置信度", confidence_score)
            self._file.new_value(index, "评分依据", scoring_criteria)

    def rating_score_test(self, scoring_rule: list[ScoringRule]):
        header_index = self._get_header_index()
        self._file.new_column("大模型意向")
        self._file.new_column("置信度")
        for index, row in self._file.file.iterrows():
            chat_history = row.iloc[header_index["chat_history"]]
            chat_history = self._format_chat_history(chat_history)
            print(f"chat_history:{chat_history}")
            completion = self._openai_client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": f"""请告诉我，用户的第一句话是否说了：'你好，(任意内容)通话'"""},
                    {
                        "role": "user",
                        "content": f"""
# 聊天记录
{chat_history}
"""
                    }]
            )

            print(f"completion:{completion}")

    def inference_scoring_rule_advanced(self, scoring_rule: list[ScoringRule]):
        self.rating_score(scoring_rule)
        header_index = self._get_header_index()
        system_prompt = '''# 角色
我希望你扮演逆向规则工程师，根据多组聊天记录、评级错误结果、评级错误原因，完善现有评级规则。

# 输出要求
1.聚类相同评级，不要一个评级输出多次规则
2.不要更改或合并评级'''
        user_prompt = f"""# 现有评级规则：
{scoring_rule}\n"""
        for index, row in self._file.file.iterrows():
            human_intent = row.iloc[header_index["human_intent"]]
            chat_history = row.iloc[header_index["chat_history"]]
            chat_history = self._format_chat_history(chat_history)
            error_reasons = row.iloc[header_index["error_reasons"]]
            user_prompt += f"""# 第{index + 1}组
## 聊天记录
{chat_history}
## 错误评级结果
{human_intent}

## 评级错误原因
{error_reasons}\n\n"""
        completion = self._openai_client.beta.chat.completions.parse(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {
                    "role": "user",
                    "content": user_prompt
                }
            ],
            response_format=ScoringRuleList
        )
        return json.loads(completion.choices[0].message.content)["rules"]

    def iterate_scoring_rule(self, iteration_count: int, scoring_rule: Optional[list[ScoringRule]] = None):
        # if iteration_count < 1:
        #     return
        if scoring_rule is None:
            scoring_rule = self.inference_scoring_rule()
        self.rating_score(scoring_rule)
        iterate = False
        header_index = self._get_header_index()
        system_prompt = '''# 角色
我希望你扮演逆向规则工程师，帮我完善评级现有评级规则。

# 过程
1. 根据评级错误的组，完善旧的评级规则
2. 完善后的评级规则应当能够正确评级所有组

# 输出要求
1.聚类相同评级，不要一个评级输出多次规则
2.不要更改或合并评级'''
        user_prompt = "# 现有评级规则：{scoring_rule}\n"
        i = 0
        for index, row in self._file.file.iterrows():
            llm_intent = row.iloc[header_index["llm_intent"]]
            human_intent = row.iloc[header_index["human_intent"]]
            chat_history = row.iloc[header_index["chat_history"]]
            chat_history = self._format_chat_history(chat_history)
            if llm_intent != human_intent:
                iterate = True
                i += 1
                print(f"{i}第{index + 2}行，大模型评级：{llm_intent}，人工评级：{human_intent}")
                user_prompt += f"""# 第{index + 1}组：有错误
## 聊天记录
{chat_history}
## 根据现有规则产生的错误评级结果
{llm_intent}
## 正确的评级结果应当是
{human_intent}\n"""
            else:
                user_prompt += f"""# 第{index + 1}组：无错误
## 聊天记录
{chat_history}
## 根据现有规则产生了正确的评级结果
{llm_intent}\n"""
        if iteration_count < 1:
            return scoring_rule
        if not iterate:
            return scoring_rule

        completion = self._openai_client.beta.chat.completions.parse(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {
                    "role": "user",
                    "content": user_prompt
                }
            ],
            response_format=ScoringRuleList
        )
        print(iteration_count)
        print(json.loads(completion.choices[0].message.content)["rules"])
        return self.iterate_scoring_rule(iteration_count - 1,
                                         json.loads(completion.choices[0].message.content)["rules"])

    def inference_scoring_rule(self):
        header_index = self._get_header_index()
        system_prompt = '''# 角色
我希望你扮演推理机器，根据多组聊天记录和评级结果，推理出评级规则。

# 输出要求
1.聚类相同评级，不要一个评级输出多次规则
2.不要更改或合并评级'''
        user_prompt = ""
        for index, row in self._file.file.iterrows():
            human_intent = row.iloc[header_index["human_intent"]]
            chat_history = row.iloc[header_index["chat_history"]]
            chat_history = self._format_chat_history(chat_history)
            user_prompt += f"""# 第{index + 1}组
## 聊天记录
{chat_history}
## 评级结果
{human_intent}\n"""
        completion = self._openai_client.beta.chat.completions.parse(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {
                    "role": "user",
                    "content": user_prompt
                }
            ],
            response_format=ScoringRuleList
        )
        return json.loads(completion.choices[0].message.content)["rules"]

    def _get_header_index(self):
        header_index: HeaderIndex = {}
        for index, column in enumerate(self._file.file.columns):
            if "人工意向" == column:
                header_index["human_intent"] = index
            elif "人工评级" == column:
                header_index["human_rating"] = index
            elif "机器人意向" == column:
                header_index["robot_intent"] = index
            elif "机器人评级" == column:
                header_index["robot_rating"] = index
            elif "大模型意向" == column:
                header_index["llm_intent"] = index
            elif "大模型评级" == column:
                header_index["llm_rating"] = index
            elif "大模型意向（通用规则）" == column:
                header_index["general_llm_intent"] = index
            elif "大模型评级（通用规则）" == column:
                header_index["general_llm_rating"] = index
            elif "聊天记录" == column:
                header_index["chat_history"] = index
            elif "精简聊天记录" == column:
                header_index["chat_history_simple"] = index
            elif "评分依据" == column:
                header_index["scoring_criteria"] = index
            elif "错误原因" == column:
                header_index["error_reasons"] = index
        return header_index

    def _get_rating_type(self):
        column_values = self._file.file.loc[:, '人工意向']
        # 遍历列的值，获取评级类型
        rating_type = set()
        for value in column_values:
            rating_type.add(value)
        return rating_type

    @staticmethod
    def _format_chat_history(chat_history):
        chat_history_data = json.loads(chat_history)
        format_chat_history = ""
        for message in chat_history_data:
            if message["type"] == "ai":
                role = "机器人"
            elif message["type"] == "user":
                role = "用户"
            else:
                role = "系统"
            format_chat_history += f"{role}: {message['content']}\n"
        return format_chat_history