app.py

import os
import json
import copy
from datetime import datetime
import pandas as pd
import gradio as gr
from openai import OpenAI
from supabase import create_client
from language_info import *

# 시스템 프롬프트 정의
FIRST_PROMPT = """
너는 유저의 질문을 분석하는 AI야! 유저의 질문을 바탕으로 주어진 조건에 맞도록 질문을 분석해줘!
조건 1. 아래 Task 리스트 중 1개의 작업 선택!
Task : ["ReAsk", "FindProduct", "Recommandation", "Etc"]
조건 2. 만약, Task가 "FindProduct"라면, 원하는 "Product"를 "Option"으로 정의를 해줘!
조건 3. 만약, Task가 "Recommandation"라면, 유저의 조건을 "Condition"이라고 "Option"에 제공해줘!
조건 4. 만약, Task가 "ReAsk"라면, 유저의 질문에서 추가적으로 요구되는 조건을 "Condition"이라고 "Option"에 제공해줘!
** 조건 5. 화장품 관련 이야기를 제외한 다른 이야기는 답할 수 없다는 내용으로 답변을 해줘! **
** 조건 6. 언어가 다르더라도 출력은 전부 "한국어"를 사용해서 출력해! 그리고 유저 입력에 대한 한국어 번역을 "Translation"으로 제공해!**

OUTPUT 포맷
json
{
    "Task":"...",
    "Option":{"...":"...",  ..., key:val},
    "Translation":"..."
}
"""

# Supabase 로깅 함수
def log_to_supabase(supabase, question, answer, history):
    """Supabase에 채팅 로그를 저장하는 함수"""
    try:
        timestamp = datetime.now().isoformat()
        supabase.table("chat_log").insert({
            "question": question,
            "answer": answer,
            "history": json.dumps(history, ensure_ascii=False),
            "timestamp": timestamp
        }).execute()
        return True
    except Exception as e:
        print(f"Error logging to Supabase: {e}")
        return False

def local_RAG(model, message, client, vector_id):
    """로컬 검색 기반 생성 함수"""
    try:
        response = client.responses.create(
            model=model,
            input=message,
            tools=[{
                "type": "file_search",
                "vector_store_ids": [vector_id],
                "max_num_results": 5
            }]
        )
        if len(response.output)==0:
            return ""
        if not response.output[1].content[0].annotations:
            return ""
        return response.output[1].content[0].text
    except Exception as e:
        print(f"Error in RAG: {e}")
        return ""

def routing(messages, client, language):
    """사용자 질문 분석 및 라우팅 함수"""
    try:
        routing_messages = copy.deepcopy(messages)
        routing_messages[0]['content'] = FIRST_PROMPT
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=routing_messages,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    except Exception as e:
        print(f"Error in routing: {e}")
        return {"Task": "Etc"}

def generate_response(query_parsing, client, vector_id, user_message, chat_history, language):
    """쿼리 분석 결과에 따른 응답 생성 함수"""

    # 메시지 형식으로 변환
    messages = [{"role": "system", "content": get_system_prompt(language)}]
    for user_msg, bot_msg in chat_history:
        if user_msg:
            messages.append({"role": "user", "content": user_msg})
        if bot_msg:
            messages.append({"role": "assistant", "content": bot_msg})
    if language!="한국어":
        messages.append({"role": "user", "content": user_message})
    else:
        messages.append({"role": "user", "content": query_parsing["Translation"]})

    # 쿼리 분석 결과가 없는 경우
    if "Task" not in query_parsing:
        return get_text('error_understanding', language)

    # ReAsk: 추가 정보 요청
    if query_parsing["Task"] == "ReAsk":
        if "Option" in query_parsing and "Condition" in query_parsing["Option"]:
            return f"{get_text('additional_info_needed', language)}{query_parsing['Option']['Condition']}"
        else:
            return get_text('request_more_info', language)

    # FindProduct: 제품 검색
    if query_parsing["Task"] == "FindProduct":
        if "Option" not in query_parsing:
            return get_text('error_product_not_found', language)

        product = query_parsing['Option']["Product"]

        # 수파베이스에서 화장품 데이터 가져오기
        try:
            product_list = supabase.table("cosmetics").select("*").execute().data
        except:
            # 실제 구현 시 데이터베이스 연결 필수
            product_list = []  # 임시 빈 리스트

        find_data = ""
        for item in product_list:
            if product in item["title"]:
                find_data += f"item : {item['title']}\tbrand : {item['brand']}\tmake : {item['maker']}\tsummary : {item['summary']}\n\n"

        if len(find_data) > 2:
            # 데이터를 찾았을 때
            messages[-1]['content'] = messages[-1]['content'] + "\n\n<데이터>\n\n" + find_data + "\n\n</데이터>\n\n" + messages[-1]['content']
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            return response.choices[0].message.content
        else:
            # 데이터를 찾지 못했을 때 RAG 시도
            response = local_RAG("gpt-4o", user_message, client, vector_id)
            if response == "":
                # RAG도 실패했을 때 검색 모델 사용
                response = client.chat.completions.create(
                    model="gpt-4o-search-preview",
                    messages=messages
                )
                return response.choices[0].message.content
            return response

    # Recommendation: 제품 추천
    if query_parsing["Task"] == "Recommandation":
        response = local_RAG("gpt-4o", user_message, client, vector_id)
        if response == "":
            # RAG 실패 시 검색 모델 사용
            response = client.chat.completions.create(
                model="gpt-4o-search-preview",
                messages=messages
            )
            return response.choices[0].message.content
        return response

    # Etc: 일반 응답
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    return response.choices[0].message.content

def get_system_prompt(language):
    """언어에 맞는 시스템 프롬프트 생성"""
    return f"""
너는 한국의 화장품을 상담해 주는 AI야!
유저의 한국 화장품 관련 질문에 가장 적절한 답변을 해줘!
조건1 : 화장품과 관련되지 않은 질문은 대답하지 마.
조건2 : 주어진 language에 맞는 언어로 답변해줘!
조건3 : 유저가 별도의 데이터를 제공하면 (<데이터>...</데이터> 형식), 제공된 데이터 내에서만 답변해줘.
language: {language}
"""

# 현재 선택된 언어를 저장할 전역 변수
selected_language = "English"

def chatbot(message, history, language_choice=None):
    """챗봇의 메인 함수"""
    global selected_language

    # language_choice가 제공되었으면 업데이트
    if language_choice is not None:
        selected_language = language_choice

    # OpenAI API 키 및 환경 변수 설정
    openai_api_key = os.getenv("OPENAI_API_KEY")
    vector_id = os.getenv("vector_id")

    # Supabase 설정
    supabase_url = os.getenv("SUPABASE_URL")
    supabase_key = os.getenv("SUPABASE_KEY")
    supabase = None

    client = OpenAI(api_key=openai_api_key)

    # Supabase 클라이언트 생성 (있을 경우)
    if supabase_url and supabase_key:
        supabase = create_client(supabase_url, supabase_key)

    # 쿼리 분석 및 응답 생성
    query_parsing = routing([{"role": "system", "content": ""}, {"role": "user", "content": message}], client, selected_language)
    response = generate_response(query_parsing, client, vector_id, message, history, selected_language)

    # Supabase 로깅 (해당 환경변수가 있을 경우)
    if supabase:
        chat_history = []
        for user_msg, bot_msg in history:
            if user_msg:
                chat_history.append({"role": "user", "content": user_msg})
            if bot_msg:
                chat_history.append({"role": "assistant", "content": bot_msg})
        chat_history.append({"role": "user", "content": message})
        chat_history.append({"role": "assistant", "content": response})

        log_to_supabase(supabase, message, response, chat_history)

    return response

def change_language(language_choice):
    """언어 변경 함수"""
    global selected_language
    selected_language = language_choice
    return language_choice

# 메인 함수
def main():
    """Gradio 인터페이스 설정 함수"""
    # 언어 설정
    languages = ["English", "한국어", "Español", "中文", "日本語", "ภาษาไทย", "Tiếng Việt", "Bahasa Indonesia"]
    global selected_language
    selected_language = "English"
    current_dir = os.path.dirname(os.path.abspath(__file__))
    image_path = os.path.join(current_dir, "image.png")
    # Gradio 인터페이스 구성
    with gr.Blocks(css="footer {visibility: hidden}") as demo:

        gr.Markdown(f"# {get_text('title', selected_language)}")
        gr.Markdown(f"(Conversation logs are saved.)")

        with gr.Row():
            language_dropdown = gr.Dropdown(
                choices=languages,
                value=selected_language,
                label=get_text("language_selector", selected_language)
            )

        # 언어 상태를 저장할 state 객체 생성
        lang_state = gr.State(value=selected_language)

        # 언어가 변경될 때 실행되는 함수
        def update_language(new_lang, history):
            global selected_language
            selected_language = new_lang
            return new_lang, history

        # 챗봇의 메시지 전송 함수를 수정하여 현재 언어 상태를 전달
        def handle_message(message, history, lang):
            return chatbot(message, history, lang)

        chatbot_interface = gr.ChatInterface(
            fn=handle_message,
            additional_inputs=[language_dropdown],
            title="",
            examples=[
                ["이니스프리 그린티 제품 알려줘", "한국어"],
                ["แนะนำผลิตภัณฑ์ดูแลผิวที่ดีสำหรับผู้หญิง", "ภาษาไทย"],
                ["What products are effective for whitening?", "English"]
            ],
        )

        # 언어 변경 시 이벤트
        language_dropdown.change(
            fn=update_language,
            inputs=[language_dropdown, chatbot_interface.chatbot],
            outputs=[lang_state, chatbot_interface.chatbot]
        )
        gr.Markdown(f"# Architecture")

        gr.Image(value=image_path, show_label=False, height=400, width=400)

    # Gradio 앱 실행
    demo.launch()

if __name__ == "__main__":
    main()