<aside> 💡

本文是大型語言模型套件的學習筆記，主要參考柯克、陳葵懋、Ryan Chung，《LangChain奇幻旅程》。

</aside>

import os
from datetime import datetime
from typing import Dict, Any, Optional
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import (
    RunnableBranch, 
    RunnablePassthrough, 
    RunnableLambda,
    RunnableWithFallbacks
)

# 載入 .env 檔案中的環境變數
load_dotenv()

# 從環境變數中取得 OpenAI API 金鑰
openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
    raise ValueError("找不到 OPENAI_API_KEY 環境變數")

# 初始化 ChatOpenAI 物件
llm = ChatOpenAI(
    api_key=openai_api_key,
    model_name="gpt-3.5-turbo",
    temperature=0.7
)

# 定義文字解析器
parser = StrOutputParser()

# 定義問題分類的提示詞模板
classification_template = """
### 任務說明 ###
根據使用者的問題，判斷問題的類型。
可用類型有：
- weather: 與天氣相關，例如天氣如何、會不會下雨。
- travel: 與旅遊相關，例如景點、美食或最佳旅遊季節。
- both: 同時包含天氣和旅遊的問題。
- none: 無法判斷。

請僅輸出以下格式：
{{ "type": "<類型>" }}

### 使用者問題 ###
{question}
"""
classification_prompt = PromptTemplate(
    input_variables=["question"],
    template=classification_template
)

# 定義資訊提取提示詞模板
extraction_template = """
### 任務說明 ###
請從以下問題中提取城市名稱和月份資訊。

規則：
1. 城市名稱可能的形式：台北、臺北、台中、臺中、高雄等
2. 月份可能的形式：一月、1月、二月、2月等
3. 如果沒有提到月份，請回傳當前月份（1）

請使用以下格式回答（請只回答 JSON 格式，不要有其他文字）：
{{
    "city": "城市名稱",
    "month": 月份數字
}}

範例問題：「台北三月天氣如何？」
{{"city": "台北", "month": 3}}

範例問題：「高雄有什麼景點？」
{{"city": "高雄", "month": 1}}

### 使用者問題 ###
{question}
"""
extraction_prompt = PromptTemplate(
    input_variables=["question"],
    template=extraction_template
)

# 定義天氣提示詞模板
weather_template = """
### 任務說明 ###
請描述以下城市在 {month} 月的典型氣候特徵，包含溫度、降雨機率等資訊。
請直接描述，不要加入任何額外的開場白或結束語。

### 城市名稱 ###
{city}
"""
weather_prompt = PromptTemplate(
    input_variables=["city", "month"],
    template=weather_template
)

# 定義旅遊提示詞模板
travel_template = """
### 任務說明 ###
請為以下城市提供一份簡要的旅遊指南，包含必遊景點、特色美食和最佳旅遊季節。
請直接描述，不要加入任何額外的開場白或結束語。

### 城市名稱 ###
{city}
"""
travel_prompt = PromptTemplate(
    input_variables=["city"],
    template=travel_template
)

def format_both_response(responses: Dict[str, str]) -> str:
    return f"""天氣資訊：
{responses['weather']}

旅遊資訊：
{responses['travel']}"""

# 將字串轉換為字典的函式
def parse_json_response(text: str) -> Dict[str, Any]:
    try:
        return eval(text)
    except Exception as e:
        print(f"解析錯誤：{str(e)}")
        return {"type": "none"}

# 提取資訊的函式
def extract_info(text: str) -> Dict[str, Any]:
    try:
        info = eval(text)
        return {
            "city": info["city"],
            "month": info["month"]
        }
    except Exception as e:
        print(f"資訊提取錯誤：{str(e)}")
        return {"city": "未知", "month": datetime.now().month}

# 建立備用回應
def fallback_response(error: Exception) -> Dict[str, Any]:
    print(f"發生錯誤：{str(error)}")
    return {
        "type": "none",
        "error": str(error)
    }

def weather_fallback(error: Exception) -> str:
    return "抱歉，我無法提供準確的天氣資訊。建議您查看氣象局網站獲取最新天氣預報。"

def travel_fallback(error: Exception) -> str:
    return "抱歉，我目前無法提供旅遊建議。建議您查看當地旅遊網站或諮詢旅行社。"

"""
LangChain 的 Runnable 類型說明：

1. RunnableLambda：
   - 用途：將普通 Python 函數轉換為 Runnable 物件
   - 特點：可以串接在處理鏈中，保持一致的介面
   - 使用場景：資料轉換、格式化、驗證等

2. RunnableWithFallbacks：
   - 用途：為處理鏈添加錯誤處理機制
   - 工作原理：
     a. 首先嘗試執行主要的 runnable
     b. 如果失敗，按順序嘗試 fallbacks 列表中的備用方案
   - 使用場景：需要確保穩定性的關鍵處理環節

3. RunnableBranch：
   - 用途：根據條件選擇不同的處理分支
   - 工作原理：
     a. 按順序檢查每個條件
     b. 執行第一個條件為真的處理器
     c. 如果沒有條件匹配，執行預設處理器
   - 使用場景：需要根據輸入內容選擇不同處理邏輯時
"""

# 建立具有備用處理的鏈
# 1. 分類鏈：將用戶問題分類為天氣、旅遊或綜合查詢
base_classification_chain = classification_prompt | llm | parser | RunnableLambda(parse_json_response)
classification_chain = RunnableWithFallbacks(
    runnable=base_classification_chain,  # 主要處理鏈
    fallbacks=[RunnableLambda(fallback_response)]  # 當主鏈失敗時返回預設回應
)

# 2. 資訊提取鏈：從問題中提取城市和月份資訊
base_extraction_chain = extraction_prompt | llm | parser | RunnableLambda(extract_info)
extraction_chain = RunnableWithFallbacks(
    runnable=base_extraction_chain,  # 主要提取邏輯
    fallbacks=[RunnableLambda(lambda _: {"city": "未知", "month": datetime.now().month})]  # 提取失敗時使用預設值
)

# 3. 天氣查詢鏈：生成天氣相關回應
base_weather_chain = weather_prompt | llm | parser
weather_chain = RunnableWithFallbacks(
    runnable=base_weather_chain,  # 主要天氣查詢邏輯
    fallbacks=[RunnableLambda(weather_fallback)]  # 查詢失敗時返回通用天氣訊息
)

# 4. 旅遊查詢鏈：生成旅遊相關回應
base_travel_chain = travel_prompt | llm | parser
travel_chain = RunnableWithFallbacks(
    runnable=base_travel_chain,  # 主要旅遊查詢邏輯
    fallbacks=[RunnableLambda(travel_fallback)]  # 查詢失敗時返回通用旅遊訊息
)

# 5. 組合查詢鏈：處理同時需要天氣和旅遊資訊的查詢
both_chain = RunnableLambda(lambda inputs: {
    "weather": weather_chain.invoke(inputs),  # 獲取天氣資訊
    "travel": travel_chain.invoke(inputs)     # 獲取旅遊資訊
})

# 6. 格式化回應：將多個回應組合成易讀格式
format_response = RunnableLambda(format_both_response)

# 7. 主要處理分支：根據問題類型選擇適當的處理鏈
main_branch = RunnableBranch(
    # 條件判斷和對應的處理鏈
    (lambda x: x["type"] == "weather", weather_chain),     # 天氣查詢
    (lambda x: x["type"] == "travel", travel_chain),       # 旅遊查詢
    (lambda x: x["type"] == "both", RunnablePassthrough() | both_chain | format_response),  # 組合查詢
    lambda _: "抱歉，我無法理解您的問題。請試著詢問天氣或旅遊相關的問題。"  # 預設回應
)

def process_question(question: str) -> None:
    print("\\n問題：", question)
    print("-" * 50)
    
    try:
        # 判斷問題類型
        classification_result = classification_chain.invoke({"question": question})
        question_type = classification_result.get("type", "none")
        
        if "error" in classification_result:
            print("無法理解問題類型")
            print(classification_result["error"])
            return
            
        print(f"問題類型：{question_type}")
        
        # 提取資訊
        info = extraction_chain.invoke({"question": question})
        print(f"已提取資訊 - 城市：{info['city']}，月份：{info['month']}")
        
        # 使用 RunnableBranch 處理問題
        result = main_branch.invoke({
            "type": question_type,
            "city": info["city"],
            "month": info["month"]
        })
        
        print(result)
    except Exception as e:
        print(f"處理錯誤：{str(e)}")
    
    print("-" * 50)

# 測試問題列表
default_questions = [
    "台北在三月的天氣如何？",
    "高雄有什麼景點？",
    "台中有什麼景點，2月會不會下雨？",
    "我想吃冰淇淋",  # 測試無法理解的問題
    "xyz123",  # 測試亂碼輸入
    "",  # 測試空輸入
]

print("\\n=== 處理預設問題 ===\\n")
for question in default_questions:
    process_question(question)