Upload 2 files

qa_vector_store.py

@@ -0,0 +1,161 @@
+#%%
+from text2vec import SentenceModel
+from qdrant_client import QdrantClient
+from qdrant_client.models import VectorParams, Distance, PointStruct
+
+
+def deterministic_id(text):
+    import hashlib
+    return int(hashlib.sha256(text.encode('utf-8')).hexdigest(), 16) >> 128
+
+def build_qa_vector_store(model_name, collection_name):
+    import pandas as pd
+    # 讀取資料
+    df = pd.read_excel("data/一百問三百答.xlsx", sheet_name=0)
+    df.columns = ['Question', 'Answer']
+    original_len = len(df)
+
+    # 去除重複 QA 組合
+    df = df.drop_duplicates(subset=["Question", "Answer"]).reset_index(drop=True)
+    print(f"📊 原始資料筆數：{original_len}，去除重複後筆數：{len(df)}")
+
+    questions = df['Question'].tolist()
+    answers = df['Answer'].tolist()
+    # 初始化模型
+    model = SentenceModel(model_name)
+    question_vectors = model.encode(questions, normalize_embeddings=True)
+    embedding_dim = len(question_vectors[0])
+
+    # 初始化 Qdrant
+    client = QdrantClient(path="./qadrant_data")
+
+    # 建立新的 collection（重新指定向量維度）
+    client.recreate_collection(
+        collection_name=collection_name,
+        vectors_config=VectorParams(size=embedding_dim, distance=Distance.COSINE)
+    )
+
+    points = [
+        PointStruct(
+            id=deterministic_id(q + a),
+            vector=vector.tolist(),
+            payload={"question": q, "answer": a}
+        )
+        for q, a, vector in zip(questions, answers, question_vectors)
+    ]
+
+    client.upsert(collection_name=collection_name, points=points)
+    print(f"✅ 向量資料庫建立完成，共嵌入 {len(points)} 筆 QA。")
+    client.scroll(collection_name=collection_name, limit=100)
+
+
+# build_qa_vector_store(model_name, collection_name)
+
+# %%
+# model_name = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
+# collection_name = model_name.split("/")[-1]  
+# client = QdrantClient(path="./qadrant_data")
+# count = client.count(collection_name=collection_name, exact=True).count
+# print(f"📦 Collection {collection_name} 中有 {count} 筆資料")
+# # %%
+# from collections import Counter
+
+# records = client.scroll(collection_name=collection_name, limit=1000)[0]
+# answers = [rec.payload["answer"] for rec in records]
+# duplicates = [item for item, count in Counter(answers).items() if count > 1]
+# print("重複答案數量：", len(duplicates))
+# print("部分重複答案：", duplicates[:5])
+
+def retrieve_and_rerank(query, model_name, collection_name, cross_encoder_model, score_threshold=0.6, search_top_k=25, rerank_top_k=5):
+    from semantic_reranker import rerank_results
+    model = SentenceModel(model_name)
+    client = QdrantClient(path="./qadrant_data")
+
+    # 確認 collection 是否存在
+    if collection_name not in [c.name for c in client.get_collections().collections]:
+        print(f"⚠️ Collection {collection_name} 不存在，請先建立向量資料庫。")
+        return
+        
+    query_vector = model.encode(query, normalize_embeddings=True)
+    results = client.search(
+        collection_name=collection_name,
+        query_vector=query_vector.tolist(),
+        limit=search_top_k,
+        score_threshold=score_threshold 
+    )
+
+    retrieved_answers = [r.payload["answer"] for r in results]
+
+    # 用 cross-encoder rerank
+    reranked = rerank_results(
+        cross_encoder_model=cross_encoder_model,
+        query=query,
+        candidate_passages=retrieved_answers,
+        top_k=rerank_top_k
+    )   
+
+    if not results:
+        print("❌ 找不到相關答案。")
+    else:
+        for i, hit in enumerate(results):
+            print(f"✅ Top {i+1}：{hit.payload['answer']} (score={hit.score:.3f})")
+
+    for i, (answer, score) in enumerate(reranked):
+        print(f"🔥 Rerank Top {i+1}：{answer} (score={score:.3f})")
+    return reranked
+
+
+def generate_response_from_local_llm(query, passages, tokenizer, model, max_new_tokens=256):
+    context = "\n".join([f"- {p}" for p in passages])
+    if not passages:
+        return "抱歉，沒有找到相關資料。"
+    prompt = f"""你是一個客服助理。請根據以下資料回答使用者問題。
+問題：{query}
+相關資料：
+{context}
+回答時注意：。
+- 一定要針對問題回答
+- 若是問題太過負面一定要先反駁，再巧妙的轉移焦點或拒答。
+- 只回答問題中提到的內容，避免無關的話題。
+- 只使用相關資料的內容來回答問題，避免添加個人意見或無關的資訊，若有必要可拒答。
+- 只回答正面、積極的內容，避免使用負面或消極的語言。
+- 請以溫暖又充滿人性的方式回答問題。
+- 回答時平易近人，像和朋友交談一樣。
+- 精簡回答，避免冗長的解釋。
+回答："""
+    print(prompt)
+    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
+    outputs = model.generate(
+        **inputs,
+        max_new_tokens=max_new_tokens,
+        do_sample=True,
+        top_p=0.95,
+        temperature=0.7
+    )
+    decoded_output = tokenizer.decode(outputs[0], skip_special_tokens=True)    
+
+    # 提取回答部分    
+    answer = decoded_output.split("回答：", 1)[-1].strip() if "回答：" in decoded_output else decoded_output
+    return answer
+#%%
+# from sentence_transformers import CrossEncoder
+# from transformers import AutoTokenizer, AutoModelForCausalLM
+# model_name = "sentence-transformers/paraphrase-multilingual-mpnet-base-v2"
+# collection_name = model_name.split("/")[-1]  
+# cross_encoder_model = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1")
+# tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-0.5B", trust_remote_code=True)
+# model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2-0.5B", trust_remote_code=True)
+# # tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen1.5-7B-Chat", trust_remote_code=True)
+# # model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen1.5-7B-Chat", trust_remote_code=True)
+
+# #%%
+# user_query = "許智傑做過什麼壞事"
+# reranked = retrieve_and_rerank(user_query, model_name, collection_name, cross_encoder_model, score_threshold=0.6, search_top_k=20, rerank_top_k=5)
+# #%%
+# passages = [answer for answer, score in reranked]
+# answer = generate_response_from_local_llm(user_query, passages, tokenizer, model, max_new_tokens=256)
+# print("回答：", answer)
+
+
+
+# %%

semantic_reranker.py

@@ -0,0 +1,8 @@
+def rerank_results(cross_encoder_model, query, candidate_passages, top_k=10):
+    if not candidate_passages: 
+        return []
+    pairs = [(query, passage) for passage in candidate_passages]
+    scores = cross_encoder_model.predict(pairs)
+    reranked = sorted(zip(candidate_passages, scores), key=lambda x: x[1], reverse=True)
+
+    return reranked[:top_k]