| 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, ImageFont, ImageDraw |
| import warnings |
| import cv2 |
| import numpy as np |
| from deepface import DeepFace |
|
|
| |
| 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('restaurants.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", "炸", "甜點", "蛋糕", "buffet", "bistro"], |
| "傷心/疲憊": ["粥", "湯", "warm", "麵", "小吃", "豆花", "soup"], |
| "生氣/發洩": ["辣", "spicy", "麻辣", "鍋", "curry", "燒肉", "bbq", "fry"], |
| "平靜/放鬆": ["cafe", "coffee", "tea", "素食", "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) |
| print("✅ 成功讀取 restaurants.csv") |
| except Exception as e: |
| print(f"⚠️ 讀取 restaurants.csv 失敗: {e}") |
|
|
| |
| df_prompts = pd.DataFrame() |
| PROMPT_COL_NAME = 'Visual_prompt' |
| try: |
| if not df.empty: |
| df_prompts = df.copy() |
| if 'Name' in df_prompts.columns: |
| df_prompts.set_index('Name', inplace=True) |
| except: pass |
|
|
| |
| 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: pass |
|
|
| |
| |
| |
|
|
| 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 target_moods: |
| pattern = "|".join(target_moods) |
| if 'Mood_Tags' in candidates.columns: |
| 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): |
| 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}\n日記：{user_diary}\n餐廳：{restaurant_name}\n資料：{restaurant_detail}\n請推薦這家餐廳並給予安慰或慶祝。" |
| 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, debug_mode): |
| if not score_input: |
| yield "⚠️ 請先選擇心情分數！", None, "" |
| return |
|
|
| restaurant, is_random = get_restaurant_data(score_input, food_input) |
| if restaurant is None: |
| yield "資料庫讀取錯誤", None, "" |
| return |
|
|
| name = str(restaurant['Name']) |
| url = str(restaurant.get('URL', f'https://www.google.com/maps/search/?api=1&query={urllib.parse.quote(name)}')) |
| 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" |
| if not df_prompts.empty and name in df_prompts.index: |
| try: |
| val = df_prompts.loc[name, PROMPT_COL_NAME] |
| if isinstance(val, pd.Series): val = val.iloc[0] |
| if pd.notna(val): img_prompt = str(val) |
| except: pass |
|
|
| ai_text = generate_content_with_groq(name, rag_info, diary_input, score_input) |
| debug_text = f"\n\nPrompt: `{img_prompt}`" if debug_mode else "" |
| final_response = f"### 🍽️ 推薦：{name}\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 |
|
|
| |
| |
| |
|
|
| emotion_text_obj = {'angry': '生氣', 'disgust': '噁心', 'fear': '害怕', 'happy': '開心', 'sad': '難過', 'surprise': '驚訝', 'neutral': '正常'} |
|
|
| def putText(img, x, y, text, size=50, color=(255, 255, 255)): |
| try: |
| fontpath = 'NotoSansTC-VariableFont_wght.ttf' |
| if not os.path.exists(fontpath): return img |
| font = ImageFont.truetype(fontpath, size) |
| imgPil = Image.fromarray(img) |
| draw = ImageDraw.Draw(imgPil) |
| displayText = emotion_text_obj.get(text, text) |
| draw.text((x, y), displayText, fill=color, font=font) |
| return np.array(imgPil) |
| except: return img |
|
|
| |
| def process_emotion(frame): |
| detected_emotion = "neutral" |
| mapped_score = "3 (普通)" |
| try: |
| analyze = DeepFace.analyze(frame, actions=['emotion'], enforce_detection=False) |
| if isinstance(analyze, list): analyze = analyze[0] |
| detected_emotion = analyze['dominant_emotion'] |
| |
| frame = putText(frame, 20, 40, detected_emotion) |
|
|
| if detected_emotion == 'happy': mapped_score = "5 (超棒)" |
| elif detected_emotion == 'surprise': mapped_score = "4 (不錯)" |
| elif detected_emotion == 'neutral': mapped_score = "3 (普通)" |
| elif detected_emotion in ['sad', 'fear']: mapped_score = "2 (不太好)" |
| elif detected_emotion in ['angry', 'disgust']: mapped_score = "1 (心情差)" |
| except: pass |
| return frame, mapped_score |
|
|
| |
| def manual_detect(frame): |
| if frame is None: return None, None |
| return process_emotion(frame) |
|
|
| |
| def toggle_recording_state(is_recording): |
| |
| new_state = not is_recording |
| |
| if new_state: |
| return new_state, "⏹️ 停止錄製 (即時偵測中)" |
| else: |
| return new_state, "🔴 開始錄製 (即時偵測)" |
|
|
| |
| def auto_stream_detect(frame, is_recording, last_time): |
| |
| if not is_recording or frame is None: |
| return gr.skip(), gr.skip(), last_time |
| |
| |
| current_time = time.time() |
| if current_time - last_time < 1.0: |
| return gr.skip(), gr.skip(), last_time |
| |
| |
| processed_frame, score = process_emotion(frame) |
| |
| |
| return processed_frame, score, current_time |
|
|
| |
| |
| |
|
|
| with gr.Blocks(title="AI 心情食堂") as demo: |
| |
| |
| is_recording = gr.State(False) |
| last_process_time = gr.State(0) |
|
|
| with gr.Column(visible=True) as main_app_col: |
| gr.Markdown(f"## 🍱 AI 心情食堂導航") |
| gr.Markdown("請看著鏡頭，讓 AI 幫你判斷今天的心情分數！") |
|
|
| with gr.Row(): |
| |
| with gr.Column(scale=1): |
| gr.Markdown("### 📸 步驟 1：心情偵測") |
| |
| with gr.Row(): |
| |
| webcam_input = gr.Image(sources=["webcam"], label="即時鏡頭", streaming=True) |
| |
| captured_image = gr.Image(label="偵測結果", interactive=False) |
| |
| with gr.Row(): |
| |
| record_btn = gr.Button("🔴 開始錄製 (即時偵測)", variant="stop") |
| |
| snapshot_btn = gr.Button("📸 單次截圖", variant="secondary") |
| |
| gr.Markdown("### 📝 步驟 2：確認與補充") |
| score_input = gr.Radio( |
| ["1 (心情差)", "2 (不太好)", "3 (普通)", "4 (不錯)", "5 (超棒)"], |
| label="1. 心情分數 (AI 自動填入)", |
| value="3 (普通)" |
| ) |
| food_input = gr.Radio(["吃飯", "吃麵", "隨便"], label="2. 想吃什麼", value="隨便") |
| diary_input = gr.Textbox(lines=3, label="3. 心情日記", placeholder="選填...") |
| |
| debug_mode_btn = gr.Checkbox(label="🔧 除錯資訊", 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="地圖導航") |
|
|
| |
|
|
| |
| record_btn.click( |
| fn=toggle_recording_state, |
| inputs=[is_recording], |
| outputs=[is_recording, record_btn] |
| ) |
|
|
| |
| webcam_input.stream( |
| fn=auto_stream_detect, |
| inputs=[webcam_input, is_recording, last_process_time], |
| outputs=[captured_image, score_input, last_process_time], |
| show_progress=False |
| ) |
|
|
| |
| snapshot_btn.click( |
| fn=manual_detect, |
| inputs=[webcam_input], |
| outputs=[captured_image, score_input] |
| ) |
|
|
| |
| submit_btn.click( |
| fn=mood_agent_logic, |
| inputs=[score_input, food_input, diary_input, debug_mode_btn], |
| outputs=[agent_output, image_output, map_output] |
| ) |
|
|
| if __name__ == "__main__": |
| demo.launch() |
