A spoken English education chatbot based on ChatGPT/whsiper and gTTS.
from flask import Flask
from config import Config
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
import logging

db = SQLAlchemy()
migrate = Migrate()


def create_app(config_class=Config):
    app = Flask(__name__)
    app.config['TEMPLATES_AUTO_RELOAD'] = True
    app.config.from_object(config_class)

    handler = logging.FileHandler('app.log')  # 日志输出到一个文件
    handler.setLevel(logging.DEBUG)  # 设置日志级别为DEBUG
    app.logger.addHandler(handler)

    db.init_app(app)
    migrate.init_app(app, db)

    from app.open_ai import bp as open_ai_bp
    app.register_blueprint(open_ai_bp)

    from app.tts import bp as tts_bp
    app.register_blueprint(tts_bp)

    from app.main import bp as main_bp
    app.register_blueprint(main_bp)

    return app


from app import models
app/__init__.py
viking-man/IntroventsEnglishCorner
import os.path

from flask import Flask, render_template, request, jsonify, send_file
from app.open_ai import chatgpt_proxy
from app.tts import gtts_proxy
from app.models import Audio
from app.open_ai import whisper_proxy
from app.main import bp
import io
import time
from app import create_app
from pathlib import Path

app = create_app()


@bp.route('/')
@bp.route('/index')
def index():
    return render_template('index.html')


@bp.route('/text_chat', methods=['POST'])
def text_chat():  # put application's code here
    print(str(request))
    # 获取参数
    user_message = request.form['user_message']
    user_id = request.form['user_id']
    chat_id = request.form['chat_id']
    if chat_id is None or chat_id == 'undefined' or len(chat_id) == 0:
        chat_id = str(abs(hash(user_message)))
    gpt_version = request.form['gpt_version']

    # chatgpt请求
    ai_message, chat_history = chatgpt_proxy.text_chat(user_message, user_id, chat_id, gpt_version)

    # tts转成语音
    audio_id = gtts_proxy.convert_to_audio(user_id, chat_id, ai_message, "en")

    # 返回信息
    return jsonify({
        'ai_message': ai_message,
        'chat_history': chat_history,
        'audio_id': audio_id,
        'chat_id': chat_id
    })


@bp.route('/get_audio', methods=['POST'])
def receive_audio():
    audio_id = request.form['audio_id']
    audio = Audio.query.get_or_404(audio_id)

    return send_file(
        io.BytesIO(audio.audio_data),
        mimetype="audio/wav"  # 设置适当的MIME类型
    )


@bp.route('/speech_chat', methods=['POST'])
def speech_chat():
    print(str(request))
    user_id = request.form.get('user_id')
    chat_id = request.form.get('chat_id')
    gpt_version = request.form.get('gpt_version')

    audio_file = request.files['audio']
    # Save the audio file (optional)
    audio_path = str(
        Path("app/files/input", user_id + "_" + chat_id + "_" + str(time.time()) + ".wav").absolute())
    audio_file.save(audio_path)

    chat_text = whisper_proxy.transcribe_to_text(audio_path)

    if chat_id is None or chat_id == 'undefined' or len(chat_id) == 0:
        chat_id = str(abs(hash(chat_text)))

    # chatgpt请求
    ai_message, chat_history = chatgpt_proxy.text_chat(chat_text, user_id, chat_id, gpt_version)

    # tts转成语音
    audio_id = gtts_proxy.convert_to_audio(user_id, chat_id, ai_message, "en")

    return jsonify({
        'user_message': chat_text,
        'ai_message': ai_message,
        'audio_id': audio_id,
        'chat_id': chat_id})


@bp.route('/clear_history', methods=['POST'])
def clear_history():
    data = request.get_json()
    userid = data.get('userId')
    print(data)
    return jsonify({'message': 'History cleared successfully'})

# @bp.before_request
# def log_request_info():
#     app.logger.debug('Headers: %s', request.headers)
#     app.logger.debug('Body: %s', request.get_data())
#
#
# @bp.after_request
# def log_response_info(response):
#     app.logger.debug('Headers: %s', response.headers)
#     app.logger.debug('Body: %s', response.get_data())
#     return response

app/main/routes.py
viking-man/IntroventsEnglishCorner
from datetime import datetime
from app import db


class Chat(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.String(80), unique=False, nullable=False)
    chat_id = db.Column(db.String(80), unique=False, nullable=False)
    chat_history = db.Column(db.Text, unique=False, nullable=False)

    # audio_id = db.Column(db.Integer, db.ForeignKey('audio.id'), nullable=True)
    audios = db.relationship('Audio',
                             backref=db.backref('chat', lazy=True))

    def __repr__(self):
        return '<Chat {}>'.format(self.chat_id)


class Audio(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    chat_id = db.Column(db.Integer, db.ForeignKey('chat.id'))
    type = db.Column(db.String(10), unique=False, nullable=False)
    audio_data = db.Column(db.LargeBinary, unique=False, nullable=False)

app/models.py
viking-man/IntroventsEnglishCorner
from flask import Blueprint

bp = Blueprint('open_ai', __name__)

from app.open_ai import chatgpt_proxy, whisper_proxy, gpt_prompt, whisper_model

app/open_ai/__init__.py
viking-man/IntroventsEnglishCorner
from app.open_ai import gpt_prompt
from app.type import ChatGPTModel
from app.models import Chat, Audio
import logging

import openai
from app import db
import json


def text_chat(user_message, user_id, chat_id, gpt_version):
    gpt_model = ChatGPTModel.get_value(gpt_version)
    # 查询聊天记录
    chat_record = Chat.query.filter_by(chat_id=chat_id).filter_by(user_id=user_id).first()
    if chat_record is None:
        chat_history = []
    else:
        chat_history = json.loads(chat_record.chat_history)

    # request openai
    ai_message, chat_history = requestOpenai(user_message, chat_history, gpt_model)

    chat_history_json = json.dumps(chat_history)
    # 更新数据库
    if chat_record is None:
        chat_record = Chat(user_id=user_id, chat_id=chat_id, chat_history=chat_history_json)
        db.session.add(chat_record)
        logging.info(f'Add chat record->{str(chat_record)}')
        print(str(chat_record))
    else:
        chat_record.chat_history = chat_history_json
        logging.info(f'Update chat record->{str(chat_record)}')
    db.session.commit()

    return ai_message, chat_history


def requestOpenai(chat_context, chat_history, model):
    # 第一次请求，带上英语口语老师对应的prompt

    if len(chat_history) == 0:
        chat_history.append({"role": "system",
                             "content": gpt_prompt.english_teacher_prompt})
    chat_history.append({"role": "user",
                         "content": chat_context})
    print(f'openai chat request->{chat_history}')
    response = openai.ChatCompletion.create(
        model=model,
        messages=chat_history,
        temperature=0
    )
    print(f'openai chat response->{str(response)}')
    content_ = response.choices[0]['message']['content']
    chat_history.append({"role": "assistant", "content": content_})
    return content_, chat_history

app/open_ai/chatgpt_proxy.py
viking-man/IntroventsEnglishCorner
# 小学英语口语老师
english_primary_teacher_prompt = ("You are an elementary school English speaking teacher. "
                                  "I'm an elementary school student conversing with you. "
                                  "At the beginning of each of your responses, I'd like you to first correct the student's grammatical errors and offer a more everyday spoken expression. "
                                  "Following that, based on the context and the most recent content of the student's dialogue, you should reply to progress the conversation.")

# 成人口语老师
english_adult_teacher_prompt = (
    "You are an adult English speaking teacher, dedicated to assisting students in improving their spoken English skills. "
    "I am a student conversing with you. "
    "At the beginning of each response, I'd like you to correct any of my English grammar, structure, or other relevant mistakes, and suggest a more contextually fitting colloquial expression. "
    "After correcting my errors, you should reply to me based on the previous context and my latest response to continue the conversation.")

english_teacher_prompt = ("I want you to act as a spoken English teacher and improver. "
                          "I will speak to you in English and you will reply to me in English to practice my spoken English. "
                          "I want you to keep your reply neat, limiting the reply to 80 words. "
                          "I want you to strictly point out and correct my grammar mistakes, typos, and factual errors at the beginning of the conversation for every sentence."
                          # "I want you to ask me a question in your reply. Now let's start practicing, you could ask me a question first. "
                          "After you correct my mistakes,I want you reply to me in English to continue the conversation. "
                          "Remember, you should reply to me in English to continue the conversation.")

app/open_ai/gpt_prompt.py
viking-man/IntroventsEnglishCorner
import whisper
import time
import logging


class WhisperModel():
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        self.device = None
        self.whisper_model = None

    def load_model(self):
        self.whisper_model = whisper.load_model("small", "cpu")

    def transcribe(self, audio, lang):
        tic = time.time()

        res = self.whisper_model.transcribe(
            audio,
            task="transcribe",
            language=lang,
            verbose=True
        )

        logging.info(f"Done transcription in {time.time() - tic:.1f} sec")
        return res

app/open_ai/whisper_model.py
viking-man/IntroventsEnglishCorner
from whisper import audio
from app.open_ai.whisper_model import WhisperModel
from app.models import Chat, Audio


def transcribe_to_text(audio_path):
    load_audio = audio.load_audio(audio_path)
    whisper_model = WhisperModel()
    whisper_model.load_model()
    transcribe_result = whisper_model.transcribe(load_audio, None)
    print(f'Transcribe result->{str(transcribe_result)}')

    chat_text = ""
    segments_ = transcribe_result['segments']
    for segment in segments_:
        chat_text = chat_text + segment['text'] + ","
    chat_text = chat_text[:-1]

    return chat_text

app/open_ai/whisper_proxy.py
viking-man/IntroventsEnglishCorner
from flask import Blueprint

bp = Blueprint('tts', __name__)

from app.tts import gtts_proxy

app/tts/__init__.py
viking-man/IntroventsEnglishCorner
import os.path

from gtts import gTTS
from app.models import Chat, Audio
import time
from app.type import AudioType
from app import db
from pathlib import Path


def convert_to_audio(user_id, chat_id, text_content, lang):
    # 定时任务清理
    tts = gTTS(text_content, lang=lang)
    audio_filename = str(Path("app/files/output",
                              user_id + "_" + chat_id + "_" + str(int(time.time())) + ".wav").absolute())
    tts.save(audio_filename)

    with open(audio_filename, 'rb') as f:
        audio_data = f.read()

    audio = Audio(chat_id=chat_id, type=AudioType.INPUT.name, audio_data=audio_data)
    db.session.add(audio)
    db.session.commit()

    return audio.id

app/tts/gtts_proxy.py
viking-man/IntroventsEnglishCorner
from enum import Enum


class ChatGPTModel(Enum):
    GPT4 = "gpt-4-0613"
    GPT3 = "gpt-3.5-turbo-0613"

    @staticmethod
    def get_values():
        return [i.value for i in ChatGPTModel]

    @staticmethod
    def get_value(version):
        for model in ChatGPTModel:
            if model.name == version:
                return model.value
            return ChatGPTModel.GPT3.value


class AudioType(Enum):
    INPUT = 'input',
    OUTPUT = 'output'

    @staticmethod
    def get_values():
        return [i.value for i in AudioType]

app/type.py
viking-man/IntroventsEnglishCorner
import pytest
import time
from app.type import AudioType


def test_hash():
    message = "how to learn English?"
    s = str(hash(str(len(message))))
    print(s)

    print(str(int(time.time())))

    print(abs(hash(message)))


test.py
viking-man/IntroventsEnglishCorner