人工客服末日？幫你的 WordPress 裝上 AI 智慧客服

import os
import json
import time
import openai
from pinecone import Pinecone, ServerlessSpec
from datetime import datetime
from dotenv import load_dotenv

class PineconeUploader:
    def __init__(self):
        """初始化設定"""
        # 載入 .env 檔案
        load_dotenv()
        
        # 從環境變數中取得 API 金鑰和環境
        OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
        PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
        PINECONE_ENVIRONMENT = os.getenv("PINECONE_ENVIRONMENT")  # 例如 'us-west1-gcp'
        
        # 檢查是否成功取得所有必要的變數
        if not all([OPENAI_API_KEY, PINECONE_API_KEY, PINECONE_ENVIRONMENT]):
            raise ValueError("請確保已設定 OPENAI_API_KEY、PINECONE_API_KEY 和 PINECONE_ENVIRONMENT")
        
        # 設定 OpenAI API 金鑰
        openai.api_key = OPENAI_API_KEY
        
        # 創建 Pinecone 客戶端實例
        self.pc = Pinecone(
            api_key=PINECONE_API_KEY,
            environment=PINECONE_ENVIRONMENT
        )
        
        # 索引名稱
        index_name = "your-index"  # 請替換為您的索引名稱
        self.namespace = "your-namespace  # 定義命名空間
        self.batch_size = 50
        
        # 檢查索引是否存在，若不存在則創建
        existing_indexes = [index.name for index in self.pc.list_indexes()]
        if index_name not in existing_indexes:
            self.create_index(index_name)
        else:
            # 獲取索引資訊
            index_description = self.pc.describe_index(index_name)
            index_dimension = index_description.dimension
            if index_dimension != 1536:
                print(f"索引 '{index_name}' 的維度為 {index_dimension}，與嵌入向量維度 1536 不匹配，將刪除並重新創建索引。")
                # 刪除索引
                self.pc.delete_index(index_name)
                time.sleep(5)  # 等待索引刪除
                # 創建新索引
                self.create_index(index_name)
        
        # 連接到索引
        self.index = self.pc.Index(index_name)
    
    def create_index(self, index_name):
        """創建新的 Pinecone 索引"""
        self.pc.create_index(
            name=index_name,
            dimension=1536,  # 'text-embedding-ada-002' 的維度為 1536
            metric='cosine',
            spec=ServerlessSpec(
                cloud='aws',  # 或 'gcp'，根據您的環境
                region=os.getenv("PINECONE_ENVIRONMENT")  # 使用您的環境變數
            )
        )
        print(f"已創建索引 '{index_name}'，維度為 1536。")
    
    def prepare_text(self, record):
        """將記錄轉換為結構化文本"""
        return f"""
建案名稱: {record.get('project_name', '')}
單價: {record.get('price_per_ping', '')}
總價: {record.get('total_price', '')}
車位價格: {record.get('parking_price', '')}
地址: {record.get('address', '')}
格局坪數: {record.get('layout_ping', '')}
完工時間: {record.get('delivery_date', '')}
付款方式: {record.get('payment_method', '')}
車位規劃: {record.get('parking_plan', '')}
建材說明: {record.get('building_materials', '')}
公共設施: {record.get('public_facilities', '')}
棟戶規劃: {record.get('building_plan', '')}
樓層規劃: {record.get('floor_plan', '')}
座向規畫: {record.get('orientation', '')}
結構工程: {record.get('structure_type', '')}
用途規劃: {record.get('usage_plan', '')}
土地分區: {record.get('land_zone', '')}
管理費用: {record.get('management_fee', '')}
公設比: {record.get('public_area_ratio', '')}
建蔽率: {record.get('building_coverage_ratio', '')}
基地面積: {record.get('land_area', '')}
開發商: {record.get('developer', '')}
建築師: {record.get('architect', '')}
營造公司: {record.get('construction_company', '')}
銷售公司: {record.get('sales_company', '')}
        """.strip()
    
    def get_embedding(self, text):
        """使用 OpenAI API 生成文本的嵌入向量"""
        try:
            response = openai.Embedding.create(
                input=text,
                model='text-embedding-ada-002'  # 使用最新的模型
            )
            return response['data'][0]['embedding']
        except Exception as e:
            print(f"生成嵌入向量時發生錯誤: {str(e)}")
            return None
    
    def process_and_upload(self, json_file):
        """處理 JSON 檔案並上傳到 Pinecone"""
        try:
            with open(json_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            print(f"開始處理 {len(data)} 筆資料")
            
            vectors = []
            for i, record in enumerate(data):
                text = self.prepare_text(record)
                embedding = self.get_embedding(text)
                
                if embedding is None:
                    continue
                
                vector = {
                    'id': f"property_{i}",
                    'values': embedding,
                    'metadata': {
                        'project_name': record.get('project_name', ''),
                        'price_per_ping': record.get('price_per_ping', ''),
                        'total_price': record.get('total_price', ''),
                        'address': record.get('address', ''),
                        'property_type': record.get('category', ''),
                        'status': record.get('house_status', ''),
                        'text': text
                    }
                }
                vectors.append(vector)
                
                if len(vectors) >= self.batch_size:
                    # 使用指定的 namespace 上傳
                    self.index.upsert(
                        vectors=vectors,
                        namespace=self.namespace
                    )
                    print(f"已上傳 {i + 1} 筆資料")
                    vectors = []
                    time.sleep(1)
            
            if vectors:
                self.index.upsert(
                    vectors=vectors,
                    namespace=self.namespace
                )
                print(f"已上傳剩餘 {len(vectors)} 筆資料")
            
            # 測試查詢
            self.test_query()
            
            print("資料上傳完成！")
            
        except Exception as e:
            print(f"處理資料時發生錯誤: {str(e)}")
    
    def test_query(self):
        """測試查詢功能"""
        try:
            # 使用示例查詢文本
            query_text = "三房兩廳的新建案"
            query_embedding = self.get_embedding(query_text)
            
            if query_embedding is None:
                print("無法生成查詢嵌入向量")
                return
            
            response = self.index.query(
                namespace=self.namespace,
                vector=query_embedding,
                top_k=2,
                include_values=False,
                include_metadata=True
            )
            print("\n測試查詢結果:")
            for match in response.matches:
                print(f"ID: {match.id}")
                print(f"Score: {match.score}")
                print(f"Metadata: {match.metadata}")
                print("-----")
        except Exception as e:
            print(f"測試查詢時發生錯誤: {str(e)}")

def main():
    try:
        uploader = PineconeUploader()
        uploader.process_and_upload('房地產資料.json')
    except Exception as e:
        print(f"程式執行失敗: {str(e)}")

if __name__ == "__main__":
    main()