qa_vector_store.py

#%%
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