app.py

import logging
logging.basicConfig(level=logging.DEBUG)
import gradio as gr
import pandas as pd
import numpy as np
import os
from openai import OpenAI
import pickle
import faiss
import json  # 用於解析 JSON

# Initialize OpenAI client
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# Load Excel data
df = pd.read_excel("product_list.xlsx")  # 修改為你的 xlsx 文件路徑

# Initialize FAISS Index
def initialize_embeddings_from_pkl(pkl_path: str, faiss_path: str):
    """從 PKL 檔案載入嵌入，並加載 FAISS 索引"""
    if not os.path.exists(pkl_path):
        raise FileNotFoundError(f"Embedding file not found at {pkl_path}")
    
    if not os.path.exists(faiss_path):
        raise FileNotFoundError(f"FAISS index file not found at {faiss_path}")
    
    print(f"Loading embeddings from {pkl_path}...")
    with open(pkl_path, "rb") as f:
        embeddings = pickle.load(f)
    
    print(f"Loading FAISS index from {faiss_path}...")
    index = faiss.read_index(faiss_path)
    
    print(f"FAISS index loaded with {index.ntotal} embeddings.")
    return index, embeddings

# Paths to FAISS index and PKL file
faiss_path = "product_index.faiss"  # 修改為你的 FAISS 索引文件路徑
pkl_path = "product_embeddings.pkl"  # 修改為你的嵌入文件路徑

# Initialize embeddings
print("Initializing embeddings...")
faiss_index, product_embeddings_array = initialize_embeddings_from_pkl(pkl_path, faiss_path)
assert faiss_index.ntotal == len(df), "FAISS 索引與 xlsx 文件的行數不一致！"
print(f"Loaded embeddings with shape: {len(product_embeddings_array)} x {len(product_embeddings_array[0])}")
print("Embeddings initialized")


# Generate embeddings for query
def get_embedding(text: str, model="text-embedding-ada-002"):
    """Get embeddings for a text using OpenAI's API"""
    try:
        text = text.replace("\n", " ")
        response = client.embeddings.create(
            input=[text],
            model=model
        )
        return response.data[0].embedding
    except Exception as e:
        print(f"Error getting embedding: {e}")
        return None


# Find similar products
def find_similar_products(query_embedding, top_k=8):
    """Find most similar products using FAISS index"""
    if faiss_index is None:
        raise ValueError("FAISS index is not loaded.")
    
    # FAISS expects float32 type embeddings
    query_embedding = np.array(query_embedding).astype('float32').reshape(1, -1)
    
    # Perform FAISS search
    distances, indices = faiss_index.search(query_embedding, top_k)
    
    # Retrieve metadata for matching products
    matching_products = df.iloc[indices[0]].copy()  # 從 df 中提取對應的行
    matching_products["similarity"] = distances[0]  # 加入相似度數據
    
    return matching_products

json_schema = {
    "name": "CookingIngredientsSchema",
    "description": "Extract cooking ingredients and analysis from user query.",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "analysis": {
                "type": "string",
                "description": "完整的需求分析，解釋用戶的目標和需要的物品。"
            },
            "ingredients": {
                "type": "array",
                "description": "提取的食材或關鍵物品清單。",
                "items": {
                    "type": "string",
                    "description": "單一食材或關鍵物品名稱。"
                }
            }
        },
        "required": ["analysis", "ingredients"],
        "additionalProperties": False
    }
}

# Analyze query and find products
def analyze_query_and_find_products(query: str) -> str:
    if not query.strip():
        return "請輸入您的問題或搜尋需求"
    
    try:
        # Analyze the query to understand intent
        analysis_messages = [
            {"role": "system", "content": f"""You are a knowledgeable shopping assistant. 
             When given a query:
             1. Analyze what the user is looking for
             2. Predict what user will need in a supermarket
             
             Provide your analysis in Traditional Chinese, focusing on understanding user needs."""},
            {"role": "user", "content": f"Analyze this query and explain what the user needs: {query}"}
        ]
        
        analysis_response = client.chat.completions.create(
            model="gpt-4o",
            messages=analysis_messages,
            temperature=0.7,
            response_format={
            "type": "json_schema",
            "json_schema": json_schema
            }
        )
        
        analysis = analysis_response.choices[0].message.content
        analysis_json = json.loads(analysis)

        description = analysis_json["analysis"]  # 取得分析結果
        ingredients = analysis_json["ingredients"]  # 取得提取的食材清單

        print("＝＝＝＝＝＝＝")
        print("關鍵字陣列：")
        print(ingredients)
        print("＝＝＝＝＝＝＝")
        # Generate embedding for the query

        # 為每個食材生成嵌入並進行查詢
        search_results = []
        for item in ingredients:
            print(f"正在搜尋：{item}")
            query_embedding = get_embedding(item)  # 生成食材嵌入
            matching_products = find_similar_products(query_embedding)  # 搜索結果
            search_results.append((item, matching_products))

            
        query_embedding = get_embedding(query + " " + description)
        
        # Find similar products
        matching_products = find_similar_products(query_embedding)
        print(f"Found {len(matching_products)} matching products")
        
        # Format the response
        response_parts = [
            "🔍 需求分析：",
            analysis,
            "\n📦 相關商品推薦：\n"
        ]
        
        for item, matching_products in search_results:
            response_parts.append(f"### {item} 的推薦商品：")
            for _, row in matching_products.iterrows():
                product_info = f"""
• {row['item_name']}
  ID: {row['item_id']}
  描述: {row['description']}
  分類: {row['tags']}
  相似度: {row['similarity']:.2f}"""
                response_parts.append(product_info)

        response_parts.append("\n💡 購物建議：")
        response_parts.append("根據您的需求，以上是推薦的商品！")

        return "\n".join(response_parts)

    except Exception as e:
        print(f"Error in search: {str(e)}")
        return f"搜尋發生錯誤: {str(e)}"

# Get system status
def get_system_status():
    """Get system initialization status"""
    embeddings_loaded = faiss_index is not None
    embedding_count = faiss_index.ntotal if embeddings_loaded else 0
    product_count = len(df)

    return {
        "embeddings_loaded": embeddings_loaded,
        "embedding_count": embedding_count,
        "product_count": product_count
    }


# Gradio Interface
with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown(
        """
        # 🛒 智慧商品推薦系統
        
        輸入您的問題或需求，系統會：
        1. 分析您的需求
        2. 推薦相關商品
        3. 提供實用建議
        """
    )
    
    # System status
    with gr.Row():
        status = get_system_status()
        status_md = f"""
        ### 系統狀態：
        - 資料庫商品數：{status['product_count']}
        - 向量嵌入狀態：{'✅ 已載入' if status['embeddings_loaded'] else '❌ 未載入'}
        """
        gr.Markdown(status_md)
    
    # Main interface
    with gr.Column():
        input_text = gr.Textbox(
            label="請輸入您的問題或需求",
            placeholder="例如：需要適合便當的食材",
            lines=3
        )
        output_text = gr.Textbox(
            label="分析結果與建議",
            lines=20
        )
        submit_btn = gr.Button("搜尋")
        
        submit_btn.click(
            fn=analyze_query_and_find_products,
            inputs=input_text,
            outputs=output_text
        )
    
    gr.Markdown("--- 系統使用 AI 分析需求並推薦商品。")

# Launch the app
demo.launch()