Là một kỹ sư backend có 7 năm kinh nghiệm triển khai hệ thống tìm kiếm, tôi đã chứng kiến quá nhiều đội ngũ vật lộn với việc tích hợp semantic search vào codebase của họ. Trong bài viết này, tôi sẽ chia sẻ một case study thực tế và hướng dẫn chi tiết cách di chuyển hệ thống để đạt hiệu suất tối ưu.
Bối Cảnh: Startup AI Ở Hà Nội Gặp Khó Khăn
Tình huống ban đầu: Một startup AI ở Hà Nội chuyên cung cấp giải pháp tìm kiếm thông minh cho các sàn thương mại điện tử Việt Nam. Đội ngũ 12 kỹ sư, codebase 300,000 dòng Python/Go, phục vụ 2 triệu request mỗi ngày.
Điểm đau của nhà cung cấp cũ:
- Độ trễ trung bình 420ms mỗi truy vấn semantic search
- Chi phí API $4,200/tháng — quá cao so với ngân sách Series A
- Không hỗ trợ tiếng Việt tốt, đặc biệt là tìm kiếm có dấu/không dấu
- Thời gian chờ phản hồi không ổn định, peak time lên tới 1.2 giây
- Không có data center ở khu vực Đông Nam Á
Lý do chọn HolySheep AI:
- Tỷ giá quy đổi theo USD: ¥1 = $1 — tiết kiệm 85%+ chi phí
- Độ trễ <50ms với server Đông Nam Á
- Hỗ trợ thanh toán WeChat/Alipay — thuận tiện cho các startup có nguồn vốn Trung Quốc
- Tín dụng miễn phí khi đăng ký tài khoản
- API endpoint tương thích với OpenAI format
Kết Quả Sau 30 Ngày Go-Live
Sau khi hoàn tất di chuyển sang HolySheep AI, đội ngũ đã ghi nhận những cải thiện ngoạn mục:
- Độ trễ: 420ms → 180ms (giảm 57%)
- Chi phí hàng tháng: $4,200 → $680 (tiết kiệm 84%)
- Throughput: Tăng 3.2x với cùng cấu hình server
- Error rate: Giảm từ 2.1% xuống 0.08%
- Support SLA: Phản hồi trong 15 phút thay vì 24 giờ
Các Bước Di Chuyển Chi Tiết
Bước 1: Cập Nhật Cấu Hình Base URL và API Key
Việc đầu tiên cần làm là cập nhật tất cả các file cấu hình environment. Tôi khuyến nghị sử dụng config map trong Kubernetes hoặc environment variable để quản lý tập trung.
# File: config/settings.py
import os
from dataclasses import dataclass
@dataclass
class HolySheepConfig:
"""Cấu hình HolySheep AI - Semantic Search"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
embedding_model: str = "text-embedding-3-large"
search_model: str = "gpt-4.1" # $8/MTok
max_tokens: int = 2048
temperature: float = 0.7
timeout: int = 30
Khởi tạo singleton config
config = HolySheepConfig()
Validation khi khởi động
def validate_config():
if config.api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("HOLYSHEEP_API_KEY chưa được cấu hình!")
if not config.base_url.startswith("https://api.holysheep.ai"):
raise ValueError("base_url phải sử dụng endpoint HolySheep AI")
print(f"✓ HolySheep AI configured: {config.base_url}")
print(f"✓ Embedding model: {config.embedding_model}")
print(f"✓ Search model: {config.search_model}")
# File: .env.example
=== HOLYSHEEP AI CONFIGURATION ===
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Model selection (so sánh chi phí 2026)
HOLYSHEEP_EMBEDDING_MODEL=text-embedding-3-large
HOLYSHEEP_SEARCH_MODEL=gpt-4.1 # $8/MTok
Các option khác:
- claude-sonnet-4.5: $15/MTok (chất lượng cao hơn)
- gemini-2.5-flash: $2.50/MTok (tiết kiệm)
- deepseek-v3.2: $0.42/MTok (rẻ nhất)
Timeout settings
HOLYSHEEP_TIMEOUT=30
HOLYSHEEP_MAX_RETRIES=3
Rate limiting
HOLYSHEEP_REQUESTS_PER_MINUTE=1000
HOLYSHEEP_TOKENS_PER_MINUTE=100000
Bước 2: Triển Khhai Canary Deploy
Để đảm bảo zero-downtime migration, tôi áp dụng canary deploy: 5% → 25% → 50% → 100% traffic trong 7 ngày.
# File: services/holy_sheep_client.py
import httpx
import asyncio
from typing import List, Dict, Optional
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class SemanticSearchResult:
"""Kết quả tìm kiếm ngữ nghĩa"""
content: str
score: float
metadata: Dict
latency_ms: float
class HolySheepSemanticSearch:
"""
Client tìm kiếm ngữ nghĩa sử dụng HolySheep AI
- Base URL: https://api.holysheep.ai/v1
- Hỗ trợ tiếng Việt tối ưu
- Độ trễ <50ms với server SEA
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url.rstrip("/")
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.timeout = httpx.Timeout(30.0, connect=5.0)
async def embed_texts(self, texts: List[str]) -> List[List[float]]:
"""
Tạo embeddings cho danh sách văn bản
Sử dụng endpoint /embeddings của HolySheep
"""
async with httpx.AsyncClient(timeout=self.timeout) as client:
payload = {
"input": texts,
"model": "text-embedding-3-large",
"encoding_format": "float"
}
response = await client.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json=payload
)
response.raise_for_status()
data = response.json()
return [item["embedding"] for item in data["data"]]
async def semantic_search(
self,
query: str,
documents: List[str],
top_k: int = 5,
threshold: float = 0.7
) -> List[SemanticSearchResult]:
"""
Tìm kiếm ngữ nghĩa trong codebase
- query: câu hỏi tìm kiếm (hỗ trợ tiếng Việt)
- documents: danh sách documents cần tìm
- top_k: số lượng kết quả trả về
"""
import time
start_time = time.time()
# Bước 1: Embed query và documents
embeddings = await self.embed_texts([query] + documents)
query_embedding = embeddings[0]
doc_embeddings = embeddings[1:]
# Bước 2: Tính cosine similarity
results = []
for idx, (doc, emb) in enumerate(zip(documents, doc_embeddings)):
similarity = self._cosine_similarity(query_embedding, emb)
if similarity >= threshold:
results.append(SemanticSearchResult(
content=doc,
score=similarity,
metadata={"index": idx},
latency_ms=(time.time() - start_time) * 1000
))
# Bước 3: Sort và return top_k
results.sort(key=lambda x: x.score, reverse=True)
return results[:top_k]
@staticmethod
def _cosine_similarity(a: List[float], b: List[float]) -> float:
"""Tính cosine similarity giữa 2 vectors"""
dot_product = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
return dot_product / (norm_a * norm_b + 1e-8)
# File: services/canary_deploy.py
import random
import hashlib
from typing import Callable, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
@dataclass
class CanaryConfig:
"""Cấu hình Canary Deploy"""
stage: str = "production" # canary | production
canary_percentage: int = 5 # 5% traffic đi qua HolySheep
rollout_days: int = 7
start_date: Optional[datetime] = None
def should_use_holy_sheep(self, user_id: str) -> bool:
"""
Quyết định request có đi qua HolySheep không
- Hash user_id để đảm bảo consistent routing
- Dần tăng percentage theo rollout schedule
"""
if self.stage == "production":
return True
# Hash user_id thành số 0-99
hash_value = int(hashlib.md5(f"{user_id}{self.start_date}".encode()).hexdigest(), 16) % 100
# Tính toán percentage theo ngày rollout
if self.start_date:
days_elapsed = (datetime.now() - self.start_date).days
if days_elapsed >= self.rollout_days:
return True
current_percentage = min(100, (days_elapsed + 1) * 100 // self.rollout_days)
else:
current_percentage = self.canary_percentage
return hash_value < current_percentage
class HybridSearchRouter:
"""
Router chuyển đổi giữa legacy system và HolySheep AI
- Canary: %5 → 25% → 50% → 100%
- Monitoring: latency, error rate, cost
"""
def __init__(self, config: CanaryConfig):
self.config = config
self.metrics = {
"total_requests": 0,
"holy_sheep_requests": 0,
"legacy_requests": 0,
"holy_sheep_latency_ms": [],
"legacy_latency_ms": []
}
async def search(self, query: str, user_id: str) -> Any:
"""Thực hiện tìm kiếm với canary routing"""
self.metrics["total_requests"] += 1
if self.config.should_use_holy_sheep(user_id):
return await self._search_holy_sheep(query, user_id)
else:
return await self._search_legacy(query, user_id)
async def _search_holy_sheep(self, query: str, user_id: str) -> Any:
"""Tìm kiếm qua HolySheep AI"""
import time
self.metrics["holy_sheep_requests"] += 1
start = time.time()
try:
# Import và gọi HolySheep client
from services.holy_sheep_client import HolySheepSemanticSearch
client = HolySheepSemanticSearch(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
result = await client.semantic_search(query, documents=[])
latency = (time.time() - start) * 1000
self.metrics["holy_sheep_latency_ms"].append(latency)
logger.info(f"HolySheep latency: {latency:.2f}ms")
return result
except Exception as e:
logger.error(f"HolySheep error: {e}, falling back to legacy")
return await self._search_legacy(query, user_id)
async def _search_legacy(self, query: str, user_id: str) -> Any:
"""Fallback sang legacy system"""
import time
self.metrics["legacy_requests"] += 1
start = time.time()
# Legacy search implementation
latency = (time.time() - start) * 1000
self.metrics["legacy_latency_ms"].append(latency)
return {"source": "legacy", "latency_ms": latency}
def get_metrics(self) -> dict:
"""Lấy metrics để monitoring"""
holy_sheep_avg = sum(self.metrics["holy_sheep_latency_ms"]) / len(self.metrics["holy_sheep_latency_ms"]) if self.metrics["holy_sheep_latency_ms"] else 0
return {
"total_requests": self.metrics["total_requests"],
"holy_sheep_pct": self.metrics["holy_sheep_requests"] / max(1, self.metrics["total_requests"]) * 100,
"avg_latency_holy_sheep_ms": round(holy_sheep_avg, 2),
"avg_latency_legacy_ms": round(sum(self.metrics["legacy_latency_ms"]) / max(1, len(self.metrics["legacy_latency_ms"])), 2)
}
Bước 3: Xoay API Key An Toàn
# File: scripts/rotate_api_key.py
#!/usr/bin/env python3
"""
Script xoay API key HolySheep an toàn
Chạy trước khi deploy production
"""
import os
import sys
import base64
import secrets
from datetime import datetime
def generate_new_api_key() -> str:
"""Tạo API key mới theo format HolySheep"""
# Format: hs_live_ + 48 ký tự random
prefix = "hs_live_"
random_part = secrets.token_hex(24) # 48 ký tự hex
return prefix + random_part
def validate_api_key(key: str) -> bool:
"""Validate format API key"""
if not key.startswith("hs_live_"):
return False
if len(key) != 55: # hs_live_ (8) + 48 ký tự
return False
return True
def update_env_file(new_key: str, env_file: str = ".env.production"):
"""Cập nhật .env file với key mới"""
backup_file = f"{env_file}.backup.{datetime.now().strftime('%Y%m%d_%H%M%S')}"
with open(env_file, 'r') as f:
content = f.read()
# Backup file cũ
with open(backup_file, 'w') as f:
f.write(content)
print(f"✓ Backup created: {backup_file}")
# Thay thế key cũ
lines = content.split('\n')
new_lines = []
for line in lines:
if line.startswith('HOLYSHEEP_API_KEY='):
new_lines.append(f'HOLYSHEEP_API_KEY={new_key}')
print(f"✓ API key updated")
else:
new_lines.append(line)
with open(env_file, 'w') as f:
f.write('\n'.join(new_lines))
return True
def notify_team(new_key: str):
"""Gửi notification cho team về key rotation"""
print(f"""
╔══════════════════════════════════════════════════════════╗
║ 🔄 HOLYSHEEP API KEY ROTATED ║
╠══════════════════════════════════════════════════════════╣
║ New Key: {new_key[:20]}...{'*' * 30} ║
║ Time: {datetime.now().isoformat()} ║
║ Next Rotation: {(datetime.now() + timedelta(days=90)).date()} ║
╚══════════════════════════════════════════════════════════╝
""")
if __name__ == "__main__":
print("=== HolySheep API Key Rotation ===\n")
# Generate new key
new_key = generate_new_api_key()
print(f"Generated new key: {new_key[:20]}...")
# Validate
if not validate_api_key(new_key):
print("❌ Invalid key format!")
sys.exit(1)
print("✓ Key format validated")
# Update .env (chạy trong CI/CD pipeline)
if "--dry-run" not in sys.argv:
update_env_file(new_key)
notify_team(new_key)
print("\n✅ Key rotation completed!")
else:
print("\n[DRY RUN] Skipping actual update")
Bảng So Sánh Chi Phí 2026
Để giúp các bạn đưa ra quyết định tối ưu chi phí, đây là bảng so sánh giá các model phổ biến trên HolySheep AI:
| Model | Giá/MTok | Use Case | Tốc độ |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 | Batch processing, cost optimization | ★★★★★ |
| Gemini 2.5 Flash | $2.50 | General purpose, balanced | ★★★★☆ |
| GPT-4.1 | $8.00 | High quality reasoning | ★★★☆☆ |
| Claude Sonnet 4.5 | $15.00 | Premium quality, complex tasks | ★★★☆☆ |
Với startup Hà Nội trong case study, việc chuyển từ GPT-4o ($30/MTok) sang DeepSeek V3.2 ($0.42/MTok) giúp tiết kiệm 98.6% chi phí embedding.
Code Tích Hợp Search Trong Codebase Thực Tế
# File: routers/codebase_search.py
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
from services.holy_sheep_client import HolySheepSemanticSearch
from services.canary_deploy import HybridSearchRouter, CanaryConfig
import logging
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/search", tags=["codebase-search"])
Dependency injection cho HolySheep client
def get_holy_sheep_client() -> HolySheepSemanticSearch:
return HolySheepSemanticSearch(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
def get_canary_router() -> HybridSearchRouter:
return HybridSearchRouter(CanaryConfig(
stage="production",
canary_percentage=100 # 100% production sau khi validate
))
class SearchRequest(BaseModel):
"""Request body cho tìm kiếm codebase"""
query: str
repository_id: str
top_k: int = 10
file_extensions: Optional[List[str]] = None
language: Optional[str] = "vi" # Hỗ trợ tiếng Việt
class SearchResponse(BaseModel):
"""Response body"""
results: List[dict]
total_found: int
search_time_ms: float
source: str # "holy_sheep" | "legacy"
@router.post("/codebase", response_model=SearchResponse)
async def search_codebase(
request: SearchRequest,
client: HolySheepSemanticSearch = Depends(get_holy_sheep_client),
router: HybridSearchRouter = Depends(get_canary_router)
):
"""
Tìm kiếm ngữ nghĩa trong codebase
Ví dụ request:
{
"query": "hàm xử lý thanh toán online",
"repository_id": "repo-123",
"top_k": 10,
"language": "vi"
}
"""
import time
start_time = time.time()
try:
# Load documents từ repository
documents = await load_repository_documents(
request.repository_id,
request.file_extensions
)
if not documents:
raise HTTPException(status_code=404, detail="No documents found")
# Thực hiện semantic search
results = await client.semantic_search(
query=request.query,
documents=[doc["content"] for doc in documents],
top_k=request.top_k,
threshold=0.6
)
search_time = (time.time() - start_time) * 1000
return SearchResponse(
results=[
{
"file": documents[r.metadata["index"]]["file"],
"line": documents[r.metadata["index"]]["line"],
"content": r.content,
"score": round(r.score, 4),
"language": documents[r.metadata["index"]].get("language", "unknown")
}
for r in results
],
total_found=len(results),
search_time_ms=round(search_time, 2),
source="holy_sheep"
)
except Exception as e:
logger.error(f"Search error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
async def load_repository_documents(
repo_id: str,
extensions: Optional[List[str]] = None
) -> List[dict]:
"""
Load documents từ repository để index
Hỗ trợ: .py, .js, .ts, .go, .java, .md
"""
# Implementation sẽ đọc từ git/database
return [
{
"file": "payment_service.py",
"line": 42,
"content": "def process_online_payment(order_id, amount):",
"language": "python"
}
# ... more documents
]
Lỗi Thường Gặp Và Cách Khắc Phục
Qua quá trình triển khai cho nhiều khách hàng, tôi đã gặp và xử lý các lỗi phổ biến sau:
Lỗi 1: Lỗi Authentication - "Invalid API Key"
Mô tả: Khi mới bắt đầu, nhiều bạn copy sai format API key hoặc quên thay thế placeholder.
# ❌ SAI - Copy placeholder trực tiếp
client = HolySheepSemanticSearch(
api_key="YOUR_HOLYSHEEP_API_KEY" # Lỗi đây!
)
✅ ĐÚNG - Load từ environment variable
import os
client = HolySheepSemanticSearch(
api_key=os.environ.get("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1" # Phải match với key scope
)
Verify connection
import httpx
async def verify_connection():
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY')}"}
)
if response.status_code == 200:
print("✅ HolySheep AI connection verified!")
elif response.status_code == 401:
raise ValueError("❌ Invalid API Key - check HOLYSHEEP_API_KEY environment variable")
elif response.status_code == 403:
raise ValueError("❌ API Key lacks permissions - ensure key has embedding access")
Lỗi 2: Độ Trễ Cao - Request Timeout
Mô tả: Độ trễ vượt ngưỡng timeout mặc định, đặc biệt khi xử lý batch lớn.
# ❌ SAI - Timeout quá ngắn cho batch processing
client = HolySheepSemanticSearch(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
timeout=httpx.Timeout(5.0) # Chỉ 5 giây - không đủ!
)
✅ ĐÚNG - Dynamic timeout theo batch size
from typing import List
def calculate_timeout(batch_size: int) -> httpx.Timeout:
"""Tính timeout phù hợp với batch size"""
base_time = 10.0 # 10 giây base
per_item_time = 0.1 # 100ms per item
timeout = base_time + (batch_size * per_item_time)
return httpx.Timeout(
timeout.connect,
timeout.read=min(timeout, 60.0) # Max 60 giây
)
class OptimizedHolySheepClient(HolySheepSemanticSearch):
"""Client với timeout thông minh và retry logic"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
super().__init__(api_key, base_url)
self.max_retries = 3
self.retry_delay = 1.0
async def embed_texts_with_retry(self, texts: List[str]) -> List[List[float]]:
"""Embed với retry logic cho độ tin cậy cao"""
for attempt in range(self.max_retries):
try:
return await self.embed_texts(texts)
except httpx.TimeoutException as e:
if attempt == self.max_retries - 1:
raise
wait_time = self.retry_delay * (2 ** attempt)
print(f"⏳ Retry {attempt + 1} after {wait_time}s timeout...")
import asyncio
await asyncio.sleep(wait_time)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429: # Rate limit
import asyncio
await asyncio.sleep(5)
else:
raise
Lỗi 3: Semantic Search Không Tìm Thấy Kết Quả Tiếng Việt
Mô tả: Query tiếng Việt không trả về kết quả chính xác, đặc biệt với các từ có dấu/không dấu.
# ❌ SAI - Không xử lý text normalization
async def basic_search(query: str, documents: List[str]):
results = await client.semantic_search(query, documents)
return results
✅ ĐÚNG - Text normalization cho tiếng Việt
import unicodedata
import re
def normalize_vietnamese_text(text: str) -> str:
"""
Normalize tiếng Việt:
1. Lowercase
2. Remove accents (optional - cho fuzzy matching)
3. Normalize unicode
4. Remove extra spaces
"""
# Lowercase
text = text.lower()
# Normalize unicode (NFC → NFD)
text = unicodedata.normalize('NFD', text)
# Remove combining diacritical marks (bỏ dấu)
text = ''.join(c for c in text if unicodedata.category(c) != 'Mn')
# Remove special characters nhưng giữ tiếng Việt
text = re.sub(r'[^\w\sàáảãạăằắẳẵặâầấẩẫậèéẻẽẹêềếểễệìíỉĩịòóỏõọôồốổỗộơờớởỡợùúủũụưừứửữựỳýỷỹỵđ]', ' ', text)
# Normalize spaces
text = ' '.join(text.split())
return text
class VietnameseAwareSearch(HolySheepSemanticSearch):
"""Client tìm kiếm tối ưu cho tiếng Việt"""
async def semantic_search_vietnamese(
self,
query: str,
documents: List[str],
top_k: int = 5,
search_both_normalized: bool = True # Tìm cả normalized và original
) -> List[SemanticSearchResult]:
"""Tìm kiếm với Vietnamese-aware preprocessing"""
# Normalize query
normalized_query = normalize_vietnamese_text(query)
# Search với query gốc
results_original = await self.semantic_search(
query, documents, top_k=top_k * 2, threshold=0.5
)
# Search với query normalized (để bắt các từ không dấu)
if search_both_normalized and normalized_query != query:
results_normalized = await self.semantic_search(
normalized_query, documents, top_k=top_k * 2, threshold=0.5
)
# Merge results, deduplicate by content
seen = set()
merged_results = []
for r in results_original + results_normalized:
if r.content not in seen:
seen.add(r.content)
merged_results.append(r)
# Sort lại và return top_k
merged_results.sort(key=lambda x: x.score, reverse=True)
return merged_results[:top_k]
return results_original[:top_k]
Lỗi 4: Rate Limit Exceeded
Mô tả: Vượt quá giới hạn request/_phút của HolySheep AI.
# ❌ SAI - Không có rate limiting
for doc in documents:
await client.embed_texts([doc]) # Có thể trigger rate limit
✅ ĐÚNG - Implement rate limiter
import asyncio
from collections import deque
from datetime import datetime, timedelta
class RateLimiter:
"""Token bucket rate limiter cho HolySheep API"""
def __init__(self, requests_per_minute: int = 1000):
self.requests_per_minute = requests_per_minute
self.requests = deque()
self._lock = asyncio.Lock()
async def acquire(self):
"""Chờ đến khi có quota available"""
async with self._lock:
now = datetime.now()
# Remove requests cũ hơn 1 phút
while self.requests and self.requests[0] < now - timedelta(minutes=1):
self.requests.popleft()
if len(self.requests) >= self.requests_per_minute:
# Tính thời gian chờ
wait_time = (self.requests[0] - (now - timedelta(minutes=1))).total_seconds()
print(f"⏳ Rate limit reached, waiting {wait_time:.2f}s...")
await asyncio.sleep(wait_time + 0.1)
self.requests.append(now)
async def embed_with_rate_limit(self, client, texts: List[str], batch_size: int = 100):
"""Embed documents với rate limiting"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
await self.acquire() # Chờ quota
embeddings = await client.embed_texts(batch)
all_embeddings.extend(embed