import gradio as gr import google.generativeai as genai import os from dotenv import load_dotenv import json import pandas as pd import re import tempfile import textwrap from ddgs import DDGS load_dotenv() # 初始化：載入 .env 檔案中的變數 # --- 1. 設定 Google Gemini API --- api_key = os.getenv("GEMINI_API_KEY") if not api_key: print("❌ 警告：未偵測到 API Key，請在 Settings > Secrets 中設定 'GEMINI_API_KEY'。") else: genai.configure(api_key=api_key) # --- 外包資訊獲取函式 (使用 DuckDuckGo) --- def search_weather_info(location, date): """ 使用 DuckDuckGo 搜尋當地的天氣資訊，不需 API Key。 """ search_query = f"{location} {date} 天氣 氣溫 降雨機率 中央氣象署" print(f"🔍 正在搜尋：{search_query}") try: results = DDGS().text(search_query, max_results=3) # 將搜尋結果組合成一段文字 context_text = "" for res in results: context_text += f"- 標題：{res['title']}\n 內容：{res['body']}\n" return context_text except Exception as e: print(f"⚠️ 搜尋失敗: {e}") return "無法取得即時天氣資訊，請依據歷史氣候推估。" # --- 2. RAG 知識庫載入與檢索邏輯 --- RAG_FILE = "rag_knowledge.json" KNOWLEDGE_BASE = [] def load_knowledge_base(): """程式啟動時載入 JSON 資料集""" global KNOWLEDGE_BASE if os.path.exists(RAG_FILE): try: with open(RAG_FILE, "r", encoding="utf-8") as f: KNOWLEDGE_BASE = json.load(f) print(f"📚 RAG 知識庫載入成功！共 {len(KNOWLEDGE_BASE)} 筆片段。") except Exception as e: print(f"⚠️ 知識庫讀取失敗: {e}") else: print(f"⚠️ 找不到 {RAG_FILE}，將無法使用 RAG 功能。") # 執行載入 load_knowledge_base() def retrieve_context(activity): """ 根據活動類型，從知識庫中篩選相關的內容 這是一個輕量化的 Metadata Filtering RAG """ if not KNOWLEDGE_BASE: return "無知識庫資料" related_chunks = [] # 定義關鍵字對映 (Activity -> PDF Filename keywords) keywords = [] if activity == "登山健行": keywords = ["登山", "玉山"] # 會抓取「登山裝備」和「玉山檢查表」 elif activity == "自行車騎行": keywords = ["單車"] elif activity == "馬拉松": keywords = ["跑步"] # 篩選相關片段 for chunk in KNOWLEDGE_BASE: source_name = chunk.get("source", "") # 如果來源檔名包含對應的關鍵字，就納入 Context if any(k in source_name for k in keywords): # 格式化內容：[來源: 頁數] 內容... formatted_text = f"[出處: {os.path.basename(source_name)} P.{chunk['page']}]\n{chunk['content']}" related_chunks.append(formatted_text) # 將所有片段合併成一個字串 if related_chunks: print(f"🔍 RAG 檢索: 針對 '{activity}' 找到了 {len(related_chunks)} 個相關片段。") return "\n\n".join(related_chunks) else: print(f"⚠️ RAG 檢索: 找不到 '{activity}' 的相關資料。") return "無相關知識庫資料" # --- 3. 核心邏輯 --- def adapt_gear_advisor(activity, location, date, duration, feedback): print(f"--- 開始處理請求: {activity} @ {location} ---") # 先執行「外包」的搜尋工作 weather_context = search_weather_info(location, date) print(f"📄 獲取外部資訊長度: {len(weather_context)} 字") # 步驟 A: 執行 RAG 檢索 rag_context = retrieve_context(activity) # 步驟 B: 設定模型 (啟用 Google Search) model_name = 'models/gemini-2.5-flash' try: model = genai.GenerativeModel(model_name) except Exception as e: return f"❌ 模型初始化失敗: {str(e)}", None, None # 步驟 C: 組合 Prompt (注入 RAG Context) prompt = f""" 你是一個專業的戶外運動裝備顧問 'AdaptGear'。 【外部搜尋到的天氣資訊】 以下是剛剛從網路上搜尋到的真實資料： {weather_context} 【任務目標】 1. **天氣推估**：請根據 '{location}' 與 '{date}'，自行推估該季節的平均氣候（氣溫、降雨機率）。 2. **核心任務**：參考下方的【RAG 知識庫內容】，生成一份符合該活動規範的客製化裝備清單。 請優先使用知識庫中提到的專業裝備名稱（例如：若知識庫提到「GTX 外套」，就不要只寫「雨衣」）。 3. 結合使用者的【歷史回饋】進行反思與調整。 【RAG 知識庫內容 (請嚴格參考此資料建立清單)】 {rag_context} 【使用者輸入】 - 活動類型：{activity} - 地點/路線：{location} - 日期與時間：{date} - 行程時長：{duration} - 歷史回饋：{feedback if feedback else "無"} 【輸出格式要求】 只輸出一個標準 JSON 物件： {{ "weather_forecast": "真實天氣數據 (氣溫/降雨機率)", "smart_advice": "結合「RAG 知識庫規範」與「使用者回饋」的綜合建議", "checklist": [ {{ "category": "分類(如:個人裝備)", "item": "裝備名稱", "reason": "說明(例如: 依據玉山檢查表規定必帶)", "quantity": "數量" }}, ... ] }} """ try: print(f"正在呼叫模型 (含搜尋 & RAG)...") response = model.generate_content(prompt) print("✅ 生成成功") raw_text = response.text # 解析 JSON try: # 1. 尋找字串中第一個 '{' 和最後一個 '}' start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') + 1 if start_idx != -1 and end_idx != -1: # 只擷取大括號中間的內容 (過濾掉前面的 "Here is JSON" 或後面的雜訊) json_str = raw_text[start_idx:end_idx] data = json.loads(json_str) else: raise ValueError("無法在回應中找到 JSON 區塊 (找不到大括號)") except json.JSONDecodeError as e: # 如果還是失敗，印出原始文字以供除錯 print(f"❌ JSON 解析失敗。原始回應:\n{raw_text}") return f"解析錯誤：模型回傳了非 JSON 格式的內容。\n錯誤細節: {e}", None, None # 處理顯示文字 weather_info = data.get('weather_forecast', '無資料') advice_info = data.get('smart_advice', '無資料') display_text = ( f"## 🌤️ 天氣預測\n\n" f"{weather_info}\n\n" f"## 💡 AdaptGear 叮嚀\n\n" f"{advice_info}\n\n" ) # 處理表格與下載 checklist = data.get('checklist', []) df = pd.DataFrame(checklist) df.columns = ['類別', '裝備項目', '推薦理由', '數量'] with tempfile.NamedTemporaryFile(delete=False, suffix=".csv", mode='w', encoding='utf-8-sig') as tmp: df.to_csv(tmp.name, index=False) file_path = tmp.name return display_text, df, file_path except Exception as e: error_msg = str(e) print(f"❌ 生成失敗: {error_msg}") return f"發生錯誤：{error_msg}", None, None # --- 3. 建置 Gradio 介面 --- with gr.Blocks() as demo: gr.Markdown("# 🏔️ AdaptGear 運動裝備智慧顧問") with gr.Row(): with gr.Column(scale=1): input_activity = gr.Dropdown( label="活動類型", choices=[ "登山健行", "自行車騎行", "馬拉松", ], value="登山健行" ) input_location = gr.Textbox(label="地點/路線", value="嘉明湖", placeholder="例如：嘉明湖") input_date = gr.DateTime(label="出發日期", include_time=False, type="string") input_duration = gr.Textbox(label="行程時長", value="三天兩夜", placeholder="例如：三天兩夜") input_feedback = gr.Textbox(label="歷史回饋 (選填)", lines=3, placeholder="例如：上次覺得水帶不夠...") submit_btn = gr.Button("生成裝備清單", variant="primary") with gr.Column(scale=2): # 區塊 1: 智慧建議與天氣 (文字) output_advice = gr.Markdown(label="智慧分析") # 區塊 2: 裝備清單 (表格) - 看起來更有系統性 output_table = gr.Dataframe( headers=["類別", "裝備項目", "推薦理由", "數量"], label="客製化裝備清單", interactive=False ) # 區塊 3: 下載按鈕 output_file = gr.File(label="下載清單 (.csv)") submit_btn.click( fn=adapt_gear_advisor, inputs=[input_activity, input_location, input_date, input_duration, input_feedback], outputs=[output_advice, output_table, output_file] ) # --- 4. 啟動應用 --- if __name__ == "__main__": demo.launch()