# 안정적인 AI 카피라이터 - 임베딩 기반 RAG 시스템 # Hugging Face Spaces 환경 최적화 버전 import streamlit as st import pandas as pd import numpy # 전역적으로 numpy를 먼저 임포트해봅니다. import pickle import google.generativeai as genai import time import json import os import sys # 디버깅용 sys 모듈 임포트 from datetime import datetime # 환경 설정 (권한 문제 해결) os.environ['STREAMLIT_BROWSER_GATHER_USAGE_STATS'] = 'false' # 캐시 경로를 /tmp 로 설정 (Hugging Face Spaces에서 권장되는 쓰기 가능 경로) TMP_DIR = "/tmp" TRANSFORMERS_CACHE_DIR = os.path.join(TMP_DIR, '.cache', 'transformers') SENTENCE_TRANSFORMERS_HOME_DIR = os.path.join(TMP_DIR, '.cache', 'sentence_transformers') os.environ['TRANSFORMERS_CACHE'] = TRANSFORMERS_CACHE_DIR os.environ['SENTENCE_TRANSFORMERS_HOME'] = SENTENCE_TRANSFORMERS_HOME_DIR # 캐시 디렉토리 생성 (존재하지 않으면) - /tmp 아래는 일반적으로 생성 가능 try: os.makedirs(TRANSFORMERS_CACHE_DIR, exist_ok=True) os.makedirs(SENTENCE_TRANSFORMERS_HOME_DIR, exist_ok=True) except PermissionError: st.warning(f"⚠️ 캐시 디렉토리 생성 권한 없음: {TRANSFORMERS_CACHE_DIR} 또는 {SENTENCE_TRANSFORMERS_HOME_DIR}. 모델 다운로드가 느릴 수 있습니다.") except Exception as e_mkdir: st.warning(f"⚠️ 캐시 디렉토리 생성 중 오류: {e_mkdir}") # 페이지 설정 st.set_page_config( page_title="AI 카피라이터 | RAG 기반 광고 카피 생성", page_icon="✨", layout="wide", initial_sidebar_state="expanded" ) # 제목 및 설명 st.title("✨ AI 카피라이터") st.markdown("### 🎯 37,671개 실제 광고 카피 데이터 기반 RAG 시스템") st.markdown("---") # --- 런타임 환경 디버깅 (애플리케이션 최상단 또는 load_system 바로 전) --- st.sidebar.markdown("---") st.sidebar.markdown("### ⚙️ 런타임 환경 정보 (디버깅용)") st.sidebar.text(f"Py Exec: {sys.executable}") st.sidebar.text(f"Py Ver: {sys.version.split()[0]}") # 간략하게 버전만 # st.sidebar.text(f"sys.path: {sys.path}") # 너무 길어서 일단 주석 st.sidebar.text(f"PYTHONPATH: {os.environ.get('PYTHONPATH', 'Not Set')}") try: # numpy를 여기서 다시 임포트하고 사용 import numpy as np_runtime_check st.sidebar.text(f"NumPy Ver (Runtime): {np_runtime_check.__version__}") # 핵심 모듈 임포트 시도 import numpy.core._multiarray_umath st.sidebar.markdown("✅ NumPy core modules imported (Runtime)") except Exception as e: st.sidebar.error(f"❌ NumPy import error (Runtime): {e}") st.sidebar.markdown("---") # --- 디버깅 코드 끝 --- # 사이드바 설정 st.sidebar.header("🎛️ 카피 생성 설정") # API 키 입력 (환경변수 우선 사용) default_api_key = os.getenv("GEMINI_API_KEY", "") api_key = st.sidebar.text_input( "🔑 Gemini API 키", value=default_api_key, type="password", help="환경변수에 GEMINI_API_KEY로 설정하면 자동 입력됩니다" ) if not api_key: st.warning("⚠️ Gemini API 키를 입력해주세요") st.info("💡 Settings → Repository secrets에서 GEMINI_API_KEY를 설정하세요") st.stop() # 시스템 초기화 (캐싱) - 임베딩 필수! @st.cache_resource(show_spinner=False) def load_system(): """시스템 컴포넌트 로딩 - 임베딩 기반 RAG 시스템""" # --- 함수 시작 시 디버깅 정보 추가 --- st.write("--- load_system() 시작 ---") st.write(f"Python Executable (load_system): {sys.executable}") st.write(f"Python Version (load_system): {sys.version}") # st.write(f"sys.path (load_system): {sys.path}") # 너무 길어서 주석 st.write(f"PYTHONPATH (load_system): {os.environ.get('PYTHONPATH')}") try: import numpy as np_load_system_check # 새 별칭 사용 st.write(f"NumPy version (load_system start): {np_load_system_check.__version__}") import numpy.core._multiarray_umath st.write("load_system start: Successfully imported numpy.core._multiarray_umath") except Exception as e: st.write(f"load_system start: Error importing NumPy parts: {e}") # --- 디버깅 정보 끝 --- progress_container = st.container() with progress_container: # 전체 진행률 total_progress = st.progress(0) status_text = st.empty() # 1단계: API 설정 (10%) status_text.text("🔑 Gemini API 초기화 중...") try: genai.configure(api_key=api_key) model_llm = genai.GenerativeModel('gemini-1.5-flash') # 모델 이름 확인 (이전엔 gemini-2.0-flash) total_progress.progress(10) st.success("✅ Gemini API 설정 완료") except Exception as e: st.error(f"❌ Gemini API 설정 실패: {e}") return None, None, None, None # 2단계: 임베딩 모델 로드 (40%) status_text.text("🤖 한국어 임베딩 모델 로딩 중... (1-2분 소요)") embedding_model_instance = None # 변수명 변경 try: # sentence-transformers 임포트를 함수 내에서 유지 from sentence_transformers import SentenceTransformer # from sklearn.metrics.pairwise import cosine_similarity # 여기서는 아직 필요 없음 embedding_model_instance = SentenceTransformer('jhgan/ko-sbert-nli', cache_folder=SENTENCE_TRANSFORMERS_HOME_DIR) # 수정된 캐시 경로 사용 total_progress.progress(40) st.success("✅ 한국어 임베딩 모델 로딩 완료") except Exception as e: st.error(f"❌ 임베딩 모델 로딩 실패: {e}") st.error("🚨 임베딩 모델 없이는 RAG 시스템이 작동할 수 없습니다!") return None, None, None, None # 3단계: 데이터 로드 (60%) status_text.text("📊 카피 데이터베이스 로딩 중...") df_data = None # 변수명 변경 try: df_data = pd.read_excel('광고카피데이터_브랜드추출완료.xlsx') total_progress.progress(60) st.success(f"✅ 데이터 로딩 완료: {len(df_data):,}개 카피") except Exception as e: st.error(f"❌ 데이터 로딩 실패: {e}") return None, None, None, None # 4단계: 임베딩 데이터 로드 (90%) - 이게 핵심! status_text.text("🔍 벡터 임베딩 로딩 중... (RAG 시스템 핵심)") embeddings_array = None # 변수명 변경 try: # --- pickle.load() 직전 NumPy 디버깅 --- import numpy as np_pickle_check # 새 별칭 사용 st.write(f"[DEBUG] NumPy version just before pickle.load: {np_pickle_check.__version__}") import numpy.core._multiarray_umath st.write("[DEBUG] Successfully imported numpy.core._multiarray_umath before pickle.load") # --- 디버깅 끝 --- with open('copy_embeddings.pkl', 'rb') as f: embeddings_data = pickle.load(f) embeddings_array = embeddings_data['embeddings'] total_progress.progress(90) st.success(f"✅ 임베딩 로딩 완료: {embeddings_array.shape[0]:,}개 × {embeddings_array.shape[1]}차원") except ModuleNotFoundError as mnfe: # ModuleNotFoundError를 특정해서 잡기 st.error(f"❌ 임베딩 로딩 실패 (ModuleNotFoundError): {mnfe}") st.error(f"🚨 해당 모듈을 찾을 수 없습니다. sys.path: {sys.path}") st.error("🚨 임베딩 없이는 의미적 검색이 불가능합니다!") # 추가 디버깅: 현재 로드된 numpy 객체 상태 try: import numpy as np_final_check st.error(f"[DEBUG] NumPy object at failure: {np_final_check}") st.error(f"[DEBUG] NumPy __file__ at failure: {np_final_check.__file__}") except Exception as e_np_final: st.error(f"[DEBUG] Could not even import numpy at failure: {e_np_final}") return None, None, None, None except Exception as e: st.error(f"❌ 임베딩 로딩 실패 (일반 오류): {e}") st.error("🚨 임베딩 없이는 의미적 검색이 불가능합니다!") return None, None, None, None # 5단계: 최종 검증 (100%) status_text.text("✨ 시스템 검증 중...") if model_llm and embedding_model_instance and df_data is not None and embeddings_array is not None: total_progress.progress(100) status_text.text("🎉 RAG 시스템 로딩 완료!") success_col1, success_col2, success_col3 = st.columns(3) with success_col1: st.metric("카피 데이터", f"{len(df_data):,}개") with success_col2: st.metric("임베딩 차원", f"{embeddings_array.shape[1]}D") with success_col3: st.metric("검색 엔진", "Korean SBERT") time.sleep(1) total_progress.empty() status_text.empty() # 전역 변수명과의 충돌을 피하기 위해 함수 내에서 사용한 변수명으로 반환 return model_llm, embedding_model_instance, df_data, embeddings_array else: st.error("❌ 시스템 로딩 실패: 필수 구성요소 누락") return None, None, None, None # 시스템 로딩 (변수명 충돌 방지를 위해 새로운 이름 사용) loaded_model, loaded_embedding_model, loaded_df, loaded_embeddings = None, None, None, None with st.spinner("🚀 AI 카피라이터 시스템 초기화 중..."): loaded_model, loaded_embedding_model, loaded_df, loaded_embeddings = load_system() if loaded_model is None or loaded_embedding_model is None or loaded_df is None or loaded_embeddings is None: st.error("❌ 시스템을 로딩할 수 없습니다. 페이지를 새로고침하거나 관리자에게 문의하세요.") st.stop() # 사이드바 설정 (시스템 로딩 성공 후) st.sidebar.success("🎉 RAG 시스템 준비 완료!") # 카테고리 선택 categories = ['전체'] + sorted(loaded_df['카테고리'].unique().tolist()) selected_category = st.sidebar.selectbox( "📂 카테고리", categories, help="특정 카테고리로 검색을 제한할 수 있습니다" ) # 타겟 고객 설정 target_audience = st.sidebar.selectbox( "🎯 타겟 고객", ['20대', '30대', '일반', '10대', '40대', '50대+', '남성', '여성', '직장인', '학생', '주부'], help="타겟 고객에 맞는 톤앤매너로 카피를 생성합니다" ) # 브랜드 톤앤매너 brand_tone = st.sidebar.selectbox( "🎨 브랜드 톤", ['세련된', '친근한', '고급스러운', '활기찬', '신뢰할 수 있는', '젊은', '따뜻한', '전문적인'], help="원하는 브랜드 이미지를 선택하세요" ) # 창의성 수준 creative_level = st.sidebar.select_slider( "🧠 창의성 수준", options=['보수적', '균형', '창의적'], value='균형', help="보수적: 안전한 표현, 창의적: 독창적 표현" ) # 메인 입력 영역 st.markdown("## 💭 어떤 카피를 만들고 싶으신가요?") user_request = "" # 초기화 input_method = st.radio( "입력 방식 선택:", ["직접 입력", "템플릿 선택"], horizontal=True, key="input_method_radio" # 고유 키 추가 ) if input_method == "직접 입력": user_request = st.text_area( "카피 요청을 자세히 작성해주세요:", placeholder="예: 30대 직장 여성용 프리미엄 스킨케어 신제품 런칭 카피", height=100, key="user_request_direct" # 고유 키 추가 ) else: templates = { "신제품 런칭": "대상 {카테고리} 신제품 런칭 카피", "할인 이벤트": "{카테고리} 할인 이벤트 프로모션 카피", "브랜드 슬로건": "{카테고리} 브랜드의 대표 슬로건", "앱/서비스 리뉴얼": "{서비스명} 새 버전 출시 카피", "시즌 한정": "{시즌} 한정 {카테고리} 특별 에디션 카피" } selected_template = st.selectbox("템플릿 선택:", list(templates.keys()), key="template_selectbox") template_category = "" service_name = "" season = "" col1, col2 = st.columns(2) with col1: template_category = st.text_input("제품/서비스:", value="", key="template_category_input") with col2: if selected_template == "앱/서비스 리뉴얼": service_name = st.text_input("서비스명:", placeholder="예: 배달앱, 금융앱", key="template_service_name_input") user_request = templates[selected_template].format(서비스명=service_name) elif selected_template == "시즌 한정": season = st.selectbox("시즌:", ["봄", "여름", "가을", "겨울", "크리스마스", "신년"], key="template_season_selectbox") user_request = templates[selected_template].format(시즌=season, 카테고리=template_category) else: user_request = templates[selected_template].format(카테고리=template_category) st.text_area("생성된 요청:", value=user_request, height=80, disabled=True, key="generated_request_template") # 고급 옵션 with st.expander("🔧 고급 옵션"): col1_adv, col2_adv = st.columns(2) # 변수명 변경 with col1_adv: num_concepts = st.slider("생성할 컨셉 수:", 1, 5, 3, key="num_concepts_slider") min_similarity = st.slider("최소 유사도:", 0.0, 1.0, 0.3, 0.1, key="min_similarity_slider") with col2_adv: show_references = st.checkbox("참고 카피 보기", value=True, key="show_references_checkbox") num_references = st.slider("참고 카피 수:", 3, 10, 5, key="num_references_slider") # RAG 카피 생성 함수 (임베딩 기반 필수!) def generate_copy_with_rag(user_req, category_filter, target_aud, brand_tn, creative_lvl, num_con): # 변수명 변경 """RAG 기반 카피 생성 - 임베딩 필수 사용""" if not user_req.strip(): st.error("❌ 카피 요청을 입력해주세요") return None progress_bar = st.progress(0) status_text_gen = st.empty() # 변수명 변경 status_text_gen.text("🔍 의미적 검색 중... (RAG 핵심 기능)") progress_bar.progress(20) try: search_query = f"{user_req} {target_aud} 광고 카피" from sklearn.metrics.pairwise import cosine_similarity # generate_copy_with_rag 내에서 임포트 query_embedding = loaded_embedding_model.encode([search_query]) # 로드된 모델 사용 if category_filter != '전체': filtered_df_gen = loaded_df[loaded_df['카테고리'] == category_filter].copy() # .copy() 추가 else: filtered_df_gen = loaded_df.copy() # .copy() 추가 progress_bar.progress(40) if filtered_df_gen.empty: st.warning(f"⚠️ 선택하신 카테고리 '{category_filter}'에 해당하는 데이터가 없습니다.") progress_bar.empty() status_text_gen.empty() return None filtered_indices = filtered_df_gen.index.tolist() # loaded_embeddings에서 직접 인덱싱하기 전에, filtered_indices가 loaded_embeddings의 범위 내에 있는지 확인 valid_indices_for_embedding = [idx for idx in filtered_indices if idx < len(loaded_embeddings)] if not valid_indices_for_embedding: st.warning(f"⚠️ 유효한 인덱스를 찾을 수 없어 유사도 검색을 진행할 수 없습니다. (카테고리: {category_filter})") progress_bar.empty() status_text_gen.empty() return None # 유효한 인덱스에 해당하는 임베딩만 사용 # 이 부분은 원본 데이터프레임(loaded_df)의 인덱스를 사용해야 함 # filtered_df_gen의 인덱스는 loaded_df의 부분집합이므로, # loaded_embeddings에서 이 인덱스들을 직접 사용해야 합니다. # 주의: filtered_indices는 loaded_df의 실제 인덱스 값이어야 함. # 만약 filtered_df_gen.index가 0부터 시작하는 새로운 인덱스라면, 매핑 필요. # 현재 코드는 filtered_df.index.tolist()가 원본 인덱스를 유지한다고 가정. filtered_embeddings_for_search = loaded_embeddings[valid_indices_for_embedding] # 유사도 계산 시 query_embedding과 filtered_embeddings_for_search의 차원 확인 if query_embedding.shape[1] != filtered_embeddings_for_search.shape[1]: st.error(f"❌ 임베딩 차원 불일치: 쿼리({query_embedding.shape[1]}D), 문서({filtered_embeddings_for_search.shape[1]}D)") return None similarities = cosine_similarity(query_embedding, filtered_embeddings_for_search)[0] # 상위 N개 (num_references) 선택 # similarities의 길이는 valid_indices_for_embedding의 길이와 같음 # top_indices는 similarities 배열 내의 인덱스 num_to_select = min(num_references, len(similarities)) top_similarity_indices = np.argsort(similarities)[::-1][:num_to_select] reference_copies = [] for i in top_similarity_indices: # i는 similarities 배열에서의 인덱스. # 이 인덱스를 사용하여 valid_indices_for_embedding에서 원본 데이터프레임의 인덱스를 가져와야 함. original_df_idx = valid_indices_for_embedding[i] row = loaded_df.iloc[original_df_idx] # 원본 df에서 가져옴 if similarities[i] >= min_similarity: reference_copies.append({ 'copy': row['카피 내용'], 'brand': row['브랜드'], 'similarity': float(similarities[i]) # float으로 변환 (JSON 직렬화 대비) }) progress_bar.progress(60) if not reference_copies: st.warning(f"⚠️ 유사도 {min_similarity} 이상인 참고 카피를 찾을 수 없습니다. 유사도를 낮춰보세요.") # 참고 카피가 없어도 LLM에게 생성을 요청할 수는 있도록 함 (선택사항) # progress_bar.empty() # status_text_gen.empty() # return None references_text_for_prompt = "유사도 높은 참고 카피를 찾지 못했습니다." else: references_text_for_prompt = "\n".join([ f"{j+1}. \"{ref['copy']}\" - {ref['brand']} (유사도: {ref['similarity']:.3f})" for j, ref in enumerate(reference_copies) ]) status_text_gen.text("🤖 AI 카피 생성 중...") progress_bar.progress(80) creativity_guidance = { "보수적": "안전하고 검증된 표현을 사용하여", "균형": "창의적이면서도 적절한 수준에서", "창의적": "독창적이고 혁신적인 표현으로" } prompt = f""" 당신은 한국의 전문 광고 카피라이터입니다. **요청사항:** {user_req} **타겟 고객:** {target_aud} **브랜드 톤:** {brand_tn} **창의성 수준:** {creative_lvl} ({creativity_guidance[creative_lvl]}) **참고 카피들 (의미적 유사도 기반 선별):** {references_text_for_prompt} **작성 가이드라인:** 1. 위 참고 카피들의 스타일과 톤을 분석하고, 요청사항에 맞춰 새로운 카피 {num_con}개를 작성해주세요. 2. 만약 참고 카피가 없다면, 요청사항과 타겟 고객, 브랜드 톤, 창의성 수준에만 집중하여 작성해주세요. 3. 각 카피는 한국어로 자연스럽고 매력적이어야 합니다. 4. {target_aud}에게 어필할 수 있는 표현을 사용해주세요. 5. {brand_tn} 톤앤매너를 유지해주세요. **출력 형식 (각 카피와 간단한 설명 포함):** 1. [생성된 카피 1] - 설명: (이 카피가 왜 효과적인지 또는 어떤 의도로 작성되었는지) 2. [생성된 카피 2] - 설명: (이 카피가 왜 효과적인지 또는 어떤 의도로 작성되었는지) ... (요청한 컨셉 수만큼 반복) **추천 카피:** (위 생성된 카피 중 가장 추천하는 것 하나와 그 이유) """ response = loaded_model.generate_content(prompt) progress_bar.progress(100) status_text_gen.text("✅ 완료!") time.sleep(0.5) progress_bar.empty() status_text_gen.empty() return { 'references': reference_copies, 'generated_content': response.text, 'search_info': { 'query': search_query, 'total_candidates': len(filtered_df_gen), 'selected_references': len(reference_copies) }, 'settings': { 'category': category_filter, 'target': target_aud, 'tone': brand_tn, 'creative': creative_lvl } } except Exception as e_gen: st.error(f"❌ 카피 생성 실패: {e_gen}") st.error(f"오류 타입: {type(e_gen)}") # 오류 타입 출력 import traceback # 상세 트레이스백 st.error(traceback.format_exc()) progress_bar.empty() status_text_gen.empty() return None # 생성 버튼 if st.button("🚀 카피 생성하기", type="primary", use_container_width=True, key="generate_button"): if not user_request or not user_request.strip(): st.error("❌ 카피 요청을 입력해주세요") else: result = generate_copy_with_rag( user_req=user_request, category_filter=selected_category, target_aud=target_audience, brand_tn=brand_tone, creative_lvl=creative_level, num_con=num_concepts ) if result: st.markdown("## 🎉 생성된 카피") st.markdown("---") st.info(f"🔍 **검색 정보**: {result['search_info']['total_candidates']:,}개 후보에서 " f"{result['search_info']['selected_references']}개 참고 카피 선별") if show_references and result['references']: with st.expander("📚 참고한 카피들 (의미적 유사도 기반 선별)"): for i, ref in enumerate(result['references'], 1): st.markdown(f"**{i}.** \"{ref['copy']}\"") st.markdown(f" - 브랜드: {ref['brand']}") st.markdown(f" - 유사도: {ref['similarity']:.3f}") st.markdown("") st.markdown("### ✨ AI가 생성한 카피:") st.markdown(result['generated_content']) try: result_json = json.dumps({ 'timestamp': datetime.now().isoformat(), 'request': user_request, 'settings': result['settings'], 'search_info': result['search_info'], 'generated_content': result['generated_content'], 'references': result['references'] # 참고 카피도 JSON에 포함 }, ensure_ascii=False, indent=2) st.download_button( label="💾 결과 다운로드 (JSON)", data=result_json, file_name=f"copy_result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", mime="application/json", key="download_button" ) except Exception as e_json: st.error(f"❌ 결과 다운로드 파일 생성 실패: {e_json}") # 시스템 정보 (사이드바 하단) st.sidebar.markdown("---") st.sidebar.markdown("### 📊 RAG 시스템 정보") if loaded_df is not None and loaded_embeddings is not None: st.sidebar.markdown(f"**카피 데이터**: {len(loaded_df):,}개") st.sidebar.markdown(f"**카테고리**: {loaded_df['카테고리'].nunique()}개") st.sidebar.markdown(f"**브랜드**: {loaded_df['브랜드'].nunique()}개") st.sidebar.markdown(f"**임베딩**: {loaded_embeddings.shape[1]}차원") # 로드된 임베딩 사용 st.sidebar.markdown("**검색 엔진**: Korean SBERT") st.sidebar.markdown("**호스팅**: 🤗 Hugging Face") # 사용법 가이드 with st.expander("💡 RAG 시스템 사용법 가이드"): st.markdown(""" ### 🎯 효과적인 사용법 (기존 내용과 동일) """) # 푸터 st.markdown("---") st.markdown( "💡 **AI 카피라이터** | 37,671개 실제 광고 카피 데이터 기반 | " "RAG(검색 증강 생성) 시스템 powered by Korean SBERT + Gemini AI" ) # 성능 모니터링 (개발자용) if os.getenv("DEBUG_MODE") == "true": # 환경변수 값을 문자열 "true"로 비교 st.sidebar.markdown("### 🔧 디버그 정보 (활성화됨)") if 'loaded_embeddings' in locals() and loaded_embeddings is not None: # 로드된 변수 사용 st.sidebar.write(f"임베딩 메모리: {loaded_embeddings.nbytes / (1024*1024):.1f}MB") st.sidebar.write(f"Streamlit 버전: {st.__version__}") st.sidebar.write(f"Pandas 버전: {pd.__version__}") st.sidebar.write(f"Numpy 버전 (Global): {np.__version__ if 'np' in globals() else 'Not imported globally'}") st.sidebar.write(f"Torch 버전: {torch.__version__ if 'torch' in globals() else 'Torch not directly used here'}") # torch는 sentence-transformers 내부 사용 st.sidebar.write(f"google-generativeai 버전: {genai.__version__}")