tools-api/batch_task/update_mismatch_record.py

import json

import asyncio
from pydantic import BaseModel

from app.admin.schema.intent_org import CurrentIntentOrgIns
from app.call_center.crud.crud_mismatch_records import mismatch_records_dao
from app.call_center.schema.mismatch_records import GetMismatchRecordsDetails
from common.log import log
from common.oai import send_request_with_retry, generate_text, generate_json_by_class
from database.db_mysql import async_db_session
from model.mismatch_records import MismatchRecords
from utils.serializers import select_as_dict


class Keyword(BaseModel):
    user_intent: str
    similar_reply: list[str]
    keywords: list[str]
    regular: list[str]

async def update_mismatch_record(org_map: dict[int, CurrentIntentOrgIns], limit: int = 1):
    async with async_db_session.begin() as db:
        record = await mismatch_records_dao.get_earliest_record(db, limit)
        if not record:
            return 0
        await asyncio.gather(*(process_mismatch_record(org_map[r.org_id], r) for r in record))
        return len(record)

async def process_mismatch_record(org_data: CurrentIntentOrgIns, record: MismatchRecords):
    record_data = GetMismatchRecordsDetails(**select_as_dict(record))
    log.info(f"[process_mismatch_record] record_data: {record_data}")

    # 开始过滤
    llm_ignore = record_data.llm_ignore
    ignore_response_data = None
    if record_data.ignore == 0:
        ignore_messages = [
            {"role": "system", "content": f"""# 任务介绍
公司在用语音识别系统巡检电话通话记录时，可能会因为通话时的背景噪音、方言等问题，导致用户的语音内容没有被正确识别，请根据以下通话记录，判断识别到的内容是否无意义或使上下文不通顺，非常感谢！

# 注意事项
不用考虑机器人的回复，因为即便识别正确，机器人的回复也可能有误

# 输出要求
1. 如放在上下文中无意义，请回复一个数字0
2. 如放在上下文中有意义，请回复一个数字1
3. 请不要回复 0 或 1 以外的文字内容"""
             },
            {
                "role": "user",
                "content": f"""# 通话记录
{record_data.chat_history}

# 可能识别有误的内容：{record_data.missed}
"""
            }
        ]
        ignore_response_data = await generate_text(org_data.openai_key, org_data.openai_base, "gpt-3.5-turbo", ignore_messages)
        if ignore_response_data and isinstance(ignore_response_data.choices, list) and len(ignore_response_data.choices) > 0:
            first_choice = ignore_response_data.choices[0]
            if first_choice and first_choice.message:
                ignore_content = first_choice.message.content
                if ignore_content == "0":
                    status = 1
                    llm_ignore = 1

                    # 推送
                    url = org_data.mismatch_callback
                    log.error(f"mismatch_callback: {url}")
                    if url:
                        headers = {
                            "Content-Type": "application/json"
                        }
                        data = {
                            "internal_id": record_data.id,
                            "external_id": record_data.external_id,
                            "ignore": record_data.ignore,
                            "llm_ignore": llm_ignore
                        }

                        is_success = await send_request_with_retry(url, data, headers, max_retries=3,
                                                                   delay_between_retries=2)
                        if is_success:
                            status = 2
                    async with async_db_session.begin() as db:
                        try:
                            await mismatch_records_dao.update_llm_ignore(db, record_data.id, llm_ignore,
                                                                         ignore_response_data.to_dict(),
                                                                         status)
                        except Exception as e:
                            log.error(f"更新意图记录时发生异常：{e}")
                        finally:
                            return None
                elif ignore_content == "1":
                    llm_ignore = 2
                    ignore_response_data = ignore_response_data.to_dict()


    # 开始分析
    keyword_messages = [
        {"role": "system", "content": f"""# 任务介绍
公司在用程序巡检电话通话记录时，有些用户的回复内容没有被关键词库识别到，请根据以下通话记录，给出一些类似的回复，以及对关键词库的具体建议，非常感谢！

# 输出要求
1. 用户意图：结合上下文，首先考虑可能的语音识别错误并纠错，然后分析未识别内容的根本意图，用户说这句话的背后真实目的是什么，结合上下文挖掘用户最深的意图，请不要停留在表面的意思。
2. 类似回复：首先应包含未识别的内容原文：{record_data.missed}。其次生成尽可能多的类似意图的回复。
2. 关键词库：从类似回复中抽取特征词，要求有两点：一是特征词应尽可能的准确，即要覆盖住此类问题，又要尽可能避免和其他语境相冲突。二是需要注重通用性，一些专有名词、人名、地名、产品名、公司名等需要排除在外。
3. 正则表达式：对关键词词进行解耦，用正则表达式进行表示，专注于主要内容，排除次要内容

# 注意事项
1. 应贴近真实的外呼场景，用户的素质参差不齐，请避免使用任何浮夸的用词，避免使用高级词汇，避免使用任何的礼貌用语或敬语，适当的低素质些，请至少给出20条结果
2. 通话记录为录音转译，因此可能有错别字或音同字不同的情况（例如：借和接），请根据上下文分析后就成可能的错字错误
3. 正则关键词库和正则表达式中也应该考虑到音同字不同的情况，且避免使用匹配次数相关的语法如{0,2}"""
         },
        {
            "role": "user",
            "content": f"""# 通话记录
{record_data.chat_history}

# 可能识别有误的内容：{record_data.missed}
"""
        }
    ]
    response_data = await generate_json_by_class(org_data.openai_key, org_data.openai_base, "gpt-4o", keyword_messages, Keyword)
    if response_data and isinstance(response_data.choices, list) and len(response_data.choices) > 0:
        first_choice = response_data.choices[0]
        if first_choice and first_choice.message:
            response_json = first_choice.message.content
            if response_json:
                intent = json.loads(response_json)

                user_intent = intent.get('user_intent')
                similar_reply = intent.get('similar_reply')
                keywords = intent.get('keywords', [])
                regular = intent.get('regular', [])
                if user_intent and similar_reply:
                    status = 1

                    # 推送
                    url = org_data.mismatch_callback
                    log.error(f"mismatch_callback: {url}")
                    if url:
                        headers = {
                            "Content-Type": "application/json"
                        }
                        data = {
                            "internal_id": record_data.id,
                            "external_id": record_data.external_id,
                            "ignore": record_data.ignore,
                            "llm_ignore": llm_ignore,
                            "user_intent": user_intent,
                            "similar_reply": similar_reply,
                            "keywords": keywords,
                            "regular": regular
                        }

                        is_success = await send_request_with_retry(url, data, headers, max_retries=3, delay_between_retries=2)
                        if is_success:
                            status = 2

                    async with async_db_session.begin() as db:
                        try:
                            await mismatch_records_dao.update(db, record_data.id, llm_ignore, user_intent, similar_reply, keywords, regular, {"messages": keyword_messages}, response_data.to_dict(), status, ignore_response_data)
                        except Exception as e:
                            log.error(f"更新意图记录时发生异常：{e}")