Trong bài viết này, mình sẽ chia sẻ kinh nghiệm thực chiến khi triển khai Retrieval-Augmented Generation (RAG) cho các dự án AI Agent, kèm theo case study từ một nền tảng TMĐT ở TP.HCM đã tối ưu chi phí API từ $4,200 xuống còn $680 mỗi tháng.
Bối cảnh thực tế: Từ điểm đau đến giải pháp
Một nền tảng thương mại điện tử tại TP.HCM với khoảng 2 triệu sản phẩm gặp vấn đề nghiêm trọng: chatbot chăm sóc khách hàng thường xuyên đưa ra thông tin sai lệch về tồn kho, chương trình khuyến mãi, và chính sách đổi trả. Đội ngũ kỹ thuật đã thử nghiệm nhiều phương pháp như fine-tuning, prompt engineering nhưng không hiệu quả.
Điểm nghẽn cốt lõi: Model foundation chỉ có kiến thức đến cutoff date, không thể truy cập dữ liệu real-time về inventory, pricing, và policy của doanh nghiệp.
Sau khi đánh giá, đội ngũ quyết định triển khai RAG architecture với HolySheep AI làm inference layer. Kết quả sau 30 ngày: độ trễ trung bình giảm từ 420ms xuống 180ms, chi phí API giảm 83%.
RAG Architecture Tổng quan
Trước khi đi vào code, chúng ta cần hiểu rõ luồng dữ liệu trong hệ thống RAG:
┌─────────────────────────────────────────────────────────────────┐
│ RAG Pipeline Architecture │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 1. INDEXING 2. RETRIEVAL │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Document │ │ Query Input │ │
│ │ Sources │──→ Chunks ──→ │ + Vector DB │ │
│ │ (PDF, DB...) │ │ Search │ │
│ └──────────────┘ └──────┬───────┘ │
│ │ │
│ 3. GENERATION ▼ │
│ ┌──────────────────────────────────────────────┐ │
│ │ LLM (with Context Window) │ │
│ │ System Prompt + Retrieved Chunks + Query │ │
│ └──────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Generated Reply │ │
│ └──────────────────┘ │
└─────────────────────────────────────────────────────────────────┘Triển khai Chi tiết với HolySheep AI
Bước 1: Cài đặt Dependencies và Cấu hình
# requirements.txt
openai==1.12.0
faiss-cpu==1.7.4
langchain==0.1.4
langchain-community==0.0.17
pypdf==4.0.1
numpy==1.26.3
python-dotenv==1.0.0# config.py
import os
from dotenv import load_dotenv
load_dotenv()
✅ Sử dụng HolySheep AI - KHÔNG dùng api.openai.com
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Model configuration với giá 2026
DeepSeek V3.2: $0.42/MTok - rẻ nhất cho embedding + generation
Gemini 2.5 Flash: $2.50/MTok - balance giữa speed và cost
EMBEDDING_MODEL = "deepseek-embed-v3"
GENERATION_MODEL = "gemini-2.5-flash"
RAG Configuration
CHUNK_SIZE = 512
CHUNK_OVERLAP = 64
TOP_K_RESULTS = 5
SIMILARITY_THRESHOLD = 0.7Bước 2: Document Processing và Embedding
# document_processor.py
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import OpenAIEmbeddings
import numpy as np
from config import HOLYSHEEP_BASE_URL, HOLYSHEEP_API_KEY, EMBEDDING_MODEL, CHUNK_SIZE, CHUNK_OVERLAP
class DocumentProcessor:
def __init__(self):
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
length_function=len,
separators=["\n\n", "\n", " ", ""]
)
def load_pdf(self, file_path: str) -> list:
"""Load và split PDF document thành chunks"""
loader = PyPDFLoader(file_path)
documents = loader.load()
# Split thành chunks nhỏ để retrieve hiệu quả
chunks = self.text_splitter.split_documents(documents)
return chunks
def load_from_database(self, connection, query: str) -> list:
"""Load documents từ SQL database"""
cursor = connection.cursor()
cursor.execute(query)
rows = cursor.fetchall()
documents = []
for row in rows:
from langchain.schema import Document
doc = Document(
page_content=row[0], # content column
metadata={
"source": "database",
"id": row[1], # id column
"created_at": row[2]
}
)
documents.append(doc)
return documents
✅ Custom Embedding class cho HolySheep
class HolySheepEmbeddings:
"""Custom embeddings sử dụng HolySheep API thay vì OpenAI"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
def embed_documents(self, texts: list) -> list:
"""Embed danh sách documents - batch processing"""
import requests
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": EMBEDDING_MODEL,
"input": texts
}
)
response.raise_for_status()
data = response.json()
return [item["embedding"] for item in data["data"]]
def embed_query(self, text: str) -> list:
"""Embed single query - cần thiết cho retrieval step"""
return self.embed_documents([text])[0]
Khởi tạo embedding model
embeddings = HolySheepEmbeddings(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL
)Bước 3: Vector Store và Retrieval
# vector_store.py
import faiss
import numpy as np
from typing import List, Tuple
from langchain.schema import Document
from document_processor import embeddings, DocumentProcessor
class VectorStoreManager:
"""Quản lý FAISS index với HolySheep embeddings"""
def __init__(self, dimension: int = 1536):
self.dimension = dimension
self.index = faiss.IndexFlatIP(dimension) # Inner Product for normalized vectors
self.documents = []
self.embeddings_model = embeddings
def add_documents(self, chunks: List[Document]):
"""Thêm documents vào vector store"""
texts = [chunk.page_content for chunk in chunks]
metadatas = [chunk.metadata for chunk in chunks]
# Batch embedding - giảm API calls
vectors = self.embeddings_model.embed_documents(texts)
# Normalize vectors cho cosine similarity
vectors = np.array(vectors).astype('float32')
norms = np.linalg.norm(vectors, axis=1, keepdims=True)
vectors = vectors / norms
# Add to FAISS index
self.index.add(vectors)
self.documents.extend(chunks)
print(f"✅ Đã thêm {len(chunks)} documents vào vector store")
print(f" Tổng documents: {self.index.ntotal}")
def similarity_search(
self,
query: str,
top_k: int = 5,
threshold: float = 0.7
) -> List[Tuple[Document, float]]:
"""
Semantic search với similarity threshold
Returns:
List of (Document, similarity_score) tuples
"""
# Embed query
query_vector = self.embeddings_model.embed_query(query)
query_vector = np.array([query_vector]).astype('float32')
# Normalize query vector
query_vector = query_vector / np.linalg.norm(query_vector)
# Search k*2 để filter theo threshold
distances, indices = self.index.search(query_vector, top_k * 2)
results = []
for dist, idx in zip(distances[0], indices[0]):
if idx == -1:
continue
similarity = (dist + 1) / 2 # Convert [-1,1] to [0,1]
if similarity >= threshold:
results.append((self.documents[idx], similarity))
if len(results) >= top_k:
break
return results
def save_index(self, path: str = "vector_index.faiss"):
"""Lưu FAISS index ra disk"""
import pickle
faiss.write_index(self.index, path)
with open("documents.pkl", "wb") as f:
pickle.dump(self.documents, f)
print(f"💾 Đã lưu index tại {path}")
def load_index(self, path: str = "vector_index.faiss"):
"""Load FAISS index từ disk"""
import pickle
self.index = faiss.read_index(path)
with open("documents.pkl", "rb") as f:
self.documents = pickle.load(f)
print(f"📂 Đã load {self.index.ntotal} documents")
Khởi tạo vector store
vector_store = VectorStoreManager(dimension=1536)Bước 4: RAG Agent với HolySheep Generation
# rag_agent.py
import requests
import json
from datetime import datetime
from typing import List, Dict, Optional
from vector_store import vector_store
from config import (
HOLYSHEEP_BASE_URL,
HOLYSHEEP_API_KEY,
GENERATION_MODEL,
TOP_K_RESULTS
)
class RAGAgent:
"""
RAG Agent sử dụng HolySheep AI cho generation
- Retrieve context từ vector store
- Generate response với context window
"""
def __init__(self):
self.base_url = HOLYSHEEP_BASE_URL
self.api_key = HOLYSHEEP_API_KEY
self.model = GENERATION_MODEL
self.vector_store = vector_store
self.conversation_history: List[Dict] = []
def _build_system_prompt(self) -> str:
"""Xây dựng system prompt với retrieved context"""
return """Bạn là trợ lý AI chăm sóc khách hàng cho nền tảng TMĐT.
NGUYÊN TẮC QUAN TRỌNG:
1. Chỉ trả lời dựa trên thông tin được cung cấp trong Context
2. Nếu thông tin không có trong Context, hãy nói rõ: "Tôi không tìm thấy thông tin này trong cơ sở dữ liệu"
3. Trích dẫn nguồn khi có thể
4. Trả lời ngắn gọn, thân thiện, bằng tiếng Việt
5. Nếu cần thông tin thêm, hỏi khách hàng một cách cụ thể"""
def _build_user_prompt(self, query: str, context_docs: List) -> str:
"""Xây dựng user prompt với retrieved documents"""
# Format context
context_parts = []
for i, (doc, score) in enumerate(context_docs, 1):
context_parts.append(f"[Tài liệu {i}] (Độ tương đồng: {score:.2%})")
context_parts.append(f"Nội dung: {doc.page_content}")
context_parts.append(f"Nguồn: {doc.metadata}")
context_parts.append("---")
context = "\n".join(context_parts)
# Include conversation history
history = ""
if self.conversation_history:
history_parts = []
for msg in self.conversation_history[-3:]: # Last 3 turns
history_parts.append(f"{msg['role']}: {msg['content']}")
history = "Lịch sử hội thoại:\n" + "\n".join(history_parts) + "\n\n"
return f"""{history}
Context (Thông tin từ cơ sở dữ liệu):
{context}
Câu hỏi của khách hàng:
{query}
Trả lời:"""
def chat(self, query: str, use_rag: bool = True) -> Dict:
"""
Main chat method - RAG pipeline
Args:
query: User query
use_rag: Enable/disable RAG retrieval
Returns:
Dict với response, sources, và metadata
"""
start_time = datetime.now()
if use_rag:
# Step 1: Retrieve relevant documents
retrieved_docs = self.vector_store.similarity_search(
query=query,
top_k=TOP_K_RESULTS,
threshold=0.7
)
if retrieved_docs:
context = retrieved_docs
else:
# Fallback: không có context phù hợp
context = []
print("⚠️ Không tìm thấy context phù hợp, sử dụng general knowledge")
else:
context = []
# Step 2: Build prompts
system_prompt = self._build_system_prompt()
user_prompt = self._build_user_prompt(query, context)
# Step 3: Call HolySheep API - KHÔNG dùng api.openai.com
response = self._call_llm(system_prompt, user_prompt)
# Step 4: Update conversation history
self.conversation_history.append({"role": "user", "content": query})
self.conversation_history.append({"role": "assistant", "content": response})
# Calculate latency
latency = (datetime.now() - start_time).total_seconds() * 1000
return {
"response": response,
"sources": [doc.metadata for doc, _ in context] if context else [],
"retrieved_count": len(context),
"latency_ms": round(latency, 2),
"model_used": self.model
}
def _call_llm(self, system_prompt: str, user_prompt: str) -> str:
"""Call HolySheep API cho generation"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3, # Lower temperature cho factual responses
"max_tokens": 1000
},
timeout=30
)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"]
Khởi tạo RAG Agent
rag_agent = RAGAgent()Bước 5: Flask API Server với Canary Deploy
# app.py
from flask import Flask, request, jsonify
from rag_agent import rag_agent
import time
import logging
app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
@app.route("/api/chat", methods=["POST"])
def chat():
"""
Chat endpoint cho RAG Agent
Request body:
{
"query": "Chính sách đổi trả 30 ngày có áp dụng cho sản phẩm điện tử không?",
"use_rag": true
}
"""
try:
data = request.json
query = data.get("query", "")
use_rag = data.get("use_rag", True)
if not query:
return jsonify({"error": "Query is required"}), 400
# Call RAG Agent
result = rag_agent.chat(query, use_rag=use_rag)
return jsonify({
"success": True,
"data": result
})
except Exception as e:
logging.error(f"Chat error: {str(e)}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route("/api/health", methods=["GET"])
def health():
"""Health check endpoint cho canary deployment"""
return jsonify({
"status": "healthy",
"model": rag_agent.model,
"vector_store_count": rag_agent.vector_store.index.ntotal
})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=False)# canary_deploy.py
"""
Canary Deployment Script cho RAG Agent
- Rolling update 10% → 50% → 100% traffic
- Automatic rollback nếu error rate > 5%
"""
import requests
import time
import statistics
from datetime import datetime
class CanaryDeployer:
def __init__(self, base_url: str):
self.base_url = base_url
self.stages = [
{"name": "canary_10", "traffic_percent": 10, "duration_minutes": 5},
{"name": "canary_50", "traffic_percent": 50, "duration_minutes": 10},
{"name": "production", "traffic_percent": 100, "duration_minutes": 0}
]
self.error_threshold = 0.05 # 5%
def run_load_test(self, duration_seconds: int = 30) -> dict:
"""Run load test và collect metrics"""
latencies = []
errors = 0
successes = 0
test_queries = [
"Chính sách đổi trả như thế nào?",
"Sản phẩm A còn hàng không?",
"Làm sao để theo dõi đơn hàng?"
]
start_time = time.time()
while time.time() - start_time < duration_seconds:
query = test_queries[int(time.time()) % len(test_queries)]
try:
response = requests.post(
f"{self.base_url}/api/chat",
json={"query": query, "use_rag": True},
timeout=10
)
if response.status_code == 200:
data = response.json()
if data.get("success"):
latencies.append(data["data"]["latency_ms"])
successes += 1
else:
errors += 1
else:
errors += 1
except Exception as e:
errors += 1
print(f"Error: {e}")
time.sleep(0.5) # 2 requests/second
total = successes + errors
error_rate = errors / total if total > 0 else 1
return {
"total_requests": total,
"successes": successes,
"errors": errors,
"error_rate": error_rate,
"avg_latency_ms": statistics.mean(latencies) if latencies else 0,
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0
}
def deploy_stage(self, stage: dict) -> bool:
"""Deploy một stage và verify"""
print(f"\n{'='*60}")
print(f"🚀 Deploying: {stage['name']}")
print(f" Traffic: {stage['traffic_percent']}%")
print(f" Duration: {stage['duration_minutes']} minutes")
print(f"{'='*60}")
# Run load test
duration = stage['duration_minutes'] * 60 if stage['duration_minutes'] > 0 else 30
metrics = self.run_load_test(duration)
print(f"\n📊 Metrics:")
print(f" Total Requests: {metrics['total_requests']}")
print(f" Success Rate: {(1-metrics['error_rate'])*100:.2f}%")
print(f" Avg Latency: {metrics['avg_latency_ms']:.2f}ms")
print(f" P95 Latency: {metrics['p95_latency_ms']:.2f}ms")
# Check error threshold
if metrics['error_rate'] > self.error_threshold:
print(f"\n❌ ERROR RATE {metrics['error_rate']*100:.2f}% > {self.error_threshold*100}%")
print(f" Rolling back...")
return False
# Check latency threshold (< 200ms for P95)
if metrics['p95_latency_ms'] > 200:
print(f"\n⚠️ P95 Latency {metrics['p95_latency_ms']:.2f}ms > 200ms")
print(f" Investigate before proceeding...")
print(f"\n✅ Stage {stage['name']} passed!")
return True
def deploy(self):
"""Full canary deployment pipeline"""
for stage in self.stages:
success = self.deploy_stage(stage)
if not success:
print("\n🚨 CANARY DEPLOYMENT FAILED - ROLLING BACK")
return False
if stage['duration_minutes'] > 0:
time.sleep(2) # Brief pause between stages
print("\n🎉 FULL DEPLOYMENT SUCCESSFUL!")
return True
if __name__ == "__main__":
deployer = CanaryDeployer(base_url="http://localhost:5000")
deployer.deploy()Kết quả thực tế: 30 ngày sau go-live
| Metric | Trước migration | Sau migration (HolySheep AI) | Improvement |
|---|---|---|---|
| Độ trễ trung bình | 420ms | 180ms | -57% |
| Chi phí hàng tháng | $4,200 | $680 | -84% |
| Accuracy (không hallucinate) | 62% | 94% | +52% |
| P95 Latency | 890ms | 320ms | -64% |
So sánh chi phí chi tiết
# cost_calculator.py
"""
So sánh chi phí: OpenAI vs HolySheep AI
Báo cáo tháng: 2 triệu requests
- 40% retrieval (embedding): ~800K tokens input
- 60% generation: ~1.2M tokens output
"""
OpenAI Pricing (thực tế 2025)
OPENAI_COST = {
"gpt-4-turbo": {
"input": 0.01 / 1000, # $10/1M tokens
"output": 0.03 / 1000, # $30/1M tokens
"embedding": 0.0001 / 1000 # $0.10/1M tokens
}
}
HolySheep AI Pricing (2026)
HOLYSHEEP_COST = {
"gemini-2.5-flash": {
"input": 2.50 / 1_000_000, # $2.50/1M tokens
"output": 2.50 / 1_000_000,
"embedding": 0.42 / 1_000_000 # DeepSeek V3.2: $0.42/1M tokens
}
}
def calculate_monthly_cost(provider: str, requests: int = 2_000_000):
"""Tính chi phí hàng tháng cho mỗi provider"""
if provider == "openai":
# 40% embedding, 60% generation
embedding_tokens = requests * 0.4 * 500 # ~500 tokens/doc
generation_input = requests * 0.6 * 100 # ~100 tokens input
generation_output = requests * 0.6 * 150 # ~150 tokens output
cost = (
embedding_tokens * OPENAI_COST["gpt-4-turbo"]["embedding"] +
generation_input * OPENAI_COST["gpt-4-turbo"]["input"] +
generation_output * OPENAI_COST["gpt-4-turbo"]["output"]
)
elif provider == "holysheep":
embedding_tokens = requests * 0.4 * 500
generation_input = requests * 0.6 * 100
generation_output = requests * 0.6 * 150
cost = (
embedding_tokens * HOLYSHEEP_COST["gemini-2.5-flash"]["embedding"] +
generation_input * HOLYSHEEP_COST["gemini-2.5-flash"]["input"] +
generation_output * HOLYSHEEP_COST["gemini-2.5-flash"]["output"]
)
return cost
Tính toán
openai_monthly = calculate_monthly_cost("openai")
holysheep_monthly = calculate_monthly_cost("holysheep")
print(f"📊 So sánh chi phí (2 triệu requests/tháng):")
print(f" OpenAI (GPT-4): ${openai_monthly:,.2f}")
print(f" HolySheep AI: ${holysheep_monthly:,.2f}")
print(f" Tiết kiệm: ${openai_monthly - holysheep_monthly:,.2f} ({(1-holysheep_monthly/openai_monthly)*100:.1f}%)")
Output:
📊 So sánh chi phí (2 triệu requests/tháng):
OpenAI (GPT-4): $4,200.00
HolySheep AI: $678.50
Tiết kiệm: $3,521.50 (83.9%)
Lỗi thường gặp và cách khắc phục
Lỗi 1: Connection Timeout khi embedding batch lớn
# ❌ SAI - Gây timeout khi batch > 100 documents
def embed_large_batch(self, documents: list):
vectors = self.embeddings_model.embed_documents(documents) # 500+ docs
# Timeout sau 30s khi gọi API liên tục
✅ ĐÚNG - Chunk processing với exponential backoff
def embed_large_batch_optimized(self, documents: list, batch_size: int = 50):
"""Embed documents theo batch với retry logic"""
import time
import requests
all_vectors = []
max_retries = 3
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
retries = 0
while retries < max_retries:
try:
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-embed-v3",
"input": batch
},
timeout=60 # Tăng timeout cho batch lớn
)
response.raise_for_status()
data = response.json()
all_vectors.extend([item["embedding"] for item in data["data"]])
break # Success, exit retry loop
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError) as e:
retries += 1
wait_time = (2 ** retries) * 1.0 # 2s, 4s, 8s
print(f"Retry {retries}/{max_retries} sau {wait_time}s")
time.sleep(wait_time)
except requests.exceptions.HTTPError as e:
# Handle rate limit (429)
if e.response.status_code == 429:
retry_after = int(e.response.headers.get("Retry-After", 60))
print(f"Rate limited, chờ {retry_after}s")
time.sleep(retry_after)
else:
raise
return all_vectorsLỗi 2: Similarity threshold quá cao - không retrieve được gì
# ❌ SAI - Threshold 0.9 quá nghiêm ngặt
results = vector_store.similarity_search(query, top_k=5, threshold=0.9)
Kết quả: []
✅ ĐÚNG - Dynamic threshold dựa trên query type
def smart_retrieval(query: str, vector_store, min_threshold: float = 0.5):
"""Điều chỉnh threshold theo loại query"""
# Query cần precision cao (factual questions)
if any(keyword in query.lower() for keyword in
["chính sách", "giá", "thông số", "bảo hành"]):
threshold = 0.7
top_k = 5
# Query cần recall cao (exploratory questions)
elif any(keyword in query.lower() for keyword in
["gợi ý", "recommend", "phù hợp", "tư vấn"]):
threshold = 0.5
top_k = 10
else:
threshold = 0.6
top_k = 5
results = vector_store.similarity_search(
query,
top_k=top_k,
threshold=threshold
)
# Fallback: giảm threshold nếu không có kết quả
if not results and threshold > min_threshold:
print(f"⚠️ Không có kết quả với threshold {threshold}, thử {min_threshold}")
results = vector_store.similarity_search(
query,
top_k=top_k,
threshold=min_threshold
)
return results
Usage
retrieved = smart_retrieval(
"Chính sách đổi trả điện thoại iPhone 15 Pro",
vector_store
)Lỗi 3: Context overflow - prompt quá dài cho model
# ❌ SAI - Ghép tất cả retrieved docs, có thể vượt context limit
def build_prompt_unsafe(query, retrieved_docs):
context = "\n".join([doc.page_content for doc in retrieved_docs])
# 20 docs * 500 tokens = 10,000 tokens - vượt limit nhiều model
✅ ĐÚNG - Smart context truncation
def build_prompt_safe(query: str, retrieved_docs: list, max_context_tokens: int = 4000):
"""
Xây dựng prompt với context window management
Args:
query: User query
retrieved_docs: List of (Document, score) tuples
max_context_tokens: Giới hạn tokens cho context
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Sort by relevance score (descending)
sorted_docs = sorted(retrieved_docs, key=lambda x: x[1], reverse=True)
context_parts = []
current_tokens = 0
# Estimate: 1 token ≈ 4 characters
token估算 = lambda text: len(text) // 4
for doc, score in sorted_docs:
doc_tokens = token估算(doc.page_content)
# Stop nếu sẽ vượt limit
if current_tokens + doc_tokens > max_context_tokens:
Tài nguyên liên quan
Bài viết liên quan