app.py

import gradio as gr
import os
import requests
import json
import time
from dotenv import load_dotenv

# .env 파일 로드 (있는 경우)
load_dotenv()

def create_deepseek_interface():
    # 환경 변수에서 API 키 가져오기
    api_key = os.getenv("FW_API_KEY")
    serphouse_api_key = os.getenv("SERPHOUSE_API_KEY")
    
    if not api_key:
        print("경고: FW_API_KEY 환경 변수가 설정되지 않았습니다.")
    if not serphouse_api_key:
        print("경고: SERPHOUSE_API_KEY 환경 변수가 설정되지 않았습니다.")
    
    # SerpHouse API를 사용하여 검색 수행 함수
    def search_with_serphouse(query):
        if not serphouse_api_key:
            return "SERPHOUSE_API_KEY가 설정되지 않았습니다."
        
        url = "https://api.serphouse.com/serp/live"
        payload = {
            "q": query,
            "domain": "google.com",
            "loc": "us",
            "lang": "en",
            "device": "desktop",
            "serp_type": "web",
            "page": 1,
            "num": 5  # 상위 5개 결과만 가져오기
        }
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {serphouse_api_key}"
        }
        
        try:
            response = requests.post(url, headers=headers, json=payload)
            response.raise_for_status()
            
            search_results = response.json()
            
            # 검색 결과 파싱 및 포맷팅
            formatted_results = []
            if "organic" in search_results and len(search_results["organic"]) > 0:
                for result in search_results["organic"][:5]:  # 상위 5개 결과만 사용
                    title = result.get("title", "제목 없음")
                    snippet = result.get("snippet", "내용 없음")
                    link = result.get("link", "#")
                    formatted_results.append(f"제목: {title}\n내용: {snippet}\n링크: {link}\n\n")
                
                return "".join(formatted_results)
            else:
                return "검색 결과가 없습니다."
                
        except Exception as e:
            return f"검색 중 오류 발생: {str(e)}"
    
    # 스트리밍 방식으로 DeepSeek API 호출 함수
    def query_deepseek_streaming(message, history, use_deep_research):
        if not api_key:
            yield history, "환경 변수 FW_API_KEY가 설정되지 않았습니다. 서버에서 환경 변수를 확인해주세요."
            return
        
        search_context = ""
        if use_deep_research:
            # 검색 수행
            search_results = search_with_serphouse(message)
            if not search_results.startswith("검색 중 오류 발생") and not search_results.startswith("SERPHOUSE_API_KEY"):
                search_context = f"다음은 사용자 질문과 관련된 최신 검색 결과입니다. 이 정보를 참고하여 정확하고 최신 정보가 반영된 응답을 제공하세요:\n\n{search_results}\n\n사용자 질문에 답변하세요.\n\n"
        
        # API 요청을 위한 대화 기록 준비
        messages = []
        for user, assistant in history:
            messages.append({"role": "user", "content": user})
            messages.append({"role": "assistant", "content": assistant})
        
        # 검색 컨텍스트가 있으면 시스템 메시지 추가
        if search_context:
            # DeepSeek 모델은 시스템 메시지를 지원합니다
            messages.insert(0, {"role": "system", "content": search_context})
        
        # 새 사용자 메시지 추가
        messages.append({"role": "user", "content": message})
        
        # API 요청 준비
        url = "https://api.fireworks.ai/inference/v1/chat/completions"
        payload = {
            "model": "accounts/fireworks/models/deepseek-v3-0324",
            "max_tokens": 20480,
            "top_p": 1,
            "top_k": 40,
            "presence_penalty": 0,
            "frequency_penalty": 0,
            "temperature": 0.6,
            "messages": messages,
            "stream": True  # 스트리밍 활성화
        }
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"
        }
        
        try:
            # 스트리밍 응답 요청
            response = requests.request("POST", url, headers=headers, data=json.dumps(payload), stream=True)
            response.raise_for_status()  # HTTP 오류 발생 시 예외 발생
            
            # 메시지를 추가하고 초기 빈 응답으로 시작
            new_history = history.copy()
            new_history.append((message, ""))
            
            # 응답 전체 텍스트
            full_response = ""
            
            # 스트리밍 응답 처리
            for line in response.iter_lines():
                if line:
                    line_text = line.decode('utf-8')
                    
                    # 'data: ' 접두사 제거
                    if line_text.startswith("data: "):
                        line_text = line_text[6:]
                    
                    # 스트림 종료 메시지 확인
                    if line_text == "[DONE]":
                        break
                    
                    try:
                        # JSON 파싱
                        chunk = json.loads(line_text)
                        chunk_content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
                        
                        if chunk_content:
                            full_response += chunk_content
                            # 채팅 기록 업데이트
                            new_history[-1] = (message, full_response)
                            yield new_history, ""
                    except json.JSONDecodeError:
                        continue
            
            # 최종 응답 반환
            yield new_history, ""
            
        except requests.exceptions.RequestException as e:
            error_msg = f"API 오류: {str(e)}"
            if hasattr(e, 'response') and e.response and e.response.status_code == 401:
                error_msg = "인증 실패. 환경 변수 FW_API_KEY를 확인해주세요."
            yield history, error_msg
    
    # Gradio 인터페이스 생성
    with gr.Blocks(theme="soft", fill_height=True) as demo:
        # 헤더 섹션
        gr.Markdown(
            """
            # 🤖 DeepSeek V3 스트리밍 인터페이스
            ### Fireworks AI가 제공하는 고급 AI 모델 - 실시간 응답 지원
            """
        )
        
        # 메인 레이아웃
        with gr.Row():
            # 메인 콘텐츠 영역
            with gr.Column():
                # 채팅 인터페이스
                chatbot = gr.Chatbot(
                    height=500,
                    show_label=False,
                    container=True
                )
                
                # Deep Research 토글 추가
                with gr.Row():
                    use_deep_research = gr.Checkbox(
                        label="Deep Research 활성화",
                        info="최신 웹 검색 결과를 활용한 응답 생성"
                    )
                
                # 입력 영역
                with gr.Row():
                    msg = gr.Textbox(
                        label="메시지",
                        placeholder="여기에 프롬프트를 입력하세요...",
                        show_label=False,
                        scale=9
                    )
                    submit = gr.Button("전송", variant="primary", scale=1)
                
                # 대화 초기화 버튼
                with gr.Row():
                    clear = gr.ClearButton([msg, chatbot], value="🧹 대화 초기화")
                    
                # 예제 쿼리
                gr.Examples(
                    examples=[
                        "딥러닝에서 트랜스포머와 RNN의 차이점을 설명해주세요.",
                        "특정 범위 내의 소수를 찾는 파이썬 함수를 작성해주세요.",
                        "강화학습의 주요 개념을 요약해주세요."
                    ],
                    inputs=msg
                )
                
                # 오류 메시지 표시
                error_box = gr.Markdown("")
        
        # 버튼과 기능 연결
        submit.click(
            query_deepseek_streaming,
            inputs=[msg, chatbot, use_deep_research],
            outputs=[chatbot, error_box]
        ).then(
            lambda: "",
            None,
            [msg]
        )
        
        # Enter 키 제출 허용
        msg.submit(
            query_deepseek_streaming,
            inputs=[msg, chatbot, use_deep_research],
            outputs=[chatbot, error_box]
        ).then(
            lambda: "",
            None,
            [msg]
        )
        
    return demo

# 인터페이스 실행
if __name__ == "__main__":
    demo = create_deepseek_interface()
    demo.launch(debug=True)