| 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 |
|
|
| |
| warnings.filterwarnings("ignore") |
|
|
| |
| |
| |
| GROQ_API_KEY = os.getenv("GROQ_API_KEY") |
| HF_TOKEN = os.getenv("HF_TOKEN") |
|
|
| |
| 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}") |
|
|
| |
| 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}") |
|
|
|
|
| |
| 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}") |
|
|
| |
| |
| |
| 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)", |
| "麵包形狀的魔法炸彈": "爆炸魔法 (喜歡刺激/重口味)", |
| "毒氣的生化魔法": "毒氣魔法 (可能想吃臭豆腐或特殊風味)", |
| "領域展開:無量空處": "領域展開 (思緒混亂或想放空)" |
| } |
|
|
| |
| |
| |
|
|
| 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 |
|
|
| |
| 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 "" |
|
|
| |
| 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 |
|
|
| |
| 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 中無此餐廳" |
|
|
| |
| ai_text = generate_content_with_groq(name, rag_info, diary_input, score_input, quiz_state) |
| |
| |
| 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'<div style="text-align:center"><a href="{url}" target="_blank" style="background:#4CAF50;color:white;padding:8px 16px;border-radius:20px;text-decoration:none">🗺️ Google Map 導航</a></div>' |
| |
| |
| yield final_response, None, map_html |
|
|
| |
| image_output = generate_image_huggingface(img_prompt) |
| yield final_response, image_output, map_html |
|
|
| |
| |
| |
|
|
| 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) |
|
|
| |
| |
| |
|
|
| with gr.Blocks(title="AI 心情食堂") as demo: |
| gr.HTML(f"<style>{custom_css}</style>") |
| quiz_state = gr.State(value={}) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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="地圖導航") |
|
|
| |
| 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]) |
|
|
| |
| 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() |
