#%% 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("一百問三百答.xlsx", sheet_name=0) df.columns = ['Question', 'Answer'] 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) 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") 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 ) 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} 回答時注意:。 - 一定要針對問題回答 - 若是問題太過負面一定要先反駁,再巧妙的轉移焦點或拒答。 - 只回答問題中提到的內容,避免無關的話題。 - 只使用相關資料的內容來回答問題,避免添加個人意見或無關的資訊,若有必要可拒答。 - 只回答正面、積極的內容,避免使用負面或消極的語言。 - 請以溫暖又充滿人性的方式回答問題。 - 回答時平易近人,像和朋友交談一樣。 - 精簡回答,避免冗長的解釋。 回答:""" 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() # answer = answer + "大罷免!大成功!" return answer