app.py

import gradio as gr
import os
import openai
from openai import OpenAI
import logging
from concurrent.futures import ThreadPoolExecutor
logging.basicConfig(level=logging.DEBUG)
os.environ["OPENAI_API_KEY"] = os.environ["gptkey"]
###
chat_history = []
client = OpenAI(
    api_key= os.environ["gptkey"]
)

import json
chat_question_index = 0
chat_history = [[None,"你好，我是訪談小助手，我將透過陪你聊天的方式，了解你的基本資料！首先，可以告訴我你任職的品牌嗎？像是Amplifi/Carat/dentsu MB/dentsu One/dentsu X/dentsu Consult/HQ/iProspect/isobar/Merkle/dentsu Solution Center，這裡面的哪一個呢？"]]
chat_history_message = []

# 問題清單
questions = [
    {"id": 0, "content": "受訪者的任職的品牌", "keywords": ["品牌"], "answered": False},
    {"id": 1, "content": "受訪者在集團服務的總年資（不同單位年資可合併計算）", "keywords": ["年資"], "answered": False},
    {"id": 2, "content": "受訪者服務的客戶產業(如交通工具類/零售通路類/美妝與美容美髮類/醫療保健類/民生消費品類/電商與網路服務類/財務金融保險類/煙酒類/時尚精品類/家電類/電信通訊類/電腦與周邊設備類/休閒娛樂類/遊戲&APP類/其他)", "keywords": ["產業"], "answered": False},
    {"id": 3, "content": "受訪者使用dentsu pro的頻率(如每周幾次、每月幾次、每季幾次或每半年幾次)", "keywords": ["使用頻率"], "answered": False},
    {"id": 4, "content": "受訪者最常進入dentsu pro的起點(選項為桌面捷徑,電子報通知,My Apps連結,自行儲存網址(如我的最愛))", "keywords": ["入口方式"], "answered": False}
]

check_json_schema = {
    "name": "CheckSchema",  # 添加 name 字段
    "description": "檢查使用者是否有回答問題",
    "strict": True,
    "schema": {  # 修正，將之前的 Schema 放入 schema 字段
        "type": "object",
        "properties":{
            "question_0": {
                "type": "boolean",
                "description": "第一題「受訪者的任職的品牌」是否有被充分回答？(答案需要是Amplifi/Carat,凱絡/dentsu MB/dentsu One/dentsu X,貝立德/dentsu Consult/HQ/iProspect/isobar/Merkle/dentsu Solution Center,DSC其一)"
            },
            "question_1": {
                "type": "boolean",
                "description": "第二題「受訪者在集團服務的總年資」是否有被充分回答？"
            },
            "question_2": {
                "type": "boolean",
                "description": "第三題「受訪者服務的客戶產業」是否有被充分回答？"
            },
            "question_3": {
                "type": "boolean",
                "description": "第四題「受訪者使用dentsu pro的頻率」是否有被充分回答？充分的定義：答案能被區分在每周幾次、每月幾次、每季幾次或每半年幾次。"
            },
            "question_4": {
                "type": "boolean",
                "description": "第五題「受訪者最常進入dentsu pro的起點」是否有被充分回答？充分的定義：答案限制在(桌面捷徑,電子報通知,My Apps連結,自行儲存網址(如我的最愛))。"

            } 
         }
        ,"required": [
            "question_0",
            "question_1",
            "question_2",
            "question_3",
            "question_4"
        ],
        "additionalProperties": False
    }
}

analyze_json_schema = {
    "name": "AnalyzeSchema",  # 添加 name 字段
    "description": "紀錄使用者回答的結果",
    "strict": True,
    "schema": {  # 修正，將之前的 Schema 放入 schema 字段
        "type": "object",
        "properties": {
            "brand": {
                "type": "string",
                "description": "受訪者的任職的品牌"
            },
            "seniority": {
                "type": "string",
                "description": "受訪者在集團服務的總年資"
            },
            "service_industry": {
                "type": "string",
                "description": "受訪者服務的客戶產業"
            },
            "frequency": {
                "type": "string",
                "description": "受訪者使用dentsu pro的頻率"
            },
            "portal": {
                "type": "string",
                "description": "受訪者最常進入dentsu pro的起點"

            },
            "interviewee_traits": {
                "type": "string",
                "description": "根據以上受訪者與AI的對話過程，分析出三個受訪者的性格特質"

            }
        }
        ,"required": [
            "brand",
            "seniority",
            "service_industry",
            "frequency",
            "portal",
            "interviewee_traits"
        ],
        "additionalProperties": False
    }
}

def clear():
    print("CLEEEEEEEEEEEEEEEER")

    
def chatbot_system(user_input):
    with ThreadPoolExecutor(max_workers=3) as executor:
        fun_1 = executor.submit(analyze_user_input, user_input)
        isAnswered = fun_1.result()
        #會有一個true/fales
        fun_2 = executor.submit(generate_question, isAnswered )
        assistant_output = fun_2.result()
    
        chat_history.append([user_input, assistant_output])
        chat_history_message.extend([{"role": "user", "content": user_input},{"role": "assistant", "content": assistant_output}])

        demand = ""
        
        if chat_question_index is None:
            print("ERROR:Noneeeeeeeeeeeee")
            dialogue = format_chat_history(chat_history)       
            fun_3 = executor.submit(analyze_dialogue, dialogue )
            demand = fun_3.result()
        
    return chat_history,demand
    
def format_chat_history(chat_history):
    formatted_chat = ""
    for user_speech, ai_speech in chat_history:
        if user_speech is not None:
            formatted_chat += f"使用者: \"{user_speech}\"\n"
        formatted_chat += f"AI助手: \"{ai_speech}\"\n"
    return formatted_chat
    
def generate_question(now_question_isAnswered):
    
    messages_base = [
        {"role": "system", "content": "你是一個訪談助手，專門協助問卷訪談。其角色是透過輕鬆的對話一步一步引導受訪者了解他們的資訊，就像一位真實的訪談者。"}
    ]
    messages_base.extend(chat_history_message)
    
    completed_text = ""
    
    if(now_question_isAnswered):
        #檢查是否全部都回答完
        chat_question = find_next_unanswered_question()
        global chat_question_index
        #if回答完:結束
        if chat_question is None:
            chat_question_index = None
            print("所有問題均已回答。")

            completed_text = "我已經釐清你的基本資訊了，感謝你這次的作答。"
        #沒回答完，跳到下一題未回答問題
        #請根據目前的聊天內容，生成後續的對話，這段對話的主旨在於了解「」
        else:
#            print("next_q")
#            print(chat_question)
            
            chat_question_content = chat_question['content']
            chat_question_index = chat_question['id']
            
#            print(f"下一個未回答的問題主指是：{chat_question_content}")
            messages_base.extend([{"role": "user", "content": f"""請你延續先前的對話與使用者回答的內容，根據我現在想要了解使用者的問題:{chat_question_content}，生成一句
            生動的提問，必須使用繁體中文。只要給我那個問題的內容就好，不要附加其他文字。"""}])
        
            response = client.chat.completions.create(
                model='gpt-4o',
                max_tokens=4096,
                temperature=0.5,
                messages= messages_base)

            completed_text = response.choices[0].message.content
            
    else:
        completed_text = "抱歉，你好像沒有回答到我的問題呢，要不要多補充一點內容呢？"

    return completed_text

def find_next_unanswered_question():
    for question in questions:
        if not question['answered']:
            return question
    return None
    
def analyze_user_input(user_input):
    print("analyze_user_input")
    messages_base = [
        {"role": "system", "content": "你是一個回覆檢查員，你會根據收到的回覆，進行回覆檢核。"},
        {"role": "user", "content": f"使用者輸入的內容如下：{user_input}"}
     ]

    response = client.chat.completions.create(
        model='gpt-4o',
        messages= messages_base,
        temperature=0.5,
        response_format={
                "type": "json_schema",
                "json_schema": check_json_schema
        }
    )
    arguments_str = response.choices[0].message.content
    arguments = json.loads(arguments_str)
    
    update_questions(questions, arguments)

    #從json_schema中得到當前問題是否有被回答
    question_key = f"question_{chat_question_index}"
    print("question_key:")
    print(question_key)
    isAnswered = arguments[question_key]
 
    return isAnswered

def analyze_dialogue(dialogue):
    messages_base = [
        {"role": "system", "content": "你是一個對話檢查員，你會根據收到的對話紀錄，進行回覆分析。"},
        {"role": "user", "content": f"整段對話如下：{dialogue}"}
     ]

    response = client.chat.completions.create(
        model='gpt-4o',
        messages= messages_base,
        temperature=0.5,
        response_format={
                "type": "json_schema",
                "json_schema": analyze_json_schema
        }
    )
    arguments_str = response.choices[0].message.content

    return arguments_str

def update_questions(questions, validation_results):
    # 遍歷檢核結果的每一個項目
    for key, value in validation_results.items():
        # 從key中提取問題的id
        question_id = int(key.split('_')[1])
        # 尋找對應id的問題
        question = next((q for q in questions if q['id'] == question_id), None)
        if question:
            if value:
                # 如果檢核為True，更新答題狀態為True
                question['answered'] = True


    

#------------------------------------------------------------
#歡迎句，問使用者可以開始了嗎？
#開始
#從問題清單中，找一個還沒有回答的問題，生成問題(函式一)，id=1
#接收使用者的輸入
#解析使用者的輸入(函式二)
#1.是否有回答id=1的問題？答案是多少；是否有回答其他問題？答案是多少
#如果使用者有回答問題清單任一問題，填入答案，生成下一個問題，如果沒有
#如果使用者沒有回答題，繼續跟使用者進行互動，請使用者回答本題

demo = gr.Interface(
    fn=chatbot_system,
    inputs=[
        gr.Textbox(label="對話")
    ],
    outputs=[
        gr.Chatbot(label="Perosna",height=300,value=chat_history),
        gr.Textbox(label="對話陣列")
    ],
    title="訪談助手",
    description="透過與AI聊天，完成dentsu pro種子代表調查。",
    allow_flagging="never", )

demo.launch(share=True)