tools-api/batch_task/update_llm_intent.py

import json
from time import sleep
from typing import Dict

import requests
from openai import OpenAI
import asyncio
import aiohttp
from app.admin.schema.intent_org import CurrentIntentOrgIns
from app.call_center.crud.crud_intent_records import intent_records_dao
from app.call_center.schema.intent_records import GetIntentRecordsDetails
from common.log import log
from core.conf import settings
from database.db_mysql import async_db_session
from database.db_redis import redis_client
from utils.serializers import select_as_dict


async def update_llm_intent():
    async with async_db_session.begin() as db:
        record = await intent_records_dao.get_earliest_record(db)
        if not record:
            return None
        record_data = GetIntentRecordsDetails(**select_as_dict(record))

        # 机构 id
        org_id = record_data.org_id

        # 从缓存中获取机构信息
        key = f'{settings.TOKEN_CALL_REDIS_PREFIX}:{org_id}'
        org_json = await redis_client.get(key)
        if not org_json:
            # 缓存中没有，从数据库中获取
            from app.admin.crud.crud_intent_org import intent_org_dao
            org = await intent_org_dao.get(db, org_id)
            if not org and org.status is not 1:
                log.error(f"意向评级时，机构不存在 org_id: {org_id}")
                return None
            org_data = CurrentIntentOrgIns(**select_as_dict(org))
            # 将数据放进缓存
            await redis_client.setex(
                key,
                settings.JWT_USER_REDIS_EXPIRE_SECONDS,
                org_data.model_dump_json(),
            )
        else:
            org_data = CurrentIntentOrgIns(**json.loads(org_json))

        # 开始评级
        intent_schema = {
            "name": "intent_schema",
            "schema": {  # 添加 schema 字段
                "type": "object",
                "description": "为通话记录进行评级",
                "properties": {
                    "score": {"type": "string", "description": "评分结果：有意向、待进一步分析、暂时无法沟通、其他"},
                    "confidence_score": {"type": "number", "description": "置信度分数，范围从0.0到1.0"},
                    "scoring_criteria": {"type": "string", "description": "请逐步介绍为何评为这个结果"}
                },
                "required": ["score", "confidence_score", "scoring_criteria"]
            }
        }

        rules = [
            {
                'rule': "用户表现出对产品有积极回应，询问有关产品的问题，或愿意进行后续操作，如确认微信号、询问发货地址或表达接受产品（例：'行'、'好'、'可以'、'是的'）。",
                'score': '有意向'},
            {
                'rule': "用户虽然未积极回应或表示不需要，但却有相关需求、或着符合某项购买条件、又或着询问了机器人问题，即便后期疲软或持否定态度，可移交销售顾问进行进一步的需求分析，以便改善产品或定制符合客户要求的产品",
                'score': '待进一步分析'},
            {
                'rule': "用户表示自己在忙、在开会等，现在不太方便沟通。又或者对电话的真实性存疑，怀疑是诈骗电话，可后续让专业客服继续跟进",
                'score': '暂时无法沟通'},
            {
                'rule': "除上述规则外的其他用户，请注意客户资源珍贵，请尽量归类为'有意向'或'待进一步分析'，后续再由销售专家定性。",
                'score': '其他'}
        ]

        score_map = {
            '有意向': 1,
            '待进一步分析': 2,
            '暂时无法沟通': 3,
            '其他': 4
        }

        messages = [
            {"role": "system", "content": f"""# 任务
1. 首先，判断用户的第一句话是否说了：“你好，(任意内容)通话”，如果说了，则不用理会评级规则，直接强制分配为"语音助手"
2. 如果不属于“语音助手”，请根据评级规则，对聊天记录给出评级、置信度、评分依据（逐项分析不要遗漏）

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

            # 聊天记录
            {record_data.chat_history}
            """
            }
        ]
        response_data = generate_json(org_data.openai_key, org_data.openai_base, messages, intent_schema)
        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)

                    score = intent.get('score', "未知")
                    llm_intent = score_map.get(score, 0)
                    # confidence_score = intent.get('confidence_score', 0)
                    # scoring_criteria = intent.get('scoring_criteria', "未知")
                    log.info(f"response_data.to_dict(): {response_data.to_dict()}")

                    status = 2

                    # 推送
                    url = org_data.intent_callback
                    if url:
                        headers = {
                            "Content-Type": "application/json"
                        }
                        data = {
                            "internal_id": record_data.id,
                            "external_id": record_data.external_id,
                            "score": llm_intent,
                            "intent": score
                        }

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

                    # 存储结果
                    status = 2
                    async with async_db_session.begin() as db:
                        try:
                            await intent_records_dao.update_llm_intent(db, record_data.id, llm_intent,
                                                                       {"messages": messages},
                                                                       response_data.to_dict(),
                                                                       status)
                        except Exception as e:
                            log.error(f"更新意图记录时发生异常：{e}")



def generate_json(api_key: str, openai_base: str, messages: list[dict], json_schema: dict):
    try:
        client_args = {}
        if api_key:
            client_args["api_key"] = api_key
        if openai_base:
            client_args["base_url"] = openai_base

        oai_client = OpenAI(**client_args)

        completion = oai_client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            response_format={
                "type": "json_schema",
                "json_schema": json_schema
            }
        )
        if completion and isinstance(completion.choices, list) and len(completion.choices) > 0:
            first_choice = completion.choices[0]
            if first_choice and first_choice.message:
                # return first_choice.message.content
                return completion
    except Exception as e:
        log.error(f"[oai] generate_json failed: {e}")

async def send_request_with_retry(url: str, data: Dict, headers: Dict[str, str], max_retries: int, delay_between_retries: int) -> bool:
    for attempt in range(max_retries):
        try:
            async with session.ClientSession() as session:
                async with session.post(url, json=data, headers=headers, timeout=10) as response:
                    if response.status == 200:
                        return True
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            log.error(f"请求异常：{e}")

        if attempt < max_retries - 1:
            print("重试中...")
            await asyncio.sleep(delay_between_retries)
    return False