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
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
    ignore_model = None
    ignore_prompt_tokens = None
    ignore_completion_tokens = 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_model = "gpt-3.5-turbo"
        if org_data.model == "deepseek-v3" or org_data.model == "DeepSeek-V3":
            ignore_model = org_data.model

        ignore_content, ignore_response_data = await generate_text(org_data.openai_key, org_data.openai_base, ignore_model, ignore_messages)
        if ignore_content:
            ignore_prompt_tokens = ignore_response_data.usage.prompt_tokens
            ignore_completion_tokens = ignore_response_data.usage.completion_tokens
            if ignore_content == "0":
                status = 1
                llm_ignore = 1

                # 推送
                url = org_data.mismatch_callback
                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, ignore_model, ignore_prompt_tokens, ignore_completion_tokens)
                    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()
        else:
            return None


    # 开始分析
    if org_data.model == "deepseek-v3" or org_data.model == "DeepSeek-V3":
        keyword_schema = """{
    "user_intent": str, #用户意图
    "similar_reply": list[str], #类似回复
    "keywords": list[str], #关键词库
    "regular": list[str], #正则表达式
}"""
    else:
        keyword_schema = {
            "name": "keyword_schema",
            "schema": {  # 添加 schema 字段
                "type": "object",
                "description": "从通话记录中提取表单值",
                "properties": {
                    "user_intent": {"type": "string", "description": "用户意图"},
                    "similar_reply": {
                        "type": "array",
                        "description": "类似回复",
                        "items": {"type": "string"}
                    },
                    "keywords": {
                        "type": "array",
                        "description": "关键词库",
                        "items": {"type": "string"}
                    },
                    "regular": {
                        "type": "array",
                        "description": "正则表达式",
                        "items": {"type": "string"}
                    },
                },
                "required": ["user_intent", "similar_reply", "keywords", "regular"]
            }
        }
    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_json, response_data = await generate_json(org_data.openai_key, org_data.openai_base, org_data.model, keyword_messages, keyword_schema)
    if response_json:
        user_intent = response_json.get('user_intent')
        similar_reply = response_json.get('similar_reply')
        keywords = response_json.get('keywords', [])
        regular = response_json.get('regular', [])
        if user_intent and similar_reply:
            status = 1

            # 推送
            url = org_data.mismatch_callback
            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, ignore_model, ignore_prompt_tokens, ignore_completion_tokens, org_data.model, response_data.usage.prompt_tokens, response_data.usage.completion_tokens)
                except Exception as e:
                    log.error(f"更新意图记录时发生异常：{e}")