使用阿里语音转文字，使用openai判断用户意图

app/gpt/api/route.py

@@ -10,6 +10,8 @@ from starlette.responses import FileResponse, StreamingResponse
 
 from common.log import log
 
+from app.gpt.service.intent import get_intent
+
 router = APIRouter()
 
 class Result(BaseModel):
@@ -48,4 +50,9 @@ async def record_file(filepath:str):
 async def user_intent(body:dict):
     log.info(json.dumps(body,indent=4))
 
+    if 'notify' in body:
+        intent = await get_intent(body)
+    else:
+        return Response(result=Result(error=1, msg="notify not found"))
+
     return Response()

app/gpt/service/ali_filetrans.py

@@ -0,0 +1,106 @@
+import json
+import os
+import time
+from aliyunsdkcore.acs_exception.exceptions import ClientException
+from aliyunsdkcore.acs_exception.exceptions import ServerException
+from aliyunsdkcore.client import AcsClient
+from aliyunsdkcore.request import CommonRequest
+
+from common.log import log
+
+
+def file_trans(fileLink):
+    # 地域ID，固定值。
+    REGION_ID = "cn-beijing"
+    PRODUCT = "nls-filetrans"
+    DOMAIN = "filetrans.cn-beijing.aliyuncs.com"
+    API_VERSION = "2018-08-17"
+    POST_REQUEST_ACTION = "SubmitTask"
+    GET_REQUEST_ACTION = "GetTaskResult"
+    # 请求参数
+    KEY_APP_KEY = "appkey"
+    KEY_FILE_LINK = "file_link"
+    KEY_VERSION = "version"
+    KEY_ENABLE_WORDS = "enable_words"
+    # 是否开启智能分轨
+    KEY_AUTO_SPLIT = "auto_split"
+    # 响应参数
+    KEY_TASK = "Task"
+    KEY_TASK_ID = "TaskId"
+    KEY_STATUS_TEXT = "StatusText"
+    KEY_RESULT = "Result"
+    # 状态值
+    STATUS_SUCCESS = "SUCCESS"
+    STATUS_RUNNING = "RUNNING"
+    STATUS_QUEUEING = "QUEUEING"
+    # 创建AcsClient实例
+    client = AcsClient(os.getenv('ALIYUN_ACCESS_KEY_ID'), os.getenv('ALIYUN_ACCESS_SECRET'), REGION_ID)
+    # 提交录音文件识别请求
+    postRequest = CommonRequest()
+    postRequest.set_domain(DOMAIN)
+    postRequest.set_version(API_VERSION)
+    postRequest.set_product(PRODUCT)
+    postRequest.set_action_name(POST_REQUEST_ACTION)
+    postRequest.set_method('POST')
+    # 新接入请使用4.0版本，已接入（默认2.0）如需维持现状，请注释掉该参数设置。
+    # 设置是否输出词信息，默认为false，开启时需要设置version为4.0。
+    # task = {KEY_APP_KEY: "IET8NhqIaLoPzdet", KEY_FILE_LINK: fileLink, KEY_VERSION: "4.0", KEY_ENABLE_WORDS: False}
+    # 开启智能分轨，如果开启智能分轨，task中设置KEY_AUTO_SPLIT为True。
+    task = {KEY_APP_KEY : os.getenv('ALIYUN_APP_KEY'), KEY_FILE_LINK : fileLink, KEY_VERSION : "4.0", KEY_ENABLE_WORDS : False, KEY_AUTO_SPLIT : True}
+    task = json.dumps(task)
+
+    postRequest.add_body_params(KEY_TASK, task)
+    taskId = ""
+    try:
+        postResponse = client.do_action_with_exception(postRequest)
+        postResponse = json.loads(postResponse)
+
+        log.info("录音文件识别任务：" + json.dumps(postResponse,indent=4))
+
+        statusText = postResponse[KEY_STATUS_TEXT]
+        if statusText == STATUS_SUCCESS:
+            taskId = postResponse[KEY_TASK_ID]
+            log.info("录音文件识别请求成功响应！任务ID：" + str(taskId))
+        else:
+            log.error("录音文件识别任务失败！", statusText)
+            return False
+    except ServerException as e:
+        log.error("录音文件识别失败！ 服务端异常", e)
+        return False
+    except ClientException as e:
+        log.error("录音文件识别失败！ 客户端异常", e)
+        return False
+    # 创建CommonRequest，设置任务ID。
+    getRequest = CommonRequest()
+    getRequest.set_domain(DOMAIN)
+    getRequest.set_version(API_VERSION)
+    getRequest.set_product(PRODUCT)
+    getRequest.set_action_name(GET_REQUEST_ACTION)
+    getRequest.set_method('GET')
+    getRequest.add_query_param(KEY_TASK_ID, taskId)
+    # 提交录音文件识别结果查询请求
+    # 以轮询的方式进行识别结果的查询，直到服务端返回的状态描述符为"SUCCESS"、"SUCCESS_WITH_NO_VALID_FRAGMENT"，
+    # 或者为错误描述，则结束轮询。
+    statusText = ""
+    while True:
+        try:
+            getResponse = client.do_action_with_exception(getRequest)
+            getResponse = json.loads(getResponse)
+            log.info("录音文件识别结果：" + json.dumps(getResponse,indent=4))
+            statusText = getResponse[KEY_STATUS_TEXT]
+            if statusText == STATUS_RUNNING or statusText == STATUS_QUEUEING:
+                # 继续轮询
+                time.sleep(5)
+            else:
+                # 退出轮询
+                break
+        except ServerException as e:
+            log.error("录音文件识别查询失败！ 服务端异常", e)
+        except ClientException as e:
+            log.error("录音文件识别查询失败！ 客户端异常", e)
+    if statusText == STATUS_SUCCESS:
+        log.info("录音文件识别结束！成功")
+        return getResponse
+    else:
+        log.error("录音文件识别结束！失败", statusText)
+        return False

app/gpt/service/intent.py

@@ -0,0 +1,62 @@
+import json
+
+from matplotlib.font_manager import json_dump
+
+import model.records
+from app.gpt.service.ali_filetrans import file_trans
+from app.gpt.service.openai import openai_check_intent
+from common.log import log
+from database.db_mysql import async_db_session
+
+async def get_intent(body: dict):
+    notify = body['notify']
+
+    record = model.records.Records(
+        body=body,
+        staff_no=str(notify['staffNo']),
+        caller=str(notify['caller']),
+        callee=str(notify['callee']),
+        record_file=str(notify['recordFile']),
+        session=str(notify['session']),
+    )
+
+    mp3url = "https://toolsapi.gkscrm.com/api/v1/gpt/recordfile?filepath=" + notify['recordFile']
+
+    get_ali_response = file_trans(mp3url)
+
+    if not get_ali_response:
+        log.error("ali_trans error")
+        return False
+
+    if "Result" not in get_ali_response:
+        log.error("ali_trans response not Result")
+        return False
+
+    record.ali_trans = get_ali_response
+    record.ali_task_id = get_ali_response["TaskId"]
+
+    sentences = get_ali_response["Result"]["Sentences"]
+
+
+    chats = ""
+    for sentence in sentences:
+        chats += "speaker-" + sentence["SpeakerId"] + ": " + sentence["Text"] + "\n"
+
+    log.info("聊天记录如下：\n" + chats)
+
+    record.chats = chats
+
+
+
+    user_intent = openai_check_intent(chats)
+
+    record.user_intent = user_intent.user_intent
+    record.accept = user_intent.accept
+
+
+    async with async_db_session() as db:
+        db.add(record)
+        await db.commit()
+
+
+    return True

app/gpt/service/openai.py

@@ -0,0 +1,34 @@
+import os
+from typing import Optional
+
+import openai
+from openai import OpenAI
+from pydantic import BaseModel, Field
+
+from common.log import log
+
+client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_BASE_URL"))
+
+class UserIntent(BaseModel):
+    user_intent: Optional[str] = Field(description="判断的用户意图")
+    accept:Optional[bool] = Field(description="用户是否接受，是的话为 True，否的话 False")
+
+
+def openai_check_intent(chat:str):
+    try:
+        completion = client.beta.chat.completions.parse(
+            model="gpt-4o-2024-08-06",
+            messages=[
+                {"role": "system", "content": "你是一位经验丰富的电话录音质检员，你非常擅长通过通话记录，判断用户意图"},
+                {"role": "user", "content": chat},
+            ],
+            response_format=UserIntent,
+        )
+        math_reasoning  = completion.choices[0].message
+        if math_reasoning.refusal:
+            return UserIntent(user_intent=math_reasoning.refusal, accept=False)
+        else:
+            return math_reasoning.parsed
+    except Exception as e:
+        log.error(e)
+        return UserIntent(user_intent="openai请求失败", accept=False)

core/conf.py

@@ -3,6 +3,7 @@
 from functools import lru_cache
 from typing import Literal
 
+from dotenv import load_dotenv
 from pydantic import model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -145,7 +146,10 @@ class Settings(BaseSettings):
 
 @lru_cache
 def get_settings() -> Settings:
+
+
     """获取全局配置"""
+    load_dotenv()
     return Settings()
 
 

database/db_mysql.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 import sys
 
 from typing import Annotated
@@ -21,7 +19,7 @@ def create_engine_and_session(url: str | URL):
         engine = create_async_engine(url, echo=settings.MYSQL_ECHO, future=True, pool_pre_ping=True)
         # log.success('数据库连接成功')
     except Exception as e:
-        log.error('❌ 数据库链接失败 {}', e)
+        log.error('数据库链接失败 {}', e)
         sys.exit()
     else:
         db_session = async_sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
@@ -30,7 +28,7 @@ def create_engine_and_session(url: str | URL):
 
 SQLALCHEMY_DATABASE_URL = (
     f'mysql+asyncmy://{settings.MYSQL_USER}:{quote(settings.MYSQL_PASSWORD)}@{settings.MYSQL_HOST}:'
-    f'{settings.MYSQL_PORT}/{settings.MYSQL_DATABASE}?charset={settings.MYSQL_CHARSET}&ssl=true'
+    f'{settings.MYSQL_PORT}/{settings.MYSQL_DATABASE}?charset={settings.MYSQL_CHARSET}'
 )
 
 async_engine, async_db_session = create_engine_and_session(SQLALCHEMY_DATABASE_URL)
@@ -60,4 +58,4 @@ async def create_table():
 
 def uuid4_str() -> str:
     """数据库引擎 UUID 类型兼容性解决方案"""
-    return str(uuid4())
+    return str(uuid4())