流程 :
注意重點 :
實作 :
# 直接載入 Chroma,使用既有的索引
import chromadb
from sentence_transformers import SentenceTransformer
# 初始化 embedding 模型 & chroma client
embedder = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
client = chromadb.PersistentClient(path="./chroma_db")
# 直接 get 已存在的 collection(faq_collection)
collection = client.get_collection("faq_collection")
# 定義 retrieve_chroma()
def retrieve_chroma(query: str, k: int = 3):
q_emb = embedder.encode([query], convert_to_numpy=True).tolist()
res = collection.query(
query_embeddings=q_emb,
n_results=k,
include=["documents", "metadatas", "distances"]
)
results = []
for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0]):
results.append({
"question": meta["question"],
"answer": meta["answer"],
"distance": dist
})
return results
# 定義 RAG 流程
def rag_answer(query: str, retriever="chroma", k=3):
docs = retrieve_chroma(query, k=k)
context_text = "\n".join([f"Q: {d['question']}\nA: {d['answer']}" for d in docs])
prompt = f"""你是一個客服助理,根據以下的 FAQ 回答使用者的問題:
---------------------
{context_text}
---------------------
使用者的問題:{query}
請給一個簡潔且有幫助的回答:
"""
answer = docs[0]["answer"] if docs else "抱歉,目前沒有相關資訊。"
return {
"query": query,
"retrieved": docs,
"prompt": prompt,
"answer": answer
}
# 測試
query = "如何退貨?"
result = rag_answer(query)
print("==== Answer ====")
print(result["answer"])
結果 :