import gradio as gr import datetime import uuid from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from sentence_transformers import SentenceTransformer from langchain.prompts import PromptTemplate from langchain.chains import LLMChain from langchain_groq import ChatGroq import os # Qdrant 설정 QDRANT_URL = os.getenv("QDRANT_URL") QDRANT_API_KEY = os.getenv("QDRANT_API_KEY") COLLECTION_NAME = os.getenv("COLLECTION_NAME") # OpenAI API 키 (Groq도 OpenAI 호환이면 동일하게 사용) GROQ_API_URL =os.getenv("GROQ_API_URL") GROQ_API_KEY = os.getenv("GROQ_API_KEY") # 임베딩 모델 model = SentenceTransformer('sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2') # Qdrant 클라이언트 client = QdrantClient( url=QDRANT_URL, api_key=QDRANT_API_KEY, ) # LangChain LLM (Groq) llm = ChatGroq( groq_api_key=GROQ_API_KEY, # 기존 OPENAI_API_KEY 변수에 Groq 키를 넣으세요 model="gemma2-9b-it", # Groq에서 지원하는 모델명으로 교체 temperature=0.7, max_tokens=1000 ) # LangChain 프롬프트 템플릿 prompt_template = """ Below is a user's recently written diary entry. Empathize with the user's feelings, praise positive expressions and efforts, and encourage them to keep writing diaries. Identify and explain any cognitive distortions (e.g., black-and-white thinking, overgeneralization, emotional reasoning) or recurring emotional patterns in the diary. If there are similar past diary entries, point out any psychological traps or thinking errors the user repeatedly falls into based on those records. After that, kindly and realistically advise how the user can interpret the situation in a healthier way and suggest specific actions or how to write the next diary entry. The analysis should be as sincere as a professional counselor, but the tone should be warm, friendly, and like talking to a 12-year-old friend. The main focus should be on analyzing the [### Diary Entry] just written, and [### Similar Past Diaries] should be used as reference. #### 1. 🧾 Output Format 1. Empathy & Praise - Empathize with the user's feelings. - Praise positive expressions, efforts, and the act of writing a diary. - Encourage the user to continue writing diaries. 2. Cognitive Distortion/Emotional Pattern Analysis - Based on the diary sentences, explain any cognitive distortions (e.g., black-and-white thinking, overgeneralization, emotional reasoning). - If there are recurring emotions or thought patterns, point them out specifically. 3. Similar Past Diary Comparison - Summarize similar past diaries in the following format: - Entry Number / Title / Summary / Date / Similarity / Reason for Similarity Example: 1. Title: ... - Summary: ... - Date: ... - Similarity: ... - Reason for Similarity: ... 2. Title: ... - Summary: ... - Date: ... - Similarity: ... - Reason for Similarity: ... 4. Psychological Trap Summary - Summarize the psychological traps or thinking patterns the user frequently falls into compared to past diaries. (e.g., "Frequently judges oneself as incompetent," "Tends to overinterpret others' reactions.") 5. Kind and Realistic Advice - Kindly explain how to interpret the situation to reduce self-blame. - Suggest specific actions or how to write the next diary entry. - Use a tone as if talking to a 12-year-old friend, but with sincere counseling content. ### Diary Entry: {user_journal} ### Similar Past Diaries: {retrieved_journals} """ prompt = PromptTemplate( input_variables=["user_journal", "retrieved_journals"], template=prompt_template ) # LangChain 체인 chain = LLMChain(llm=llm, prompt=prompt) def ensure_collection_exists(): try: collections = client.get_collections() collection_names = [col.name for col in collections.collections] if COLLECTION_NAME not in collection_names: client.create_collection( collection_name=COLLECTION_NAME, vectors_config=VectorParams(size=384, distance=Distance.COSINE), ) except Exception as e: pass def save_diary_to_qdrant(title, content): if not title or not content: return "Please enter both a title and content." try: ensure_collection_exists() full_text = f"Title: {title}\nContent: {content}" vector = model.encode(full_text).tolist() timestamp = datetime.datetime.now() point = PointStruct( id=str(uuid.uuid4()), vector=vector, payload={ "title": title, "content": content, "timestamp": timestamp.isoformat(), "date": timestamp.strftime("%Y-%m-%d"), "full_text": full_text } ) client.upsert( collection_name=COLLECTION_NAME, points=[point] ) save_message = f"Diary entry saved! \nSaved at: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}" # Chatbot analysis message analysis_message = chat_with_diary_context(content, []) # Paragraph separation analysis_message = analysis_message.replace('\n\n', '\n\n').replace('\n', '\n\n') return save_message + "\n\n---\n\n" + analysis_message except Exception as e: return f"An error occurred while saving: {str(e)}" def search_diary_for_chat(query, limit=2): if not query: return [] try: query_vector = model.encode(query).tolist() search_results = client.search( collection_name=COLLECTION_NAME, query_vector=query_vector, limit=limit ) diary_data = [] for result in search_results: payload = result.payload diary_data.append({ "title": payload['title'], "content": payload['content'], "date": payload['date'], "score": result.score }) return diary_data except Exception as e: return [] def chat_with_diary_context(message, history): # 과거 유사 일기 검색 diary_context = search_diary_for_chat(message, limit=2) user_journal = message if diary_context: retrieved_journals = "\n".join([ f"[{d['date']}] {d['title']}: {d['content'][:200]}..." for d in diary_context ]) else: retrieved_journals = "(유사 일기 없음)" # LangChain 체인 실행 result = chain.run(user_journal=user_journal, retrieved_journals=retrieved_journals) return result def search_diary(query, limit=5): if not query: return "Please enter a search keyword." try: query_vector = model.encode(query).tolist() search_results = client.search( collection_name=COLLECTION_NAME, query_vector=query_vector, limit=limit ) if not search_results: return "No search results found." results = [] for result in search_results: payload = result.payload score = result.score results.append( f"**{payload['title']}** (Similarity: {score:.3f})\nDate: {payload['date']}\nContent: {payload['content'][:100]}...\n\n" ) return "\n".join(results) except Exception as e: return f"An error occurred during search: {str(e)}" def create_interface(): with gr.Blocks(title="📝 Refeel - Re-feel your emotions. Reframe your story.") as demo: gr.Markdown("# 📝 Refeel - Re-feel your emotions. Reframe your story.") with gr.Tabs() as tabs: with gr.Tab("✏️ Write Diary") as tab_write: with gr.Column(): diary_title = gr.Textbox( label="Diary Title", placeholder="Enter the title of your diary entry" ) diary_content = gr.Textbox( label="Diary Content", placeholder="Write freely about your day, thoughts, and feelings", lines=10 ) save_btn = gr.Button("Save Diary", variant="primary") save_result = gr.Markdown(label="Save Result") more_chat_btn = gr.Button("Continue Chatting", visible=False) def save_and_show_more(title, content): result = save_diary_to_qdrant(title, content) notice = "\n\n👉 Click the '🤖 Smart Chat' tab above to continue chatting with the AI!" return result + notice, gr.update(visible=True) save_btn.click( save_and_show_more, inputs=[diary_title, diary_content], outputs=[save_result, more_chat_btn] ) def more_chat_notice(): return gr.update(value="Click the '🤖 Smart Chat' tab above!") more_chat_btn.click(more_chat_notice, None, save_result) with gr.Tab("🔍 Search Diary"): with gr.Column(): search_query = gr.Textbox( label="Search Keyword", placeholder="Enter a keyword to search your diary entries" ) search_limit = gr.Slider( minimum=1, maximum=10, value=5, step=1, label="Number of Search Results" ) search_btn = gr.Button("Search", variant="primary") search_results = gr.Markdown(label="Search Results") search_btn.click( search_diary, inputs=[search_query, search_limit], outputs=search_results ) with gr.Tab("🤖 Smart Chat"): gr.Markdown("### AI Assistant that chats with you based on your diary (LangChain)") chatbot = gr.Chatbot( height=400, type="messages" ) msg = gr.Textbox( label="Message", placeholder="Talk about your diary or ask for advice..." ) clear = gr.Button("Clear Chat") def user(user_message, history): return "", history + [{"role": "user", "content": user_message}] def bot(history): if not history: return history user_message = history[-1]["content"] bot_message = chat_with_diary_context(user_message, history[:-1]) return history + [{"role": "assistant", "content": bot_message}] msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then( bot, chatbot, chatbot ) clear.click(lambda: [], None, chatbot, queue=False) return demo demo = create_interface() if __name__ == "__main__": demo.launch() def search_diary_by_date(selected_date): try: search_results = client.scroll( collection_name=COLLECTION_NAME, scroll_filter={ "must": [ {"key": "date", "match": {"value": selected_date}} ] }, limit=20 ) if not search_results or not search_results[0]: return f"No diary entries found for {selected_date}." results = [] for result in search_results[0]: payload = result.payload results.append( f"**{payload['title']}**\nDate: {payload['date']}\nContent: {payload['content']}\n" ) return '\n---\n'.join(results) except Exception as e: return f"An error occurred during search: {str(e)}"