app_gradio.py

import os
import gradio as gr
import json
import logging
from app.app import app as flask_app  # Flask 앱 가져오기
from flask import json as flask_json
from retrying import retry

# 로거 설정
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.DEBUG  # INFO에서 DEBUG로 변경하여 더 상세한 로그 확인
)
logger = logging.getLogger(__name__)

# Flask 테스트 클라이언트 초기화
@retry(tries=5, delay=1, backoff=2)
def init_flask_client():
    """백그라운드 실행 Flask 서버에 연결하는 클라이언트 초기화"""
    try:
        test_client = flask_app.test_client()
        logger.info("Flask 테스트 클라이언트 초기화 성공")
        return test_client
    except Exception as e:
        logger.error(f"Flask 테스트 클라이언트 초기화 실패: {e}")
        raise e

# 소스 정보 포맷팅 헤루퍼 함수
def format_source_info(sources, prefix=""):
    """검색 결과 소스 정보를 형식화
    
    Args:
        sources: 소스 정보 리스트
        prefix: 로그 메시지 접두사 (기본값: "")
        
    Returns:
        형식화된 소스 정보 문자열
    """
    source_info = ""
    if sources and len(sources) > 0:
        logger.debug(f"{prefix}소스 정보 포맷팅: {len(sources)}개 소스: {json.dumps(sources, ensure_ascii=False, indent=2)}")
        
        source_list = []
        for src in sources:
            source_text = src["source"]
            # id 필드가 있으면 함께 표시
            if "id" in src:
                logger.debug(f"{prefix}ID 필드 발견: {src['id']}")
                source_text += f" (ID: {src['id']})"  
            source_list.append(source_text)
        
        if source_list:
            source_info = "\n\n참조 소스:\n" + "\n".join(source_list)
            logger.debug(f"{prefix}최종 소스 정보 형식: {source_info}")
    
    return source_info

# Flask 테스트 클라이언트 초기화
flask_client = init_flask_client()

# Gradio 버전 확인 메시지
logger.info(f"Gradio 버전: {gr.__version__}")

# Gradio 인터페이스 생성
with gr.Blocks(title="RAG 검색 챗봇 with 음성인식") as demo:
    gr.HTML("""
    <div style="text-align: center; max-width: 800px; margin: 0 auto;">
        <h1>RAG 검색 챗봇 with 음성인식</h1>
        <p>텍스트 또는 음성으로 질문을 입력하세요.</p>
    </div>
    """)

    with gr.Tab("텍스트 챗"):
        text_input = gr.Textbox(label="질문 입력", placeholder="여기에 질문을 입력하세요...")
        text_output = gr.Textbox(label="응답", interactive=False)
        text_button = gr.Button("질문 제출")

    with gr.Tab("음성 챗"):
        with gr.Row():
            # 녹음 UI 개선: 마이크로폰으로 소스 지정 및 음파 표시 활성화
            audio_input = gr.Audio(
                label="음성 입력", 
                type="filepath",                
                sources=["microphone"],
                show_label=True,
                waveform_options={"show_controls": True, "normalize": True},
                interactive=True
            )
            
        audio_transcription = gr.Textbox(label="인식된 텍스트", interactive=False)
        audio_output = gr.Textbox(label="응답", interactive=False)
        gr.Markdown("""
        <div style="text-align: center; margin: 10px 0;">
            <p>녹음 정지 버튼을 누르면 자동으로 음성이 전송됩니다.</p>
        </div>
        """)

    with gr.Tab("문서 업로드"):
        doc_input = gr.File(label="문서 업로드", file_types=[".txt", ".md", ".pdf", ".docx", ".csv"])
        doc_output = gr.Textbox(label="업로드 결과", interactive=False)
        doc_button = gr.Button("문서 업로드")

    # 텍스트 챗 기능
    def handle_text_chat(query):
        if not query:
            return "질문을 입력하세요."
        try:
            logger.info("텍스트 챗 요청: /api/chat")
            response = flask_client.post("/api/chat", json={"query": query})
            data = flask_json.loads(response.data)
            
            # 디버깅을 위한 API 응답 로그
            logger.info(f"API 응답 구조: {json.dumps(data, ensure_ascii=False, indent=2)[:500]}...")
            
            if "error" in data:
                logger.error(f"텍스트 챗 오류: {data['error']}")
                return data["error"]
            
            # 소스 정보 추출 및 포맷팅
            source_info = ""
            if "sources" in data and data["sources"]:
                source_info = format_source_info(data["sources"])
            
            # 응답과 소스 정보를 함께 반환
            return data["answer"] + source_info
        except Exception as e:
            logger.error(f"텍스트 챗 처리 실패: {str(e)}")
            return f"처리 중 오류 발생: {str(e)}"

    # 음성 챗 기능
    def handle_voice_chat(audio_file):
        if not audio_file:
            return "", "음성을 업로드하세요."
        try:
            logger.info("음성 챗 요청: /api/voice")
            with open(audio_file, "rb") as f:
                # Flask 테스트 클라이언트는 files 직접 지원 안 하므로, 데이터를 읽어 전달
                response = flask_client.post(
                    "/api/voice",
                    data={"audio": (f, "audio_file")}
                )
            data = flask_json.loads(response.data)
            
            # 디버깅을 위한 API 응답 로그
            logger.info(f"[음성챗] API 응답 구조: {json.dumps(data, ensure_ascii=False, indent=2)[:500]}...")
            
            if "error" in data:
                logger.error(f"음성 챗 오류: {data['error']}")
                return "", data["error"]
            
            # 소스 정보 추출 및 포맷팅
            source_info = ""
            if "sources" in data and data["sources"]:
                source_info = format_source_info(data["sources"], prefix="[음성챗] ")
            
            # 인식된 텍스트와 소스 정보가 포함된 응답 반환
            return data["transcription"], data["answer"] + source_info
        except Exception as e:
            logger.error(f"음성 챗 처리 실패: {str(e)}")
            return "", f"처리 중 오류 발생: {str(e)}"

    # 문서 업로드 기능
    def handle_doc_upload(doc_file):
        if not doc_file:
            return "문서를 업로드하세요."
        try:
            logger.info(f"문서 업로드 요청: /api/upload, 파일명: {doc_file.name}")
            file_extension = os.path.splitext(doc_file.name)[1].lower()
            logger.info(f"파일 확장자: {file_extension}")
            
            with open(doc_file, "rb") as f:
                response = flask_client.post(
                    "/api/upload",
                    data={"document": (f, doc_file.name)}
                )
            data = flask_json.loads(response.data)
            if "error" in data:
                logger.error(f"문서 업로드 오류: {data['error']}")
                return data["error"]
            return data["message"]
        except Exception as e:
            logger.error(f"문서 업로드 처리 실패: {str(e)}")
            return f"처리 중 오류 발생: {str(e)}"

    # 이벤트 핸들러 연결
    text_button.click(
        fn=handle_text_chat,
        inputs=text_input,
        outputs=text_output
    )
    
    # 음성 입력 값이 변경될 때 자동으로 전송
    audio_input.change(
        fn=handle_voice_chat,
        inputs=audio_input,
        outputs=[audio_transcription, audio_output]
    )
    
    doc_button.click(
        fn=handle_doc_upload,
        inputs=doc_input,
        outputs=doc_output
    )

if __name__ == "__main__":
    demo.launch(server_port=7860)