Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import json | |
| import requests | |
| class Chatbot: | |
| def __init__(self, config): | |
| self.video_id = config.get('video_id') | |
| self.content_subject = config.get('content_subject') | |
| self.content_grade = config.get('content_grade') | |
| self.jutor_chat_key = config.get('jutor_chat_key') | |
| self.transcript_text = self.get_transcript_text(config.get('trascript')) | |
| self.ai_name = config.get('ai_name') | |
| self.ai_client = config.get('ai_client') | |
| def get_transcript_text(self, transcript_data): | |
| transcript_json = json.loads(transcript_data) | |
| for entry in transcript_json: | |
| entry.pop('embed_url', None) | |
| entry.pop('screenshot_path', None) | |
| transcript_text = json.dumps(transcript_json, ensure_ascii=False) | |
| return transcript_text | |
| def chat(self, user_message, chat_history, socratic_mode=False, service_type='jutor'): | |
| messages = self.prepare_messages(chat_history, user_message) | |
| system_prompt = self.prepare_system_prompt(socratic_mode) | |
| if service_type in ['jutor', 'groq', 'claude3']: | |
| response_text = self.chat_with_service(service_type, system_prompt, messages) | |
| return response_text | |
| else: | |
| raise gr.Error("不支持此服務") | |
| def prepare_system_prompt(self, socratic_mode): | |
| content_subject = self.content_subject | |
| content_grade = self.content_grade | |
| video_id = self.video_id | |
| trascript_text = self.transcript_text | |
| socratic_mode = str(socratic_mode) | |
| ai_name = self.ai_name | |
| system_prompt = f""" | |
| 科目：{content_subject} | |
| 年級：{content_grade} | |
| 逐字稿資料：{trascript_text} | |
| ------------------------------------- | |
| 你是一個專業的{content_subject}老師， user 為{content_grade}的學生 | |
| socratic_mode = {socratic_mode} | |
| if socratic_mode is True， | |
| - 請用蘇格拉底式的提問方式，引導學生思考，並且給予學生一些提示 | |
| - 一次只問一個問題，字數在100字以內 | |
| - 不要直接給予答案，讓學生自己思考 | |
| - 但可以給予一些提示跟引導，例如給予影片的時間軸，讓學生自己去找答案 | |
| if socratic_mode is False， | |
| - 直接回答學生問題，字數在100字以內 | |
| rule: | |
| - 請一定要用繁體中文回答 zh-TW，並用台灣人的口語表達，回答時不用特別說明這是台灣人的語氣，也不用說這是「台語的說法」 | |
| - 不用提到「逐字稿」這個詞 | |
| - 如果學生問了一些問題你無法判斷，請告訴學生你無法判斷，並建議學生可以問其他問題 | |
| - 或者你可以反問學生一些問題，幫助學生更好的理解資料，字數在100字以內 | |
| - 如果學生的問題與資料文本無關，請告訴學生你「無法回答超出影片範圍的問題」，並告訴他可以怎麼問什麼樣的問題（一個就好） | |
| - 只要是參考逐字稿資料，請在回答的最後標註【參考資料：（分）：（秒）】 | |
| - 回答範圍一定要在逐字稿資料內，不要引用其他資料，請嚴格執行 | |
| - 並在重複問句後給予學生鼓勵，讓學生有學習的動力 | |
| - 請用 {content_grade} 的學生能懂的方式回答 | |
| - 回答時數學式請用數學符號代替文字（Latex 用 $ 字號 render) | |
| """ | |
| return system_prompt | |
| def prepare_messages(self, chat_history, user_message): | |
| messages = [] | |
| if chat_history is not None: | |
| if len(chat_history) > 10: | |
| chat_history = chat_history[-10:] | |
| for user_msg, assistant_msg in chat_history: | |
| if user_msg: | |
| messages.append({"role": "user", "content": user_msg}) | |
| if assistant_msg: | |
| messages.append({"role": "assistant", "content": assistant_msg}) | |
| if user_message: | |
| user_message += "/n (請一定要用繁體中文回答 zh-TW，並用台灣人的禮貌口語表達，回答時不要特別說明這是台灣人的語氣，不用提到「逐字稿」這個詞，用「內容」代替)，回答時請用數學符號代替文字（Latex 用 $ 字號 render)" | |
| messages.append({"role": "user", "content": user_message}) | |
| return messages | |
| def chat_with_service(self, service_type, system_prompt, messages): | |
| if service_type == 'jutor': | |
| return self.chat_with_jutor(system_prompt, messages) | |
| elif service_type == 'groq': | |
| return self.chat_with_groq(system_prompt, messages) | |
| elif service_type == 'claude3': | |
| return self.chat_with_claude3(system_prompt, messages) | |
| else: | |
| raise gr.Error("不支持的服务类型") | |
| def chat_with_jutor(self, system_prompt, messages): | |
| messages.insert(0, {"role": "system", "content": system_prompt}) | |
| api_endpoint = "https://ci-live-feat-video-ai-dot-junyiacademy.appspot.com/api/v2/jutor/hf-chat" | |
| headers = { | |
| "Content-Type": "application/json", | |
| "x-api-key": self.jutor_chat_key, | |
| } | |
| data = { | |
| "data": { | |
| "messages": messages, | |
| "max_tokens": 512, | |
| "temperature": 0.9, | |
| "model": "gpt-4-1106-preview", | |
| "stream": False, | |
| } | |
| } | |
| response = requests.post(api_endpoint, headers=headers, data=json.dumps(data)) | |
| response_data = response.json() | |
| response_completion = response_data['data']['choices'][0]['message']['content'].strip() | |
| return response_completion | |
| def chat_with_groq(self, system_prompt, messages): | |
| # system_prompt insert to messages 的最前面 {"role": "system", "content": system_prompt} | |
| messages.insert(0, {"role": "system", "content": system_prompt}) | |
| request_payload = { | |
| "model": "mixtral-8x7b-32768", | |
| "messages": messages, | |
| "max_tokens": 1000 # 設定一個較大的值，可根據需要調整 | |
| } | |
| groq_client = self.ai_client | |
| response = groq_client.chat.completions.create(**request_payload) | |
| response_completion = response.choices[0].message.content.strip() | |
| return response_completion | |
| def chat_with_claude3(self, system_prompt, messages): | |
| if not system_prompt.strip(): | |
| raise ValueError("System prompt cannot be empty") | |
| model_id = "anthropic.claude-3-sonnet-20240229-v1:0" | |
| # model_id = "anthropic.claude-3-haiku-20240307-v1:0" | |
| kwargs = { | |
| "modelId": model_id, | |
| "contentType": "application/json", | |
| "accept": "application/json", | |
| "body": json.dumps({ | |
| "anthropic_version": "bedrock-2023-05-31", | |
| "max_tokens": 1000, | |
| "system": system_prompt, | |
| "messages": messages | |
| }) | |
| } | |
| print(messages) | |
| # 建立 message API，讀取回應 | |
| bedrock_client = self.ai_client | |
| response = bedrock_client.invoke_model(**kwargs) | |
| response_body = json.loads(response.get('body').read()) | |
| response_completion = response_body.get('content')[0].get('text').strip() | |
| return response_completion | |