import gradio as gr from openai import OpenAI from huggingface_hub import InferenceClient import os import time import requests import urllib.parse import pandas as pd from langchain_community.document_loaders import DataFrameLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_huggingface import HuggingFaceEmbeddings from langchain_community.vectorstores import FAISS from PIL import Image import warnings # 忽略 pandas 的一些警告 warnings.filterwarnings("ignore") # ========================================== # 0. 環境變數與 RAG 初始化 # ========================================== GROQ_API_KEY = os.getenv("GROQ_API_KEY") HF_TOKEN = os.getenv("HF_TOKEN") # --- 1. 資料讀取 (主要資料庫) --- df = pd.DataFrame() try: df = pd.read_csv('google_maps_rag_data.csv', engine='python') df.columns = df.columns.str.strip() # 清理欄位空白 def classify_mood(row): name_str = str(row.get('Name', '')) rag_str = str(row.get('RAG_Content', '')) text = (name_str + " " + rag_str).lower() tags = [] rules = { "開心/慶祝": ["牛排", "steak", "pizza", "炸", "雞排", "甜點", "蛋糕", "cake", "冰", "waffle", "吃到飽", "buffet", "burger", "bistro", "餐酒館"], "傷心/疲憊": ["粥", "湯", "warm", "congee", "麵", "noodle", "小吃", "comfort food", "豆花", "關東煮", "soup"], "生氣/發洩": ["辣", "spicy", "麻辣", "鍋", "curry", "咖哩", "燒肉", "bbq", "臭豆腐", "fry"], "平靜/放鬆": ["cafe", "coffee", "tea", "茶", "素食", "vegetable", "早午餐", "brunch", "壽司", "sushi", "居酒屋"] } for mood, keywords in rules.items(): for kw in keywords: if kw in text: tags.append(mood) if not tags: tags.append("隨意/探索") return ", ".join(list(set(tags))) if 'Mood_Tags' not in df.columns: df['Mood_Tags'] = df.apply(classify_mood, axis=1) except Exception as e: print(f"⚠️ 讀取 google_maps_rag_data.csv 失敗: {e}") # --- 2. 資料讀取 (Prompt 資料庫) --- df_prompts = pd.DataFrame() PROMPT_COL_NAME = 'Image_Prompt' try: df_prompts = pd.read_csv('restaurants_prompts_v6.csv', engine='python') df_prompts.columns = df_prompts.columns.str.strip() if 'Name' in df_prompts.columns: df_prompts.set_index('Name', inplace=True) else: print(f"⚠️ Prompts CSV 警告: 找不到 'Name' 欄位") if PROMPT_COL_NAME not in df_prompts.columns: print(f"⚠️ Prompts CSV 警告: 找不到 '{PROMPT_COL_NAME}' 欄位") except FileNotFoundError: print("⚠️ 找不到 restaurants_prompts_v6.csv") except Exception as e: print(f"⚠️ 讀取 Prompt CSV 發生錯誤: {e}") # --- RAG 建置 --- retriever = None if not df.empty and 'RAG_Content' in df.columns: try: loader = DataFrameLoader(df, page_content_column="RAG_Content") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) docs = text_splitter.split_documents(documents) embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") vectorstore = FAISS.from_documents(docs, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) except Exception as e: print(f"⚠️ RAG 初始化失敗: {e}") # ========================================== # 1. 小測驗設定 # ========================================== custom_css = """ .vertical-radio fieldset { display: flex !important; flex-direction: column !important; align-items: flex-start !important; gap: 12px !important; } .vertical-radio label { width: 100% !important; margin: 0 !important; display: flex !important; } """ img_path_1 = "images/image_0.png" img_path_2 = "images/image_1.png" if not os.path.exists("images"): os.makedirs("images") try: if not os.path.exists(img_path_1): Image.new('RGB', (400, 300), color='gray').save(img_path_1) if not os.path.exists(img_path_2): Image.new('RGB', (400, 300), color='lightblue').save(img_path_2) except: pass q1_options_map = { "不要再問白癡問題了我只想吃飯": "現實主義者 (只想吃飯)", "去過隱居生活": "隱士 (嚮往平靜)", "加入禁衛軍,保衛這個世界": "守護者 (充滿正義感)", "神羅天征,毀滅這個世界": "破壞神 (心情可能很差或很中二)" } q2_options_map = { "不要再問白癡問題了我只想吃飯": "無魔法 (飢餓度MAX)", "麵包形狀的魔法炸彈": "爆炸魔法 (喜歡刺激/重口味)", "毒氣的生化魔法": "毒氣魔法 (可能想吃臭豆腐或特殊風味)", "領域展開:無量空處": "領域展開 (思緒混亂或想放空)" } # ========================================== # 2. 核心功能函式 # ========================================== def get_restaurant_data(mood_score_str, food_choice): if df.empty: return None, True score_map = { "1 (心情差)": ["傷心/疲憊", "生氣/發洩"], "2 (不太好)": ["傷心/疲憊", "生氣/發洩"], "3 (普通)": ["平靜/放鬆", "隨意/探索"], "4 (不錯)": ["開心/慶祝", "平靜/放鬆"], "5 (超棒)": ["開心/慶祝"] } target_moods = score_map.get(mood_score_str, []) food_keyword = "" if food_choice == "吃飯": food_keyword = "飯" elif food_choice == "吃麵": food_keyword = "麵" candidates = df.copy() if 'Mood_Tags' not in candidates.columns: return df.sample(1).iloc[0], True if target_moods: pattern = "|".join(target_moods) candidates = candidates[candidates['Mood_Tags'].str.contains(pattern, regex=True, na=False)] if food_keyword: candidates = candidates[ candidates['Name'].str.contains(food_keyword, case=False, na=False) | candidates['RAG_Content'].str.contains(food_keyword, case=False, na=False) ] if candidates.empty: return df.sample(1).iloc[0], True return candidates.sample(1).iloc[0], False def generate_content_with_groq(restaurant_name, restaurant_detail, user_diary, mood_score, quiz_result): if not GROQ_API_KEY: return "⚠️ 請設定 GROQ_API_KEY (系統變數)" client = OpenAI(api_key=GROQ_API_KEY, base_url="https://api.groq.com/openai/v1") system_prompt = "你是一個幽默、懂吃且善解人意的 AI 朋友。請根據使用者的日記、心情以及「中二病測驗結果」來推薦餐廳。" user_msg = f""" 【使用者狀態】 心情分數:{mood_score} 心情日記:{user_diary} 【前測小遊戲結果】 隱藏人設:{quiz_result.get('q1', '未知')} 魔法:{quiz_result.get('q2', '未知')} 【推薦餐廳】 名稱:{restaurant_name} 資料:{restaurant_detail} 請完成任務: 【回應內容】:用繁體中文寫一段溫暖有趣的回覆。 - 結合「日記」與「前測人設」給予回應。 - 推薦上述餐廳,說明為什麼這家店適合他。 """ try: response = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": user_msg}] ) return response.choices[0].message.content except Exception as e: return f"Groq Error: {str(e)}" def generate_image_huggingface(prompt): if not HF_TOKEN: return None try: hf_client = InferenceClient(token=HF_TOKEN) return hf_client.text_to_image(prompt=prompt, model="stabilityai/stable-diffusion-xl-base-1.0") except: return None # --- 主邏輯 Agent (新增 debug_mode 參數) --- def mood_agent_logic(score_input, food_input, diary_input, quiz_state, debug_mode): restaurant, is_random = get_restaurant_data(score_input, food_input) if restaurant is None: yield "資料庫讀取錯誤或為空", None, "" return name = str(restaurant['Name']) address = str(restaurant.get('Address', '未知地址')) url = str(restaurant.get('URL', f'https://www.google.com/maps/search/?api=1&query={urllib.parse.quote(name)}')) note = "(隨機推薦)" if is_random else "" # 1. 處理 RAG rag_info = str(restaurant.get('RAG_Content', '')) if retriever: try: docs = retriever.invoke(name) if docs: rag_info = "\n".join([d.page_content for d in docs]) except: pass # 2. 準備圖片 Prompt (提前執行以便顯示除錯資訊) img_prompt = f"Delicious food from {name}, cinematic lighting, 8k, photorealistic" # 預設值 prompt_source = "⚠️ 預設生成 (未找到 CSV 對應)" if not df_prompts.empty and PROMPT_COL_NAME in df_prompts.columns: if name in df_prompts.index: try: csv_prompt = df_prompts.loc[name, PROMPT_COL_NAME] if isinstance(csv_prompt, pd.Series): csv_prompt = csv_prompt.iloc[0] if pd.notna(csv_prompt) and str(csv_prompt).strip() != "": img_prompt = str(csv_prompt) prompt_source = "✅ CSV 檔案 (restaurants_prompts_v6.csv)" else: prompt_source = "⚠️ CSV 有此餐廳但 Prompt 空白" except Exception as e: prompt_source = f"❌ 讀取錯誤: {e}" else: prompt_source = "ℹ️ CSV 中無此餐廳" # 3. 呼叫 LLM ai_text = generate_content_with_groq(name, rag_info, diary_input, score_input, quiz_state) # 4. 組合回應 (包含除錯資訊) debug_text = "" if debug_mode: debug_text = f""" \n\n--- **🛠️ Prompt 除錯模式** * **餐廳名稱**: `{name}` * **Prompt 來源**: {prompt_source} * **實際使用 Prompt**: ``` {img_prompt} ``` """ final_response = f"### 🍽️ 推薦:{name} {note}\n\n{ai_text}{debug_text}" map_html = f'
' # 5. 先回傳文字 yield final_response, None, map_html # 6. 生成圖片 image_output = generate_image_huggingface(img_prompt) yield final_response, image_output, map_html # ========================================== # 3. 頁面切換邏輯 # ========================================== def handle_q1_change(selected_label, current_state): if not selected_label: return current_state, gr.Button(interactive=False) current_state["q1"] = q1_options_map.get(selected_label, selected_label) return current_state, gr.Button(interactive=True, variant="primary") def handle_q2_change(selected_label, current_state): if not selected_label: return current_state, gr.Button(interactive=False) current_state["q2"] = q2_options_map.get(selected_label, selected_label) return current_state, gr.Button(interactive=True, variant="primary", value="完成測驗 (前往點餐) ➔") def go_to_page_2(): return gr.update(visible=False), gr.update(visible=True) def go_to_main_app(): return gr.update(visible=False), gr.update(visible=True) # ========================================== # 4. Gradio 介面建構 # ========================================== with gr.Blocks(title="AI 心情食堂") as demo: gr.HTML(f"") quiz_state = gr.State(value={}) # Page 1 with gr.Column(visible=True) as page_1_col: gr.Markdown("### 🔮 第一題:直覺測試") gr.Image(value=img_path_1, type="filepath", label="請觀察圖片", height=300) gr.Markdown("**問題:請觀察上方圖片,如果是你,你會怎麼做?**") radio_q1 = gr.Radio(choices=list(q1_options_map.keys()), label="請選擇", elem_classes="vertical-radio") btn_q1_next = gr.Button("下一頁 ➔", interactive=False) # Page 2 with gr.Column(visible=False) as page_2_col: gr.Markdown("### 🔮 第二題:魔法適性") gr.Image(value=img_path_2, type="filepath", label="請觀察圖片", height=300) gr.Markdown("**問題:身為魔導士的你,會選擇哪一個法術?**") radio_q2 = gr.Radio(choices=list(q2_options_map.keys()), label="請選擇", elem_classes="vertical-radio") btn_q2_finish = gr.Button("完成測驗 ➔", interactive=False) # Page 3 (Main) with gr.Column(visible=False) as main_app_col: gr.Markdown(f"## 🍱 AI 心情食堂導航") gr.Markdown("測驗結束!現在,告訴我你真正想吃什麼吧!") with gr.Row(): with gr.Column(scale=1): score_input = gr.Radio(["1 (心情差)", "2 (不太好)", "3 (普通)", "4 (不錯)", "5 (超棒)"], label="1. 心情分數", value="3 (普通)") food_input = gr.Radio(["吃飯", "吃麵", "隨便"], label="2. 想吃什麼", value="隨便") diary_input = gr.Textbox(lines=4, label="3. 心情日記", placeholder="寫下今天發生的事...") # [新增] 除錯模式開關 debug_mode_btn = gr.Checkbox(label="🔧 顯示 Prompt 除錯資訊", value=False) submit_btn = gr.Button("送出給 Agent", variant="primary") with gr.Column(scale=1): agent_output = gr.Markdown(label="AI 回應") image_output = gr.Image(label="AI 推薦美食圖", type="pil", width=400) map_output = gr.HTML(label="地圖導航") # Events radio_q1.change(fn=handle_q1_change, inputs=[radio_q1, quiz_state], outputs=[quiz_state, btn_q1_next]) btn_q1_next.click(fn=go_to_page_2, outputs=[page_1_col, page_2_col]) radio_q2.change(fn=handle_q2_change, inputs=[radio_q2, quiz_state], outputs=[quiz_state, btn_q2_finish]) btn_q2_finish.click(fn=go_to_main_app, outputs=[page_2_col, main_app_col]) # [修改] 加入 debug_mode_btn 到 inputs submit_btn.click( fn=mood_agent_logic, inputs=[score_input, food_input, diary_input, quiz_state, debug_mode_btn], outputs=[agent_output, image_output, map_output] ) if __name__ == "__main__": demo.launch()