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)