Create app-backup.py

app-backup.py

@@ -0,0 +1,385 @@
+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 환경 변수가 설정되지 않았습니다.")
+    
+    # 키워드 추출 함수 (LLM 기반)
+    def extract_keywords_with_llm(query):
+        if not api_key:
+            return "LLM 키워드 추출을 위한 FW_API_KEY가 설정되지 않았습니다.", query
+        
+        # LLM을 사용하여 키워드 추출 (DeepSeek 모델 사용)
+        url = "https://api.fireworks.ai/inference/v1/chat/completions"
+        payload = {
+            "model": "accounts/fireworks/models/deepseek-v3-0324",
+            "max_tokens": 200,
+            "temperature": 0.1,  # 일관된 결과를 위해 낮은 온도 사용
+            "messages": [
+                {
+                    "role": "system",
+                    "content": "사용자의 질문에서 웹 검색에 효과적인 핵심 키워드를 추출하세요. 키워드 사이에 쉼표나 공백을 넣지 말고 하나의 검색어처럼 수정해서 제공하세요. 예를 들어 '한덕수 국무총리 탄핵 결과'처럼 공백으로만 구분하세요."
+                },
+                {
+                    "role": "user",
+                    "content": query
+                }
+            ]
+        }
+        headers = {
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {api_key}"
+        }
+        
+        try:
+            response = requests.post(url, headers=headers, json=payload)
+            response.raise_for_status()
+            result = response.json()
+            
+            # 응답에서 키워드 추출
+            keywords = result["choices"][0]["message"]["content"].strip()
+            
+            # 키워드가 너무 길거나 형식이 잘못된 경우 원본 쿼리 사용
+            if len(keywords) > 100:
+                return f"추출된 키워드: {keywords}", query
+                
+            return f"추출된 키워드: {keywords}", keywords
+            
+        except Exception as e:
+            print(f"키워드 추출 중 오류 발생: {str(e)}")
+            return f"키워드 추출 중 오류 발생: {str(e)}", query
+    
+    # SerpHouse API를 사용하여 검색 수행 함수
+    def search_with_serphouse(query):
+        if not serphouse_api_key:
+            return "SERPHOUSE_API_KEY가 설정되지 않았습니다."
+        
+        try:
+            # 키워드 추출
+            extraction_result, search_query = extract_keywords_with_llm(query)
+            print(f"원본 쿼리: {query}")
+            print(extraction_result)
+            
+            # 문서 코드를 더 자세히 분석해 보니 기본 GET 방식 활용이 좋을 것 같습니다
+            url = "https://api.serphouse.com/serp/live"
+            
+            # 한글 검색어인지 확인
+            is_korean = any('\uAC00' <= c <= '\uD7A3' for c in search_query)
+            
+            # 간소화된 파라미터로 시도
+            params = {
+                "q": search_query,
+                "domain": "google.com",
+                "serp_type": "web",  # 기본 웹 검색으로 변경
+                "device": "desktop",
+                "lang": "ko" if is_korean else "en"
+            }
+            
+            headers = {
+                "Authorization": f"Bearer {serphouse_api_key}"
+            }
+            
+            print(f"SerpHouse API 호출 중... 기본 GET 방식으로 시도")
+            print(f"검색어: {search_query}")
+            print(f"요청 URL: {url} - 파라미터: {params}")
+            
+            # GET 요청 수행
+            response = requests.get(url, headers=headers, params=params)
+            response.raise_for_status()
+            
+            print(f"SerpHouse API 응답 상태 코드: {response.status_code}")
+            search_results = response.json()
+            
+            # 응답 구조 확인
+            print(f"응답 구조: {list(search_results.keys()) if isinstance(search_results, dict) else '딕셔너리 아님'}")
+            
+            # 검색 결과 파싱 및 포맷팅
+            formatted_results = []
+            formatted_results.append(f"검색어: {search_query}\n\n")
+            
+            # 다양한 가능한 응답 구조에 대한 처리
+            organic_results = None
+            
+            # 가능한 응답 구조 1
+            if "results" in search_results and "organic" in search_results["results"]:
+                organic_results = search_results["results"]["organic"]
+                
+            # 가능한 응답 구조 2
+            elif "organic" in search_results:
+                organic_results = search_results["organic"]
+                
+            # 가능한 응답 구조 3 (중첩된 results)
+            elif "results" in search_results and "results" in search_results["results"]:
+                if "organic" in search_results["results"]["results"]:
+                    organic_results = search_results["results"]["results"]["organic"]
+            
+            # organic_results가 있으면 처리
+            if organic_results and len(organic_results) > 0:
+                # 응답 구조 출력
+                print(f"첫번째 organic 결과 구조: {organic_results[0].keys() if len(organic_results) > 0 else 'empty'}")
+                
+                for result in organic_results[:5]:  # 상위 5개 결과만 표시
+                    title = result.get("title", "제목 없음")
+                    snippet = result.get("snippet", "내용 없음")
+                    link = result.get("link", "#")
+                    
+                    formatted_results.append(
+                        f"제목: {title}\n"
+                        f"내용: {snippet}\n"
+                        f"링크: {link}\n\n"
+                    )
+                
+                print(f"검색 결과 {len(organic_results)}개 찾음")
+                return "".join(formatted_results)
+            
+            # 결과가 없거나 예상치 못한 구조인 경우
+            print("검색 결과 없음 또는 예상치 못한 응답 구조")
+            print(f"응답 구조 상세: {search_results.keys() if hasattr(search_results, 'keys') else '불명확한 구조'}")
+            
+            # 응답 내용에서 오류 메시지 찾기
+            error_msg = "검색 결과가 없거나 응답 형식이 예상과 다릅니다"
+            if "error" in search_results:
+                error_msg = search_results["error"]
+            elif "message" in search_results:
+                error_msg = search_results["message"]
+                
+            return f"검색어 '{search_query}'에 대한 결과: {error_msg}"
+            
+        except Exception as e:
+            error_msg = f"검색 중 오류 발생: {str(e)}"
+            print(error_msg)
+            import traceback
+            print(traceback.format_exc())
+            
+            # 디버깅 목적으로 API 요청 상세 정보 추가
+            return f"검색 중 오류가 발생했습니다: {str(e)}\n\n" + \
+                   f"API 요청 상세 정보:\n" + \
+                   f"- URL: {url}\n" + \
+                   f"- 검색어: {search_query}\n" + \
+                   f"- 파라미터: {params}\n"
+    
+    # 스트리밍 방식으로 DeepSeek API 호출 함수
+    def query_deepseek_streaming(message, history, use_deep_research):
+        if not api_key:
+            yield history, "환경 변수 FW_API_KEY가 설정되지 않았습니다. 서버에서 환경 변수를 확인해주세요."
+            return
+        
+        search_context = ""
+        search_info = ""
+        if use_deep_research:
+            try:
+                # 검색 수행 (첫 메시지 전달)
+                yield history + [(message, "🔍 최적의 키워드 추출 및 웹 검색 중...")], ""
+                
+                # 검색 실행 - 디버깅을 위한 로그 추가
+                print(f"Deep Research 활성화됨: 메시지 '{message}'에 대한 검색 시작")
+                search_results = search_with_serphouse(message)
+                print(f"검색 결과 수신 완료: {search_results[:100]}...")  # 결과 앞부분만 출력
+                
+                if not search_results.startswith("검색 중 오류 발생") and not search_results.startswith("SERPHOUSE_API_KEY"):
+                    search_context = f"""
+다음은 사용자 질문과 관련된 최신 검색 결과입니다. 이 정보를 참고하여 정확하고 최신 정보가 반영된 응답을 제공하세요:
+{search_results}
+위 검색 결과를 기반으로 사용자의 다음 질문에 답변하세요. 검색 결과에서 명확한 답변을 찾을 수 없는 경우, 당신의 지식을 활용하여 최선의 답변을 제공하세요.
+검색 결과를 인용할 때는 출처를 명시하고, 답변이 최신 정보를 반영하도록 하세요.
+"""
+                    search_info = f"🔍 Deep Research 기능 활성화: 관련 웹 검색 결과를 기반으로 응답 생성 중..."
+                else:
+                    print(f"검색 실패 또는 결과 없음: {search_results}")
+            except Exception as e:
+                print(f"Deep Research 처리 중 예외 발생: {str(e)}")
+                search_info = f"🔍 Deep Research 기능 오류: {str(e)}"
+        
+        # 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()
+            
+            # search_info가 있으면 시작 메시지에 포함
+            start_msg = search_info if search_info else ""
+            new_history.append((message, start_msg))
+            
+            # 응답 전체 텍스트
+            full_response = start_msg
+            
+            # 스트리밍 응답 처리
+            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():
+                    with gr.Column(scale=3):
+                        use_deep_research = gr.Checkbox(
+                            label="Deep Research 활성화",
+                            info="최적의 키워드 추출 및 웹 검색을 통한 최신 정보 활용",
+                            value=False
+                        )
+                    with gr.Column(scale=1):
+                        api_status = gr.Markdown("API 상태: 준비됨")
+                
+                # API 키 상태 확인 및 표시
+                if not serphouse_api_key:
+                    api_status.value = "⚠️ SERPHOUSE_API_KEY가 설정되지 않았습니다"
+                if not api_key:
+                    api_status.value = "⚠️ FW_API_KEY가 설정되지 않았습니다"
+                if api_key and serphouse_api_key:
+                    api_status.value = "✅ API 키 설정 완료"
+                
+                # 입��� 영역
+                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)