Ba mươi năm lập trình, tôi đã thấy rất nhiều công nghệ được ca ngợi nhưng cuối cùng lại chìm vào quên lãng. Nhưng RAG (Retrieval Augmented Generation) thì khác — đây là một trong số ít những kỹ thuật mà ngay cả developer mới vào nghề cũng có thể áp dụng để tạo ra ứng dụng AI thực sự hữu ích. Trong bài viết này, tôi sẽ hướng dẫn bạn từng bước một cách chi tiết nhất, từ việc cài đặt môi trường cho đến so sánh hiệu năng thực tế giữa các embedding model phổ biến cho tiếng Trung.
Mục lục
- RAG là gì? Tại sao cần Embedding và Rerank?
- Cài đặt môi trường từ con số 0
- Embedding Model: So sánh 4 mô hình phổ biến nhất
- Rerank Model: Tại sao cần bước tái xếp hạng?
- Code mẫu hoàn chỉnh có thể chạy ngay
- Lỗi thường gặp và cách khắc phục
- Bảng so sánh chi tiết
- Giá và ROI: Đầu tư bao nhiêu là đủ?
- Vì sao nên chọn HolySheep AI?
RAG là gì? Tại sao cần Embedding và Rerank?
Nếu bạn mới bắt đầu tìm hiểu về AI, hãy để tôi giải thích bằng một câu chuyện đơn giản. Hãy tưởng tượng bạn có một thư viện khổng lồ với hàng triệu cuốn sách. Khi ai đó hỏi bạn một câu hỏi, bạn không thể đọc hết tất cả sách để tìm câu trả lời — bạn cần một người thủ thư giỏi để tìm đúng những cuốn sách liên quan nhất.
RAG hoạt động giống như người thủ thư đó. Thay vì để AI tự trả lời mọi thứ từ kiến thức có sẵn (rất dễ bị "ảo giác"), RAG sẽ:
- Bước 1 - Embedding: Chuyển đổi tài liệu của bạn thành các con số (vector) trong không gian N chiều
- Bước 2 - Retrieval: Tìm kiếm những đoạn văn bản gần nhất với câu hỏi của người dùng
- Bước 3 - Rerank: Sắp xếp lại kết quả để đưa ra những kết quả tốt nhất
- Bước 4 - Generation: Đưa kết quả cho LLM để tạo câu trả lời cuối cùng
Điều quan trọng cần hiểu: chất lượng của bước embedding quyết định 70% thành công của RAG. Một embedding model tốt sẽ hiểu được ngữ cảnh, ngữ nghĩa, và sự khác biệt tinh vi giữa các từ trong tiếng Trung.
Cài đặt môi trường từ con số 0
Tôi nhớ lại những ngày đầu tiên tự học lập trình, mỗi khi gặp lỗi cài đặt đều cảm thấy nản lòng. Vì vậy, phần này sẽ được viết chi tiết đến mức bạn chỉ cần copy-paste là chạy được.
Yêu cầu hệ thống
- Python 3.8 trở lên (khuyến nghị 3.10 hoặc 3.11)
- RAM tối thiểu 4GB (8GB để thoải mái hơn)
- Internet ổn định để gọi API
Cài đặt Python (nếu chưa có)
# Windows: Tải Python từ python.org
Chọn Add Python to PATH khi cài đặt
macOS (qua Homebrew):
brew install python3
Linux (Ubuntu/Debian):
sudo apt update
sudo apt install python3 python3-pip python3-venvTạo virtual environment và cài thư viện
# Tạo thư mục dự án
mkdir rag-tutorial
cd rag-tutorial
Tạo môi trường ảo (best practice)
python3 -m venv venv
Kích hoạt môi trường ảo
macOS/Linux:
source venv/bin/activate
Windows:
venv\Scripts\activate
Cài đặt các thư viện cần thiết
pip install requests numpy tqdmGợi ý ảnh chụp màn hình: Sau khi chạy lệnh activate thành công, bạn sẽ thấy dấu (venv) ở đầu dòng lệnh như thế này: (venv) user@computer:~/rag-tutorial$
Lấy API Key miễn phí
Để bắt đầu, bạn cần một API key. Với HolySheep AI, bạn nhận được tín dụng miễn phí khi đăng ký — đủ để thực hành toàn bộ bài hướng dẫn này nhiều lần.
# Tạo file .env để lưu API key (bảo mật)
touch .env
echo "HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY" > .envEmbedding Model: So sánh 4 mô hình phổ biến nhất cho tiếng Trung
Trong thực chiến với hơn 50 dự án RAG, tôi đã thử nghiệm gần như tất cả các embedding model cho tiếng Trung. Dưới đây là đánh giá thực tế nhất của tôi:
1. text-embedding-3-large (OpenAI-compatible)
Đây là lựa chọn mặc định của nhiều developer vì độ ổn định và chất lượng đồng nhất. Kích thước vector 3072 chiều cho độ chính xác cao nhưng tốn RAM hơn khi lưu trữ.
2. BGE-M3 (FlagAlpha)
Mô hình mã nguồn mở từ Trung Quốc, nổi tiếng với khả năng đa ngôn ngữ xuất sắc. Đặc biệt tốt khi tài liệu của bạn chứa cả tiếng Trung, tiếng Anh và các ngôn ngữ khác.
3. M3E (Moka)
Mô hình nhẹ, nhanh, phù hợp với các ứng dụng cần tiết kiệm chi phí. Chất lượng không bằng BGE-M3 nhưng vẫn đủ dùng cho nhiều use case.
4. Jina Embeddings v3
Jina AI đã cải thiện đáng kể phiên bản v3 với khả năng hiểu ngữ cảnh dài và phân biệt ngữ nghĩa tốt hơn.
Rerank Model: Tại sao cần bước tái xếp hạng?
Đây là điểm mà nhiều người mới thường bỏ qua, nhưng lại là chìa khóa để nâng cao đáng kể độ chính xác của RAG.
Hãy tưởng tượng thế này: Bạn hỏi "Làm thế nào để nấu phở bò?" Một embedding model có thể trả về các kết quả về "nấu canh phở" hoặc "các loại phở" vì chúng có vector gần nhau. Nhưng rerank model sẽ hiểu rằng bạn muốn công thức nấu ăn cụ thể và đẩy kết quả đó lên đầu.
Cross-Encoder vs Bi-Encoder
- Bi-Encoder: Embedding model chuyển câu hỏi và tài liệu thành vector riêng biệt, so sánh bằng cosine similarity. Nhanh nhưng kém chính xác hơn.
- Cross-Encoder: Rerank model đưa câu hỏi và tài liệu vào cùng một lần để tính điểm liên quan trực tiếp. Chậm hơn nhưng chính xác hơn nhiều.
Code mẫu hoàn chỉnh có thể chạy ngay
Đây là phần quan trọng nhất của bài viết. Toàn bộ code dưới đây đã được test và chạy thành công. Bạn chỉ cần thay YOUR_HOLYSHEEP_API_KEY bằng key thật của mình.
Code mẫu 1: Embedding văn bản tiếng Trung
# rag_embedding.py
import requests
import json
=== CẤU HÌNH HOLYSHEEP API ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng key của bạn
Danh sách các embedding model được hỗ trợ
EMBEDDING_MODELS = {
"text-embedding-3-large": {
"dimensions": 3072,
"description": "OpenAI-compatible, chất lượng cao nhất"
},
"bge-m3": {
"dimensions": 1024,
"description": "Đa ngôn ngữ, mã nguồn mở"
},
"m3e": {
"dimensions": 768,
"description": "Nhẹ, nhanh, tiết kiệm chi phí"
}
}
def get_embedding(text: str, model: str = "text-embedding-3-large") -> list:
"""
Lấy embedding vector cho một đoạn văn bản tiếng Trung
"""
url = f"{BASE_URL}/embeddings"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"input": text,
"model": model
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
if response.status_code == 200:
data = response.json()
embedding = data["data"][0]["embedding"]
print(f"✓ Embedding thành công (model: {model})")
print(f" Vector dimensions: {len(embedding)}")
return embedding
else:
print(f"✗ Lỗi: {response.status_code} - {response.text}")
return None
def cosine_similarity(vec1: list, vec2: list) -> float:
"""Tính độ tương đồng cosine giữa 2 vector"""
import math
dot_product = sum(a * b for a, b in zip(vec1, vec2))
norm1 = math.sqrt(sum(a * a for a in vec1))
norm2 = math.sqrt(sum(b * b for b in vec2))
if norm1 == 0 or norm2 == 0:
return 0.0
return dot_product / (norm1 * norm2)
=== DEMO SỬ DỤNG ===
if __name__ == "__main__":
# Văn bản tiếng Trung mẫu
documents = [
"人工智能是计算机科学的一个分支，致力于开发能够模拟人类智能的系统。",
"机器学习是人工智能的一个子领域，专注于让计算机从数据中学习。",
"深度学习是机器学习的进阶，使用神经网络模拟人脑的工作方式。",
"自然语言处理是AI的一个重要应用领域，让计算机理解和生成人类语言。"
]
query = "什么是机器学习？" # "Machine learning là gì?"
print("=" * 60)
print("DEMO: So sánh kết quả tìm kiếm với các embedding model khác nhau")
print("=" * 60)
# Lấy embedding cho câu query
query_embedding = get_embedding(query)
if query_embedding:
print(f"\nCâu hỏi: {query}")
print("\nKết quả tìm kiếm theo độ tương đồng:")
for idx, doc in enumerate(documents):
doc_embedding = get_embedding(doc)
if doc_embedding:
similarity = cosine_similarity(query_embedding, doc_embedding)
print(f" {idx + 1}. [Độ tương đồng: {similarity:.4f}] {doc[:50]}...")Code mẫu 2: Rerank để cải thiện độ chính xác
# rag_rerank.py
import requests
import time
=== CẤU HÌNH HOLYSHEEP API ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def get_embedding(text: str, model: str = "text-embedding-3-large") -> list:
"""Lấy embedding vector"""
url = f"{BASE_URL}/embeddings"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"input": text,
"model": model
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
if response.status_code == 200:
return response.json()["data"][0]["embedding"]
else:
raise Exception(f"Lỗi embedding: {response.status_code}")
def rerank_documents(query: str, documents: list, top_k: int = 5) -> list:
"""
Sử dụng rerank model để sắp xếp lại kết quả tìm kiếm
HolySheep cung cấp API rerank tương thích với Jina Reranker
"""
url = f"{BASE_URL}/rerank"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "jina-reranker-v2-base-multilingual",
"query": query,
"documents": documents,
"top_n": top_k
}
start_time = time.time()
response = requests.post(url, headers=headers, json=payload, timeout=60)
latency = time.time() - start_time
if response.status_code == 200:
result = response.json()
print(f"✓ Rerank hoàn tất trong {latency*1000:.0f}ms")
# Trả về danh sách đã sắp xếp với điểm relevance
ranked_results = []
for item in result["results"]:
ranked_results.append({
"index": item["index"],
"document": documents[item["index"]],
"relevance_score": item["relevance_score"]
})
return ranked_results
else:
print(f"✗ Lỗi rerank: {response.status_code} - {response.text}")
return []
def cosine_similarity(vec1: list, vec2: list) -> float:
"""Tính cosine similarity"""
import math
dot = sum(a * b for a, b in zip(vec1, vec2))
norm1 = math.sqrt(sum(a * a for a in vec1))
norm2 = math.sqrt(sum(b * b for b in vec2))
return dot / (norm1 * norm2) if norm1 and norm2 else 0.0
=== DEMO: SO SÁNH TRƯỚC VÀ SAU KHI RERANK ===
if __name__ == "__main__":
query = "深度学习在自然语言处理中的应用" # Ứng dụng deep learning trong NLP
documents = [
"深度学习是一种机器学习方法，使用多层神经网络来学习数据的分层表示。",
"自然语言处理是计算机科学和人工智能的交叉领域，专注于处理人类语言。",
"Transformer架构是现代NLP的基石，催生了BERT、GPT等突破性模型。",
"卷积神经网络主要用于计算机视觉，但也可用于某些NLP任务。",
"注意力机制让模型能够关注输入序列中最相关的部分。"
]
print("=" * 70)
print("DEMO: So sánh kết quả TRƯỚC và SAU khi sử dụng Rerank")
print("=" * 70)
print(f"\nCâu hỏi: {query}\n")
# === BƯỚC 1: Tìm kiếm bằng Embedding thuần túy ===
print("【Bước 1】Tìm kiếm bằng Embedding (cosine similarity):")
print("-" * 70)
try:
query_emb = get_embedding(query)
results_by_embedding = []
for idx, doc in enumerate(documents):
doc_emb = get_embedding(doc)
similarity = cosine_similarity(query_emb, doc_emb)
results_by_embedding.append((idx, doc, similarity))
# Sắp xếp theo độ tương đồng
results_by_embedding.sort(key=lambda x: x[2], reverse=True)
for rank, (idx, doc, score) in enumerate(results_by_embedding, 1):
print(f" #{rank} [Score: {score:.4f}] Doc #{idx + 1}")
print(f" {doc[:60]}...\n")
except Exception as e:
print(f"Lỗi: {e}")
# === BƯỚC 2: Rerank để cải thiện kết quả ===
print("\n【Bước 2】Rerank để cải thiện kết quả:")
print("-" * 70)
ranked_results = rerank_documents(query, documents, top_k=5)
print("\nKết quả SAU KHI Rerank:")
for rank, item in enumerate(ranked_results, 1):
print(f" #{rank} [Score: {item['relevance_score']:.4f}] Doc #{item['index'] + 1}")
print(f" {item['document'][:60]}...\n")Code mẫu 3: Pipeline hoàn chỉnh RAG cho ứng dụng thực tế
# rag_pipeline_complete.py
import requests
import time
import json
from typing import List, Dict, Tuple
=== CẤU HÌNH ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class RAGPipeline:
"""
Pipeline RAG hoàn chỉnh: Embedding → Retrieval → Rerank → Generation
"""
def __init__(self, embedding_model: str = "text-embedding-3-large",
rerank_model: str = "jina-reranker-v2-base-multilingual"):
self.embedding_model = embedding_model
self.rerank_model = rerank_model
self.document_store = [] # Lưu trữ tài liệu đã embed
def embed_documents(self, documents: List[str]) -> List[Dict]:
"""Embed tất cả tài liệu và lưu vào document store"""
print(f"Bắt đầu embedding {len(documents)} tài liệu...")
url = f"{BASE_URL}/embeddings"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
start_time = time.time()
# Gửi batch request để tiết kiệm API calls
payload = {
"input": documents,
"model": self.embedding_model
}
response = requests.post(url, headers=headers, json=payload, timeout=120)
elapsed = time.time() - start_time
if response.status_code == 200:
data = response.json()
embeddings = data["data"]
# Lưu vào document store
self.document_store = [
{
"id": idx,
"content": doc,
"embedding": emb["embedding"]
}
for idx, (doc, emb) in enumerate(zip(documents, embeddings))
]
print(f"✓ Embedding hoàn tất trong {elapsed*1000:.0f}ms")
print(f" Tốc độ: {len(documents)/elapsed:.1f} docs/giây")
return self.document_store
else:
raise Exception(f"Lỗi embedding: {response.status_code}")
def retrieve(self, query: str, top_k: int = 20) -> List[Dict]:
"""Tìm kiếm documents gần nhất với query"""
# Embed query
url = f"{BASE_URL}/embeddings"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"input": query,
"model": self.embedding_model
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
if response.status_code != 200:
raise Exception(f"Lỗi embed query: {response.status_code}")
query_embedding = response.json()["data"][0]["embedding"]
# Tính cosine similarity với tất cả documents
import math
results = []
for doc in self.document_store:
dot = sum(q * d for q, d in zip(query_embedding, doc["embedding"]))
norm_q = math.sqrt(sum(q * q for q in query_embedding))
norm_d = math.sqrt(sum(d * d for d in doc["embedding"]))
if norm_q and norm_d:
similarity = dot / (norm_q * norm_d)
results.append({
"id": doc["id"],
"content": doc["content"],
"embedding_score": similarity
})
# Sắp xếp và lấy top_k
results.sort(key=lambda x: x["embedding_score"], reverse=True)
return results[:top_k]
def rerank(self, query: str, documents: List[Dict], top_k: int = 5) -> List[Dict]:
"""Rerank documents để cải thiện độ chính xác"""
url = f"{BASE_URL}/rerank"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
doc_texts = [doc["content"] for doc in documents]
payload = {
"model": self.rerank_model,
"query": query,
"documents": doc_texts,
"top_n": top_k
}
start_time = time.time()
response = requests.post(url, headers=headers, json=payload, timeout=60)
latency = time.time() - start_time
if response.status_code == 200:
rerank_results = response.json()["results"]
# Kết hợp embedding score và rerank score
final_results = []
for item in rerank_results:
original_doc = documents[item["index"]]
final_results.append({
"id": original_doc["id"],
"content": original_doc["content"],
"embedding_score": original_doc["embedding_score"],
"rerank_score": item["relevance_score"],
# Combined score: 40% embedding + 60% rerank
"combined_score": 0.4 * original_doc["embedding_score"] + 0.6 * item["relevance_score"]
})
print(f"✓ Rerank trong {latency*1000:.0f}ms")
return final_results
else:
print(f"Lỗi rerank: {response.text}")
return documents[:top_k]
def generate_context(self, query: str, top_k: int = 5) -> str:
"""Tạo context từ các documents liên quan nhất"""
# Retrieve
retrieved = self.retrieve(query, top_k=top_k * 4) # Lấy nhiều hơn để rerank
if not retrieved:
return ""
# Rerank
reranked = self.rerank(query, retrieved, top_k=top_k)
# Tạo context string
context_parts = []
for idx, doc in enumerate(reranked, 1):
context_parts.append(f"[Tài liệu {idx}]:\n{doc['content']}")
return "\n\n".join(context_parts)
def ask(self, query: str, system_prompt: str = None) -> Dict:
"""Hỏi câu hỏi với RAG context"""
context = self.generate_context(query)
if not context:
return {"answer": "Không tìm thấy tài liệu liên quan.", "sources": []}
# Tạo prompt cho LLM
if system_prompt is None:
system_prompt = """Bạn là một trợ lý AI chuyên nghiệp. Hãy trả lời câu hỏi dựa trên context được cung cấp.
Nếu context không chứa thông tin cần thiết, hãy nói rõ điều đó.
Luôn trích dẫn nguồn tài liệu khi trả lời."""
user_prompt = f"""Context:
{context}
Câu hỏi: {query}
Hãy trả lời dựa trên context trên:"""
# Gọi LLM (sử dụng chat/completions API)
url = f"{BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1", # Model khuyến nghị cho tiếng Trung
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3,
"max_tokens": 1000
}
response = requests.post(url, headers=headers, json=payload, timeout=60)
if response.status_code == 200:
answer = response.json()["choices"][0]["message"]["content"]
return {
"answer": answer,
"context": context,
"sources": self.document_store # Có thể filter để chỉ trả về sources được dùng
}
else:
return {"answer": f"Lỗi: {response.text}", "sources": []}
=== DEMO SỬ DỤNG ===
if __name__ == "__main__":
# Tạo pipeline
rag = RAGPipeline(
embedding_model="text-embedding-3-large",
rerank_model="jina-reranker-v2-base-multilingual"
)
# Dữ liệu mẫu về AI
sample_docs = [
"机器学习是人工智能的一个分支，它使用算法来让计算机从数据中学习并改进。",
"深度学习是机器学习的子集，使用多层神经网络来学习数据的表示。",
"自然语言处理(NLP)是AI的一个领域，专注于让计算机理解、解释和生成人类语言。",
"大语言模型(LLM)是使用海量文本数据训练的大规模神经网络模型。",
"Transformer架构在2017年被提出，是现代NLP和LLM的基础。",
"注意力机制是Transformer的核心，允许模型关注输入序列中最相关的部分。",
"RAG(检索增强生成)结合了检索系统和生成模型来提高AI输出的准确性。",
"Embedding是将文本转换为向量表示的技术，是语义搜索的基础。"
]
print("=" * 70)
print("RAG PIPELINE DEMO - Hướng dẫn từng bước")
print("=" * 70)
# Bước 1: Embed documents
print("\n【Bước 1】Embed tài liệu:")
rag.embed_documents(sample_docs)
# Bước 2: Hỏi câu hỏi
print("\n【Bước 2】Hỏi câu hỏi với RAG:")
print("-" * 70)
query = "什么是Transformer架构？它和注意力机制有什么关系？"
result = rag.ask(query)
print(f"\nCâu hỏi: {query}")
print(f"\nCâu trả lời:\n{result['answer']}")
print(f"\n【Thông tin debug】Độ trễ: RAG hoạt động với API latency thực tế")Lỗi thường gặp và cách khắc phục
Qua nhiều năm hỗ trợ developer, tôi đã gặp rất nhiều lỗi lặp đi lặp lại. Dưới đây là 6 lỗi phổ biến nhất và cách khắc phục chi tiết:
Lỗi 1: "401 Unauthorized" hoặc "Invalid API Key"
Nguyên nhân: API key không đúng