import streamlit as st import os from pinecone import Pinecone from sentence_transformers import SentenceTransformer from typing import List, Dict import re # For parsing timestamp and extracting video ID import streamlit.components.v1 as components # For embedding HTML from openai import OpenAI # Import OpenAI library import logging # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Helper Functions (Existing: parse_timestamp_to_seconds, get_youtube_video_id, add_timestamp_to_youtube_url, generate_youtube_embed_html) --- def parse_timestamp_to_seconds(timestamp: str) -> int | None: """HH:MM:SS 또는 HH:MM:SS.ms 형식의 타임스탬프를 초 단위로 변환합니다.""" if not isinstance(timestamp, str): return None # Remove milliseconds part if present timestamp_no_ms = timestamp.split('.')[0] parts = timestamp_no_ms.split(':') try: if len(parts) == 3: h, m, s = map(int, parts) return h * 3600 + m * 60 + s elif len(parts) == 2: m, s = map(int, parts) return m * 60 + s elif len(parts) == 1: return int(parts[0]) else: return None except ValueError: return None def get_youtube_video_id(url: str) -> str | None: """YouTube URL에서 비디오 ID를 추출합니다.""" if not isinstance(url, str): return None # Standard YouTube URLs (youtube.com/watch?v=...), shortened URLs (youtu.be/...), etc. match = re.search(r"(?:v=|/|youtu\.be/|embed/|shorts/)([0-9A-Za-z_-]{11})", url) return match.group(1) if match else None def add_timestamp_to_youtube_url(youtube_url: str, timestamp: str) -> str: """YouTube URL에 타임스탬프를 추가합니다.""" seconds = parse_timestamp_to_seconds(timestamp) if seconds is None or not youtube_url: return youtube_url # Return original URL if timestamp is invalid or URL is empty separator = '&' if '?' in youtube_url else '?' # Remove existing t= parameter if present cleaned_url = re.sub(r'[?&]t=\d+s?', '', youtube_url) separator = '&' if '?' in cleaned_url else '?' # Re-check separator after cleaning return f"{cleaned_url}{separator}t={seconds}s" def generate_youtube_embed_html(youtube_url: str, timestamp: str) -> str | None: """타임스탬프가 적용된 YouTube 임베드 HTML 코드를 생성합니다. 가로 800px 고정, 세로 자동 조절.""" video_id = get_youtube_video_id(youtube_url) start_seconds = parse_timestamp_to_seconds(timestamp) if not video_id: logger.warning(f"Could not extract video ID from URL: {youtube_url}") return None # Cannot generate embed code without video ID start_param = f"start={start_seconds}" if start_seconds is not None else "" # Use aspect ratio approach with fixed width 800px return f'''
''' # --- 설정 --- # Pinecone 설정 PINECONE_API_KEY = os.getenv("PINECONE_API_KEY","pcsk_PZHLK_TRAvMCyNmJM4FKGCX7rbbY22a58fhnWYasx1mf3WL6sRasoASZXfsbnJYvCQ13w") # Load from environment variable PINECONE_ENV = os.getenv("PINECONE_ENV", "us-east-1") INDEX_NAME = "video-embeddings" EMBEDDING_MODEL = "jhgan/ko-sroberta-multitask" # OpenAI 설정 OPENAI_API_KEY = "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" # --- 리소스 로딩 (캐싱 활용) --- @st.cache_resource def init_pinecone(): """Pinecone 클라이언트를 초기화합니다.""" api_key = PINECONE_API_KEY if not api_key: st.error("Pinecone API 키가 설정되지 않았습니다. 환경 변수를 확인하세요.") st.stop() try: pc = Pinecone(api_key=api_key) logger.info("Successfully connected to Pinecone.") return pc except Exception as e: st.error(f"Pinecone 초기화 중 오류 발생: {e}") st.stop() @st.cache_resource def load_embedding_model(): """Sentence Transformer 모델을 로드합니다.""" try: model = SentenceTransformer("my_model") logger.info(f"Successfully loaded embedding model: {EMBEDDING_MODEL}") return model except Exception as e: st.error(f"임베딩 모델 로딩 중 오류 발생: {e}") st.stop() @st.cache_resource def get_pinecone_index(_pc: Pinecone, index_name: str): """Pinecone 인덱스 객체를 가져옵니다.""" try: index = _pc.Index(index_name) # Optionally, do a quick check like index.describe_index_stats() to confirm connection stats = index.describe_index_stats() logger.info(f"Successfully connected to Pinecone index '{index_name}'. Stats: {stats.get('total_vector_count', 'N/A')} vectors") return index except Exception as e: st.error(f"Pinecone 인덱스 '{index_name}' 연결 중 오류 발생: {e}. 인덱스가 존재하고 활성 상태인지 확인하세요.") st.stop() @st.cache_resource def init_openai_client(): """OpenAI 클라이언트를 초기화합니다.""" if not OPENAI_API_KEY: st.error("OpenAI API 키가 설정되지 않았습니다. 환경 변수를 확인하세요.") st.stop() try: client = OpenAI(api_key=OPENAI_API_KEY) # Test connection (optional, but recommended) client.models.list() logger.info("Successfully connected to OpenAI.") return client except Exception as e: st.error(f"OpenAI 클라이언트 초기화 또는 연결 테스트 중 오류 발생: {e}") st.stop() # --- 검색 함수 --- def search(query: str, top_k: int = 5, _index=None, _model=None) -> List[Dict]: """Pinecone 인덱스에서 검색을 수행하고 title과 original_text를 포함합니다.""" if not query or _index is None or _model is None: return [] try: query_vec = _model.encode(query, convert_to_numpy=True).tolist() result = _index.query(vector=query_vec, top_k=top_k, include_metadata=True) matches = result.get("matches", []) search_results = [] for m in matches: metadata = m.get("metadata", {}) search_results.append({ "URL": metadata.get("url", "N/A"), "타임스탬프": metadata.get("timestamp", "N/A"), "타입": metadata.get("type", "N/A"), "제목": metadata.get("title", "N/A"), # 제목 추가 "요약": metadata.get("summary", "N/A"), "원본텍스트": metadata.get("original_text", "N/A"), # 컨텍스트로 활용할 원본 텍스트 "점수": m.get("score", 0.0) }) logger.info(f"Pinecone search returned {len(search_results)} results for query: '{query[:50]}...'") return search_results except Exception as e: st.error(f"Pinecone 검색 중 오류 발생: {e}") logger.error(f"Error during Pinecone search: {e}", exc_info=True) return [] # --- OpenAI 답변 생성 함수 --- def generate_khan_answer(query: str, search_results: List[Dict], client: OpenAI) -> str: """사용자 질문과 검색 결과를 바탕으로 Khan 페르소나 답변을 생성합니다.""" if not search_results: # Return a persona-consistent message even when no results are found return "현재 질문에 대해 참고할 만한 관련 영상을 찾지 못했습니다. 질문을 조금 더 명확하게 해주시거나 다른 방식으로 질문해주시면 도움이 될 것 같습니다." # Build context string for OpenAI more robustly, including timestamped URL context_parts = [] for i, r in enumerate(search_results): original_text_snippet = "" if r.get('원본텍스트'): snippet = r['원본텍스트'][:200] original_text_snippet = f"\n(원본 내용 일부: {snippet}...)" # Generate timestamped URL if possible timestamped_url_str = "N/A" url = r.get('URL', 'N/A') timestamp = r.get('타임스탬프', 'N/A') is_youtube = url and isinstance(url, str) and ('youtube.com' in url or 'youtu.be' in url) has_valid_timestamp = timestamp and timestamp != 'N/A' and parse_timestamp_to_seconds(timestamp) is not None if is_youtube and has_valid_timestamp: try: timestamped_url_str = add_timestamp_to_youtube_url(url, timestamp) except Exception: timestamped_url_str = url # Fallback to original URL on error elif url != "N/A": timestamped_url_str = url # Use original URL if not YouTube/no timestamp context_parts.append( f"관련 정보 {i+1}:\n" f"제목: {r.get('제목', 'N/A')}\n" f"영상 URL (원본): {url}\n" f"타임스탬프: {timestamp}\n" f"타임스탬프 적용 URL: {timestamped_url_str}\n" # Add the timestamped URL here f"내용 타입: {r.get('타입', 'N/A')}\n" f"요약: {r.get('요약', 'N/A')}" f"{original_text_snippet}" # Append the snippet safely ) context = "\n\n---\n\n".join(context_parts) # Join the parts # Updated system prompt to instruct Markdown link usage system_prompt = """너는 현실적인 조언을 잘하는 PM 멘토 Khan이다. - 말투는 단호하지만 공감력이 있다. "~입니다." 또는 "~죠."와 같이 명확하게 끝맺는다. 존댓말을 사용한다. - 완곡한 표현을 활용하며, 상대방의 감정을 함부로 단정 짓지 않는다. 예: "그럴 수 있습니다", "음, 그렇게 느낄 수 있죠" 등. - 단순한 위로보다는 구조적이고 실용적인 제안을 우선한다. 질문자가 놓친 맥락이나 구조를 짚어주고, 다음 단계 또는 전략적 선택지를 제시한다. - 질문이 막연하거나 추상적이면, 핵심을 좁혀 다시 되물어본다. 예: "그 상황에서 가장 답답했던 순간은 언제였나요?"와 같이 질문을 구체화한다. - 긴 설명보다는 핵심을 빠르게 전달한다. 다만, 필요한 경우 짧은 비유나 예시로 직관적인 이해를 돕는다. - 답변 중 관련 정보를 참조할 때는, 반드시 '타임스탬프 적용 URL'을 사용하여 다음과 같은 Markdown 링크 형식으로 제시해야 한다: `[영상 제목](타임스탬프_적용_URL)`. 예: "자세한 내용은 [비개발자가 연봉 2억을 받는 현실적인 방법](https://www.youtube.com/watch?v=VIDEO_ID&t=178s) 영상을 참고하시면 도움이 될 겁니다." - 이전 대화 기록은 없으므로, 반복 질문이 들어올 경우에는 "이전에 유사한 내용을 찾아봤었죠. 다시 한번 살펴보면..."처럼 자연스럽게 이어간다. - 답변은 반드시 한국어로 한다. Khan은 전략적으로 사고하며, 본질과 방향을 중시한다. 단정적으로 단언하기보다는 "~일 수도 있습니다", "그렇게도 볼 수 있죠"와 같이 여지를 남긴다. 상대방이 스스로 선택지를 판단할 수 있도록 돕는 방향으로 조언한다. 예시처럼 말투와 사고 흐름을 유지해야 한다: --- Q: 요즘 팀원과의 관계가 어려운데, 제가 뭘 놓치고 있는 걸까요? A: 음, 그럴 수 있습니다. 관계가 어려울 때는 감정보다는 기대가 엇갈렸던 순간을 먼저 봐야 하죠. 그 팀원이 무언가를 기대했는데, 내가 그걸 놓쳤을 가능성이 있습니다. 혹시 최근에 서로 오해가 생긴 순간이 있었는지, 먼저 짚어보는 게 좋겠습니다. --- Q: 회사를 옮기고 싶은데, 성과 없이 퇴사하면 안 좋을까요? A: 단기적으로는 맞습니다. 성과 없이 퇴사하면 이력서에 남죠. 하지만 지금 상황에서 배울 게 없다면, 그 자체가 리스크이기도 합니다. '내가 남아서 얻을 수 있는 게 무엇인가'와 '지금 나가서 시작할 수 있는 게 무엇인가'를 나란히 두고 비교해 보시죠. --- 이런 식의 말투와 흐름을 바탕으로 질문에 답변하세요.""" # Use triple quotes for the multi-line f-string user_message = f"""사용자 질문: {query} 아래 관련 정보를 바탕으로 Khan 멘토로서 답변해주세요: {context}""" try: logger.info("Calling OpenAI API...") completion = client.chat.completions.create( model="gpt-4o-mini", # Use gpt-4 if available and preferred messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ], temperature=0.5, # Slightly less creative, more focused on instructions ) answer = completion.choices[0].message.content logger.info("Received response from OpenAI.") return answer.strip() except Exception as e: st.error(f"OpenAI 답변 생성 중 오류 발생: {e}") logger.error(f"Error during OpenAI API call: {e}", exc_info=True) return "답변을 생성하는 중에 문제가 발생했습니다. OpenAI API 키 또는 서비스 상태를 확인해주세요." # --- Streamlit 앱 UI (Khan 멘토 단일 루프 구조) --- st.set_page_config(page_title="Khan 멘토 (PM 영상 기반)", layout="wide") # --- 사이드바 메뉴 --- menu = st.sidebar.radio( "기능 선택", ("Khan 멘토에게 상담하기", "상사에게 잘보이기") ) # 사이드바 맨 아래에 설문조사 링크 st.sidebar.markdown('
', unsafe_allow_html=True) st.sidebar.markdown( '📝 서비스 어떻게 생각하세요?', unsafe_allow_html=True ) openai_client = init_openai_client() if menu == "Khan 멘토에게 상담하기": st.title("✨ Khan 멘토가 24시간 답변중입니다") # --- API 키 확인 및 리소스 초기화 --- pc = init_pinecone() model = load_embedding_model() index = get_pinecone_index(pc, INDEX_NAME) # --- 상태 관리 --- if 'user_question' not in st.session_state: st.session_state['user_question'] = '' if 'empathy_message' not in st.session_state: st.session_state['empathy_message'] = '' if 'khan_answer' not in st.session_state: st.session_state['khan_answer'] = '' if 'pinecone_results' not in st.session_state: st.session_state['pinecone_results'] = [] if 'extra_questions' not in st.session_state: st.session_state['extra_questions'] = [] if 'current_input' not in st.session_state: st.session_state['current_input'] = '' # --- 질문 입력 및 답변 생성 --- st.markdown("#### 당신의 고민을 알려 주세요!") user_q = st.text_input( "나의 고민은...", value=st.session_state['current_input'], key="main_input", placeholder="프로덕트 매니저가 가져야 할 역량은 어떤 것이 있을까요?" ) if st.button("고민 나누기", key="main_ask") or (user_q and st.session_state['user_question'] != user_q): st.session_state['user_question'] = user_q st.session_state['current_input'] = user_q # 1. 공감 비서 메시지 with st.spinner("생각중..."): empathy_prompt = f""" 너는 따뜻하고 친절한 비서야. 아래 사용자의 질문을 듣고, 감정적으로 충분하게 1~2문장으로 공감해주되 질문에 대한 답변은 하지마, 마지막에 '칸 멘토의 생각을 들어볼까요?'라고 안내해줘. \n질문: "{user_q}" """ try: empathy_response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "system", "content": empathy_prompt}], temperature=0.7, ) st.session_state['empathy_message'] = empathy_response.choices[0].message.content.strip() except Exception as e: st.session_state['empathy_message'] = f"공감 메시지 생성 중 오류: {e}" # 2. Pinecone 검색 및 Khan 멘토 답변 with st.spinner("Khan 멘토가 뜸을 들이며..."): pinecone_results = search(user_q, top_k=5, _index=index, _model=model) st.session_state['pinecone_results'] = pinecone_results khan_answer = generate_khan_answer(user_q, pinecone_results, openai_client) st.session_state['khan_answer'] = khan_answer # 3. 추가 질문 생성 with st.spinner("추가 질문을 생성하는 중..."): extra_prompt = ( f"아래 질문에서 유사하게 궁금할 수 있는 추가 질문 3~4개를 한국어로 만들어줘. 지나치게 세부적인 툴에 대한 얘기보다는 프로덕트, 프로젝트, 리더십에 대한 일반적인 질문으로 만들어. 질문: \"{st.session_state['user_question']}" ) try: extra_response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "system", "content": extra_prompt}], temperature=0.5 ) import re raw = extra_response.choices[0].message.content.strip() questions = re.findall(r'\d+\.\s*(.+)', raw) if not questions: questions = [q.strip('-• ').strip() for q in raw.split('\n') if q.strip()] st.session_state['extra_questions'] = questions[:4] st.rerun() except Exception as e: st.session_state['extra_questions'] = [f"추가 질문 생성 중 오류: {e}"] st.rerun() # --- 답변 및 추가질문 UI --- if st.session_state['user_question']: st.info(st.session_state['empathy_message']) st.subheader("💡 Khan 멘토의 답변") st.markdown(st.session_state['khan_answer']) # 참고 영상 정보 표시 pinecone_results = st.session_state['pinecone_results'] if pinecone_results: with st.expander("답변에 참고한 영상 정보 보기", expanded=True): displayed_urls = set() for i, r in enumerate(pinecone_results): url = r.get('URL', 'N/A') if url in displayed_urls or url == 'N/A': continue displayed_urls.add(url) st.markdown(f"--- **참고 자료 {len(displayed_urls)} (유사도: {r['점수']:.4f})** ---") st.markdown(f"**제목:** {r.get('제목', 'N/A')}") st.markdown(f"**요약:** {r.get('요약', 'N/A')}") timestamp = r.get('타임스탬프', 'N/A') is_youtube = url and isinstance(url, str) and ('youtube.com' in url or 'youtu.be' in url) start_seconds = None if is_youtube and timestamp and timestamp != 'N/A': start_seconds = parse_timestamp_to_seconds(timestamp) if is_youtube and start_seconds is not None: try: timestamped_link_url = add_timestamp_to_youtube_url(url, timestamp) st.markdown(f"**영상 링크 (타임스탬프 포함):** [{timestamped_link_url}]({timestamped_link_url})") except Exception as e: logger.error(f"Error creating timestamped URL for link: {e}") st.markdown(f"**영상 링크 (원본):** [{url}]({url})") elif url != "N/A" and isinstance(url, str) and url.startswith("http"): st.markdown(f"**URL:** [{url}]({url})") else: st.markdown(f"**URL:** {url}") if is_youtube and url != "N/A": col1, col2 = st.columns(2) with col1: try: st.video(url, start_time=start_seconds or 0) except Exception as e: st.error(f"비디오({url}) 재생 중 오류 발생: {e}") st.markdown(f"[YouTube에서 보기]({url})") elif url != "N/A": col1, col2 = st.columns(2) with col1: try: st.video(url) except Exception as e: logger.warning(f"st.video failed for non-YouTube URL {url}: {e}") st.markdown("---") # --- 추가 질문 생성 타이밍 제어 --- if 'extra_questions_ready' not in st.session_state or not st.session_state['extra_questions_ready']: # 답변이 렌더링된 후에만 spinner 돌리기 st.session_state['extra_questions_ready'] = True st.rerun() elif not st.session_state['extra_questions']: # 백그라운드에서 추가 질문 생성 (스피너 없이) extra_prompt = ( f"아래 질문에서 유사하게 궁금할 수 있는 추가 질문 3~4개를 한국어로 만들어줘. 지나치게 세부적인 툴에 대한 얘기보다는 프로덕트, 프로젝트, 리더십에 대한 일반적인 질문으로 만들어. 질문: \"{st.session_state['user_question']}" ) try: extra_response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "system", "content": extra_prompt}], temperature=0.5 ) import re raw = extra_response.choices[0].message.content.strip() questions = re.findall(r'\d+\.\s*(.+)', raw) if not questions: questions = [q.strip('-• ').strip() for q in raw.split('\n') if q.strip()] st.session_state['extra_questions'] = questions[:4] st.rerun() except Exception as e: st.session_state['extra_questions'] = [f"추가 질문 생성 중 오류: {e}"] else: st.markdown("#### 추가로 궁금한 점이 있으신가요? 아래 예시 질문을 클릭하거나 직접 입력해보세요!") cols = st.columns(len(st.session_state['extra_questions'])) for i, q in enumerate(st.session_state['extra_questions']): if cols[i].button(q, key=f"extra_{i}"): st.session_state['current_input'] = q st.session_state['user_question'] = '' st.rerun() user_extra = st.text_input("직접 추가 질문 입력", value="", key="extra_input") if st.button("추가 질문하기", key="extra_btn"): st.session_state['current_input'] = user_extra st.session_state['user_question'] = '' st.rerun() st.markdown("---") st.caption("Powered by Pinecone, Sentence Transformers, and OpenAI") if st.button("다른 고민 상담하기"): for k in ['user_question','empathy_message','khan_answer','pinecone_results','extra_questions','current_input','extra_questions_ready']: st.session_state[k] = '' st.rerun() else: st.title("👔 상사에게 잘보이기: 맞춤 보고문 만들기") st.markdown("상사의 MBTI 성향에 맞게 보고문을 자동으로 다듬어드립니다.") mbti_types = [ "ISTJ", "ISFJ", "INFJ", "INTJ", "ISTP", "ISFP", "INFP", "INTP", "ESTP", "ESFP", "ENFP", "ENTP", "ESTJ", "ESFJ", "ENFJ", "ENTJ" ] mbti = st.selectbox("상사의 MBTI를 선택하세요", mbti_types) user_report = st.text_area("상사에게 보고할 내용을 입력하세요 (300자 이내)", max_chars=300) if st.button("MBTI 맞춤 보고문 생성"): if not user_report.strip(): st.warning("보고문을 입력해 주세요.") else: with st.spinner("상사의 성향에 맞게 보고문을 다듬는 중..."): prompt = f""" 상사의 MBTI가 {mbti}일 때, 아래 보고문을 그 성향에 맞게 수정해줘.\n 이 유형의 상사가 중요하게 생각하는 것이 보고서에 빠져있다면 어떤 부분을 보완해야 하는지 상세히 설명해 줘\n그리고 왜 그렇게 수정했는지 이유도 설명해줘.\n보고문: "{user_report}" \n아래 형식으로 답변 해.\n수정된 보고문: ...\n이유: ... """ try: response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "system", "content": prompt}], temperature=0.5, ) answer = response.choices[0].message.content.strip() # 간단한 파싱: "수정된 보고문:" ~ "이유:" 분리 import re mod_match = re.search(r"수정된 보고문[:\n]*([\s\S]+?)이유[:\n]", answer) reason_match = re.search(r"이유[:\n]*([\s\S]+)", answer) if mod_match: st.markdown(f"**수정된 보고문**\n\n{mod_match.group(1).strip()}") logger.info(f"[MBTI 보고문] 수정된 보고문: {mod_match.group(1).strip()}") else: st.markdown(f"**수정된 보고문**\n\n{answer}") logger.info(f"[MBTI 보고문] 수정된 보고문: {answer}") if reason_match: st.markdown(f"**이유 설명**\n\n{reason_match.group(1).strip()}") logger.info(f"[MBTI 보고문] 이유 설명: {reason_match.group(1).strip()}") except Exception as e: st.error(f"GPT 호출 중 오류: {e}")