import os from dotenv import load_dotenv # 匯入 dotenv 以載入 .env 環境變數檔案 from typing import Optional from collections import defaultdict # 匯入 defaultdict，用於建立預設值的字典 from linebot import LineBotApi, WebhookHandler # 匯入 Line Bot SDK from linebot.exceptions import InvalidSignatureError # 匯入 Line 簽章無效的例外 from linebot.models import ( # 匯入 Line Bot 的各種訊息模型 MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, ImageMessage, FollowEvent ) import json from services.agents import run_agent load_dotenv() # --- 設定和初始化 --- # 設置 Line Bot 的 API 金鑰和秘密金鑰 (從環境變數讀取) # 初始化 Line Bot API 客戶端和 Webhook 處理器 try: line_bot_api = LineBotApi(os.environ["CHANNEL_ACCESS_TOKEN"]) line_handler = WebhookHandler(os.environ["CHANNEL_SECRET"]) print("Line Bot API 客戶端和 Handler 初始化成功。") except Exception as e: print(f"初始化 Line Bot 失敗: {e}") # 在實際應用中，這裡可能需要更嚴格的錯誤處理 line_bot_api = None line_handler = None # 使用 defaultdict 模擬用戶訊息歷史存儲 # 鍵(key)為 user_id，值(value)為一個儲存訊息的列表(list) user_message_history = defaultdict(list) def get_image_url_from_line(message_id): """ 從 Line 訊息 ID 獲取圖片內容並儲存到暫存檔案。 Args: message_id: Line 訊息的 ID。 Returns: 成功時回傳圖片儲存的本地路徑，失敗時回傳 None。 """ try: # 透過 Line Bot API 獲取訊息內容 message_content = line_bot_api.get_message_content(message_id) # 定義暫存檔案路徑 file_path = f"/tmp/{message_id}.png" # 將圖片內容以二進位寫入模式寫入檔案 with open(file_path, "wb") as f: for chunk in message_content.iter_content(): f.write(chunk) print(f"✅ 圖片成功儲存到：{file_path}") return file_path except Exception as e: print(f"❌ 圖片取得失敗：{e}") return None def store_user_message(user_id, message_type, message_content): """ 儲存用戶的訊息到 user_message_history 字典中。 Args: user_id: 用戶的 ID。 message_type: 訊息類型 (例如 "image" 或 "text")。 message_content: 訊息內容 (例如圖片路徑或文字)。 """ user_message_history[user_id].append( {"type": message_type, "content": message_content} ) def get_previous_message(user_id): """ 獲取用戶的上一則訊息。 Args: user_id: 用戶的 ID。 Returns: 如果歷史紀錄存在，回傳上一則訊息的字典；否則回傳預設的文字訊息。 """ if user_id in user_message_history and len(user_message_history[user_id]) > 0: # 回傳最後一則訊息 return user_message_history[user_id][-1] # 如果沒有歷史紀錄，回傳一個預設值 return {"type": "text", "content": "No message!"} # --- 事件處理器定義 --- # 1. 處理所有接收到的文字訊息事件 @line_handler.add(MessageEvent, message=(ImageMessage, TextMessage)) def handle_message(event): """ 處理接收到的文字訊息，並進行回覆。 """ # 獲取用戶 ID user_id = event.source.user_id # 情況一：處理圖片上傳 if event.message.type == "image": # 獲取 Line 傳來的圖片，並儲存到本地 image_path = get_image_url_from_line(event.message.id) if image_path: # 將圖片路徑儲存到用戶的訊息歷史中 store_user_message(user_id, "image", image_path) # 回覆用戶，告知圖片已收到，並請他輸入問題 line_bot_api.reply_message( event.reply_token, TextSendMessage(text="圖片已接收成功囉，幫我輸入你想詢問的問題喔~") ) else: line_bot_api.reply_message( event.reply_token, TextSendMessage(text="沒有接收到圖片~") ) elif event.message.type == "text": user_text = event.message.text # 獲取用戶傳來的文字 agent_input = "" # 獲取該用戶的「上一則」訊息 previous_message = get_previous_message(user_id) print(f"上一則訊息: {previous_message}") # 在後台印出除錯訊息 # 根據上一則訊息類型，動態組合給代理人的輸入 if previous_message["type"] == "image": # 如果上一則是圖片，代表用戶現在的文字是「針對圖片的提問」 image_path = previous_message["content"] agent_input = f"請根據這張圖片回答問題。圖片的路徑是 {image_path}，我的問題是：{user_text}" # 清除上一則圖片訊息，避免下一次文字訊息還被當作是圖片問答 user_message_history[user_id].pop() else: # 如果上一則不是圖片 (或沒有上一則)，代表這是一般的文字提問 (可能是要求生成圖片) agent_input = user_text out_str = "" try: # 運行 LangChain 代理人 response = run_agent(agent_input) # Agent 的輸出是 JSON 字串 (無論成功或失敗) out_str = response["output"] #print(f"out_str:{out_str}") tool_result = out_str.get("tool_result", {}) #print(f"tool_result:{tool_result}") final_response = out_str.get("final_response") #print(f"final_response:{final_response}") if "error" in tool_result: # 3A. 處理錯誤情況 (例如：圖片下載失敗、API 錯誤) error_msg = tool_result["error"] reply_text = f"🚫 圖片工具執行失敗：\n{error_msg}" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_text)) elif "image_url" in tool_result: # 3B. 處理成功情況 (帶有圖片 URL，例如：圖片生成或去模糊工具) image_url = tool_result["image_url"] text_result = tool_result.get("text_result", "圖片處理完成。") # Line 要求圖片 URL 必須是 HTTPS # 由於 HF_SPACE 預設是 https，這裡做一個保險轉換 if image_url.startswith("http://"): image_url = image_url.replace("http://", "https://") # 使用 reply_message (或 push_message) 同時推送文字和圖片 # 註：reply_message 只能回覆一次。若要一次發送多個訊息，需要使用 list # 這裡假設 text_result 是主要回覆，圖片作為輔助。 line_bot_api.reply_message( event.reply_token, [ TextSendMessage(text=f"✨ {text_result}"), ImageSendMessage(original_content_url=image_url, preview_image_url=image_url) ] ) elif "text_result" in tool_result: # 3C. 處理純文字結果 (例如：多模態分析工具) text_result = tool_result["text_result"] line_bot_api.reply_message(event.reply_token, TextSendMessage(text=text_result)) elif final_response: line_bot_api.reply_message(event.reply_token, TextSendMessage(text=final_response)) else: # 3D. 處理意外的 JSON 結構 line_bot_api.reply_message( event.reply_token, TextSendMessage(text="代理人回覆格式無法識別，請聯繫管理員。") ) except json.JSONDecodeError: # 4. 處理 Agent 返回了無法解析的純文字 (非 JSON) # 這通常發生在 Agent 決定不使用工具，直接回覆純文字，或者推理過程出錯。 line_bot_api.reply_message(event.reply_token, TextSendMessage(text=out_str)) except Exception as e: # 處理代理人執行時的錯誤 print(f"代理人執行出錯: {e}") out = f"代理人執行出錯!錯誤訊息：{e}" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=out)) # 2. 處理關注 (Follow) 事件 @line_handler.add(FollowEvent) def handle_follow(event: FollowEvent): """ 處理 Line Bot 被用戶關注的事件。 """ if event.source.user_id: user_id = event.source.user_id print(f"用戶 {user_id} 關注了 Bot。") welcome_message = "感謝您的關注！我是您的圖像助理，可以幫您生成圖片或對圖片進行去模糊處理。請直接輸入您的請求。" line_bot_api.reply_message( event.reply_token, TextSendMessage(text=welcome_message) ) # --- 核心 Webhook 處理函數 --- # 這個函數將在 app.py 的 BackgroundTasks 中被調用 def line_handle(body: str, signature: Optional[str]): """ Line Bot Webhook 的主要處理入口。 Args: body: Webhook 請求的原始 body 內容 (字串)。 signature: 來自 Line 伺服器的 X-Line-Signature 標頭值。 """ if not line_handler: print("錯誤：Line Bot Handler 未成功初始化。") return try: # line_handler 會自動驗證簽章，並根據 body 內容分派事件到上面定義的處理器 line_handler.handle(body, signature) except InvalidSignatureError: # 如果簽章無效，會在這裡捕獲 (雖然 app.py 已經有捕獲，但這裡也做一次防禦性處理) print("Invalid signature. 無效的簽章，可能請求不是來自 Line 官方伺服器。") # 這裡不應 raise，因為它是在背景任務中 except Exception as e: print(f"處理 Webhook 時發生未知錯誤: {e}") print("Webhook 事件處理流程完成。")