Trong bối cảnh chi phí AI tăng phi mã từ đầu 2026, việc tìm kiếm giải pháp long context RAG vừa mạnh mẽ vừa tiết kiệm trở nên cấp thiết hơn bao giờ hết. Là một kỹ sư đã triển khai hệ thống RAG cho 3 doanh nghiệp lớn tại Việt Nam, tôi đã trải qua quá trình thử nghiệm, đánh giá và cuối cùng chọn DeepSeek V3 thông qua HolySheep API làm backbone chính. Bài viết này sẽ chia sẻ toàn bộ kinh nghiệm thực chiến, từ lý thuyết đến code production-ready.

1. Tại Sao DeepSeek V3 Là Lựa Chọn Số Một Cho RAG Năm 2026

1.1 Bảng So Sánh Chi Phí Thực Tế (Cập Nhật Tháng 5/2026)

Model Output Cost ($/MTok) 10M Token/Tháng ($) Context Window Đánh giá
GPT-4.1 $8.00 $80 128K Đắt đỏ cho production
Claude Sonnet 4.5 $15.00 $150 200K Chất lượng cao nhưng rất tốn kém
Gemini 2.5 Flash $2.50 $25 1M Cân bằng nhưng vẫn đắt hơn DeepSeek 6x
DeepSeek V3.2 $0.42 $4.20 640K ⭐ Best Value - Tiết kiệm 95%+

💡 Lưu ý quan trọng: Với mức giá $0.42/MTok thông qua HolySheep AI, doanh nghiệp của bạn chỉ cần $4.20 để xử lý 10 triệu token mỗi tháng — rẻ hơn một ly cà phê VND 50,000!

1.2 DeepSeek V3 Có Gì Đặc Biệt Cho RAG?

2. Kiến Trúc RAG Với DeepSeek V3 Qua HolySheep

2.1 Sơ Đồ Luồng Xử Lý


┌─────────────────────────────────────────────────────────────────┐
│                    RAG ARCHITECTURE                             │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  [Documents] ──► [Chunking] ──► [Embedding] ──► [Vector DB]   │
│                         │                │              │        │
│                         ▼                ▼              ▼        │
│                   [Recursive      [bge-m3 or      [Pinecone/    │
│                    Character]     multilingual]    Chroma/Qdrant]│
│                         │                │              │        │
│                         ▼                ▼              ▼        │
│                    [User Query] ──► [Semantic    ──► [Context]  │
│                                         Search]        Retrieval│
│                                                        │        │
│                                                        ▼        │
│                                           [DeepSeek V3 via     │
│                                            HolySheep API]      │
│                                                        │        │
│                                                        ▼        │
│                                             [Final Response]   │
└─────────────────────────────────────────────────────────────────┘

2.2 Setup Môi Trường

# requirements.txt
openai>=1.12.0
chromadb>=0.4.22
sentence-transformers>=2.4.0
pypdf>=4.0.1
langchain>=0.1.0
langchain-community>=0.0.20
tiktoken>=0.5.2
numpy>=1.26.0

# Cài đặt
pip install -r requirements.txt


Import các thư viện cần thiết

import os
from openai import OpenAI
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
import tiktoken

3. Code Triển Khai Chi Tiết

3.1 Kết Nối DeepSeek V3 Qua HolySheep API

import os
from openai import OpenAI


============================================


HOLYSHEEP API CONFIGURATION


Base URL: https://api.holysheep.ai/v1


Rate: ¥1 = $1 (Tiết kiệm 85%+ so với OpenAI)


============================================


class DeepSeekRAGClient:
    """Client kết nối DeepSeek V3 qua HolySheep API cho hệ thống RAG"""
    
    def __init__(self, api_key: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # LUÔN dùng HolySheep endpoint
        )
        self.model = "deepseek-chat"  # DeepSeek V3
        self.embedding_model = "text-embedding-3-small"
        
    def generate_response(
        self, 
        query: str, 
        context_docs: list[str], 
        system_prompt: str = None,
        temperature: float = 0.3,
        max_tokens: int = 2000
    ) -> str:
        """
        Tạo response từ DeepSeek V3 với context từ RAG retrieval
        
        Args:
            query: Câu hỏi của user
            context_docs: Danh sách documents được retrieve
            system_prompt: System prompt tùy chỉnh
            temperature: Độ ngẫu nhiên (0.0-1.0)
            max_tokens: Số token tối đa cho response
        
        Returns:
            str: Response từ model
        """
        
        # Format context thành string
        context_text = "\n\n".join([
            f"[Document {i+1}]:\n{doc}" 
            for i, doc in enumerate(context_docs)
        ])
        
        # System prompt mặc định cho RAG
        if system_prompt is None:
            system_prompt = """Bạn là trợ lý AI chuyên trả lời câu hỏi dựa trên ngữ cảnh được cung cấp.
Hãy trả lời CHÍNH XÁC và DỰA TRÊN ngữ cảnh. Nếu không có thông tin trong ngữ cảnh, hãy nói rõ.
Trả lời bằng tiếng Việt, ngắn gọn và có cấu trúc."""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"""Ngữ cảnh:
{context_text}

Câu hỏi: {query}

Hãy trả lời dựa trên ngữ cảnh trên."""}
        ]
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens
            )
            return response.choices[0].message.content
            
        except Exception as e:
            print(f"❌ Lỗi khi gọi API: {e}")
            return f"Xin lỗi, đã xảy ra lỗi: {str(e)}"
    
    def create_embedding(self, text: str) -> list[float]:
        """
        Tạo embedding vector cho text sử dụng HolySheep embedding endpoint
        """
        try:
            response = self.client.embeddings.create(
                model=self.embedding_model,
                input=text
            )
            return response.data[0].embedding
            
        except Exception as e:
            print(f"❌ Lỗi khi tạo embedding: {e}")
            return []
    
    def batch_create_embeddings(self, texts: list[str], batch_size: int = 100) -> list[list[float]]:
        """
        Tạo embeddings cho nhiều texts (batch processing)
        Tiết kiệm API calls và giảm độ trễ
        """
        all_embeddings = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            try:
                response = self.client.embeddings.create(
                    model=self.embedding_model,
                    input=batch
                )
                batch_embeddings = [item.embedding for item in response.data]
                all_embeddings.extend(batch_embeddings)
                print(f"✅ Processed batch {i//batch_size + 1}/{(len(texts)-1)//batch_size + 1}")
                
            except Exception as e:
                print(f"❌ Lỗi batch {i//batch_size + 1}: {e}")
                all_embeddings.extend([[] for _ in batch])
        
        return all_embeddings



============================================


SỬ DỤNG CLIENT


============================================


Đăng ký và lấy API key tại: https://www.holysheep.ai/register

client = DeepSeekRAGClient(api_key="YOUR_HOLYSHEEP_API_KEY")


Test nhanh

test_response = client.generate_response(
    query="DeepSeek V3 có gì đặc biệt?",
    context_docs=["DeepSeek V3 là model mới nhất với context window 640K token."]
)
print(test_response)

3.2 Document Processing & Chunking Strategy

import tiktoken
from typing import List, Tuple
import re

class DocumentChunker:
    """
    Advanced chunking strategy cho RAG với DeepSeek V3
    - Hỗ trợ recursive character splitting
    - Preserve metadata và context
    - Tối ưu cho 640K context window của DeepSeek V3
    """
    
    def __init__(
        self, 
        chunk_size: int = 1000,      # Token count per chunk
        chunk_overlap: int = 200,    # Overlap để preserve context
        encoding_model: str = "cl100k_base"
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoding = tiktoken.get_encoding(encoding_model)
        
    def clean_text(self, text: str) -> str:
        """Clean và normalize text"""
        # Remove extra whitespace
        text = re.sub(r'\s+', ' ', text)
        # Remove special characters nhưng giữ dấu câu
        text = re.sub(r'[^\w\s.,!?;:()\-–—""''\n]', '', text)
        return text.strip()
    
    def split_by_tokens(self, text: str) -> List[int]:
        """Split text thành tokens"""
        return self.encoding.encode(text)
    
    def create_chunks(self, text: str, metadata: dict = None) -> List[dict]:
        """
        Tạo chunks từ document với overlap
        
        Args:
            text: Document text
            metadata: Metadata như source, page, date...
        
        Returns:
            List of chunk dictionaries với text, tokens, metadata
        """
        text = self.clean_text(text)
        tokens = self.split_by_tokens(text)
        total_tokens = len(tokens)
        
        chunks = []
        start = 0
        
        while start < total_tokens:
            end = min(start + self.chunk_size, total_tokens)
            chunk_tokens = tokens[start:end]
            chunk_text = self.encoding.decode(chunk_tokens)
            
            chunk_dict = {
                "text": chunk_text,
                "token_count": len(chunk_tokens),
                "start_token": start,
                "end_token": end,
                "metadata": metadata or {}
            }
            chunks.append(chunk_dict)
            
            # Di chuyển start với overlap
            start = end - self.chunk_overlap
            
            # Tránh infinite loop
            if start >= end:
                break
        
        return chunks
    
    def split_by_sentences(self, text: str) -> List[str]:
        """Split text thành sentences (cho hierarchical chunking)"""
        # Regex cho tiếng Việt và tiếng Anh
        sentence_pattern = r'(?<=[.!?])\s+|(?<=[।।!?])\s+'
        sentences = re.split(sentence_pattern, text)
        return [s.strip() for s in sentences if s.strip()]
    
    def split_by_paragraphs(self, text: str) -> List[str]:
        """Split text thành paragraphs"""
        paragraphs = text.split('\n\n')
        return [p.strip() for p in paragraphs if p.strip()]
    
    def smart_chunk(
        self, 
        text: str, 
        metadata: dict = None,
        strategy: str = "recursive"
    ) -> List[dict]:
        """
        Smart chunking với multiple strategies
        
        Strategies:
        - recursive: Split nhỏ dần cho đến khi đủ chunk_size
        - by_paragraph: Split theo paragraph (tốt cho documents có cấu trúc)
        - by_sentence: Split theo sentence (tốt cho Q&A data)
        """
        
        if strategy == "by_paragraph":
            paragraphs = self.split_by_paragraphs(text)
            chunks = []
            current_chunk = ""
            current_tokens = 0
            
            for para in paragraphs:
                para_tokens = len(self.split_by_tokens(para))
                
                if current_tokens + para_tokens <= self.chunk_size:
                    current_chunk += "\n\n" + para
                    current_tokens += para_tokens
                else:
                    if current_chunk:
                        chunks.append({
                            "text": current_chunk.strip(),
                            "token_count": current_tokens,
                            "metadata": metadata or {}
                        })
                    current_chunk = para
                    current_tokens = para_tokens
            
            if current_chunk:
                chunks.append({
                    "text": current_chunk.strip(),
                    "token_count": current_tokens,
                    "metadata": metadata or {}
                })
            
            return chunks
        
        elif strategy == "by_sentence":
            sentences = self.split_by_sentences(text)
            chunks = []
            current_chunk = ""
            current_tokens = 0
            
            for sentence in sentences:
                sentence_tokens = len(self.split_by_tokens(sentence))
                
                if current_tokens + sentence_tokens <= self.chunk_size:
                    current_chunk += " " + sentence
                    current_tokens += sentence_tokens
                else:
                    if current_chunk:
                        chunks.append({
                            "text": current_chunk.strip(),
                            "token_count": current_tokens,
                            "metadata": metadata or {}
                        })
                    current_chunk = sentence
                    current_tokens = sentence_tokens
            
            if current_chunk:
                chunks.append({
                    "text": current_chunk.strip(),
                    "token_count": current_tokens,
                    "metadata": metadata or {}
                })
            
            return chunks
        
        else:  # recursive (default)
            return self.create_chunks(text, metadata)



============================================


VÍ DỤ SỬ DỤNG


============================================

chunker = DocumentChunker(chunk_size=1000, chunk_overlap=200)

sample_text = """
DeepSeek V3 là mô hình ngôn ngữ lớn mới nhất được phát triển bởi DeepSeek AI. 
Mô hình này có nhiều cải tiến vượt bậc so với các phiên bản trước đó.

Đầu tiên, DeepSeek V3 sở hữu context window lên đến 640K tokens, cho phép xử lý 
toàn bộ tài liệu dài trong một lần gọi API. Điều này đặc biệt hữu ích cho các 
ứng dụng RAG cần xử lý tài liệu pháp lý, tài chính hoặc kỹ thuật.

Thứ hai, chi phí của DeepSeek V3 cực kỳ cạnh tranh. Với mức giá chỉ $0.42/MTok 
output qua HolySheep API, doanh nghiệp có thể tiết kiệm đến 95% chi phí so với 
việc sử dụng GPT-4.1 hoặc Claude Sonnet 4.5.

Cuối cùng, DeepSeek V3 hỗ trợ đa ngôn ngữ native, bao gồm tiếng Việt, tiếng Trung, 
tiếng Anh và nhiều ngôn ngữ khác với chất lượng đồng đều.
"""


Tạo chunks

chunks = chunker.smart_chunk(sample_text, metadata={"source": "sample_doc.txt"})

print(f"📄 Tổng số chunks: {len(chunks)}")
for i, chunk in enumerate(chunks):
    print(f"\n--- Chunk {i+1} ({chunk['token_count']} tokens) ---")
    print(chunk['text'][:200] + "..." if len(chunk['text']) > 200 else chunk['text'])

3.3 Vector Store Và Retrieval

import chromadb
from chromadb.config import Settings
from typing import List, Tuple
import uuid

class VectorRAGStore:
    """
    Vector store implementation sử dụng ChromaDB
    - Hỗ trợ local persistence
    - Semantic search với cosine similarity
    - Metadata filtering
    - Batch operations cho performance
    """
    
    def __init__(
        self, 
        persist_directory: str = "./chroma_db",
        collection_name: str = "rag_documents"
    ):
        self.client = chromadb.PersistentClient(
            path=persist_directory,
            settings=Settings(anonymized_telemetry=False)
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}  # Cosine similarity
        )
        print(f"✅ Connected to ChromaDB collection: {collection_name}")
        print(f"📊 Total documents: {self.collection.count()}")
    
    def add_documents(
        self, 
        texts: List[str], 
        embeddings: List[List[float]], 
        metadatas: List[dict] = None,
        ids: List[str] = None
    ) -> List[str]:
        """
        Thêm documents vào vector store
        
        Args:
            texts: Danh sách text chunks
            embeddings: Danh sách embedding vectors
            metadatas: Danh sách metadata dicts
            ids: Danh sách custom IDs (tự động generate nếu None)
        
        Returns:
            List of document IDs
        """
        n = len(texts)
        
        # Auto-generate IDs nếu không provided
        if ids is None:
            ids = [str(uuid.uuid4()) for _ in range(n)]
        
        # Auto-generate metadatas nếu không provided
        if metadatas is None:
            metadatas = [{} for _ in range(n)]
        
        try:
            self.collection.add(
                ids=ids,
                embeddings=embeddings,
                documents=texts,
                metadatas=metadatas
            )
            print(f"✅ Added {n} documents to collection")
            return ids
            
        except Exception as e:
            print(f"❌ Error adding documents: {e}")
            return []
    
    def retrieve(
        self, 
        query_embedding: List[float], 
        n_results: int = 5,
        where: dict = None,
        where_document: dict = None
    ) -> dict:
        """
        Semantic search để retrieve relevant documents
        
        Args:
            query_embedding: Query embedding vector
            n_results: Số lượng results muốn lấy
            where: Metadata filter (e.g., {"source": "legal_doc.pdf"})
            where_document: Document content filter
        
        Returns:
            dict với ids, distances, documents, metadatas
        """
        try:
            results = self.collection.query(
                query_embeddings=[query_embedding],
                n_results=n_results,
                where=where,
                where_document=where_document,
                include=["distances", "documents", "metadatas"]
            )
            
            # Format kết quả
            formatted = {
                "ids": results["ids"][0] if results["ids"] else [],
                "distances": results["distances"][0] if results["distances"] else [],
                "documents": results["documents"][0] if results["documents"] else [],
                "metadatas": results["metadatas"][0] if results["metadatas"] else []
            }
            
            return formatted
            
        except Exception as e:
            print(f"❌ Error retrieving: {e}")
            return {"ids": [], "distances": [], "documents": [], "metadatas": []}
    
    def retrieve_with_rerank(
        self,
        query: str,
        query_embedding: List[float],
        all_documents: List[str],
        all_embeddings: List[List[float]],
        top_k: int = 5,
        rerank_top_k: int = 10
    ) -> Tuple[List[str], List[str]]:
        """
        Retrieve với simple reranking strategy
        - Lấy nhiều hơn top_k từ vector search
        - Rerank dựa trên keyword matching và length penalty
        """
        
        # Initial retrieval
        results = self.retrieve(query_embedding, n_results=rerank_top_k)
        
        if not results["documents"]:
            return [], []
        
        # Simple reranking
        scored_docs = []
        for i, doc in enumerate(results["documents"]):
            distance = results["distances"][i]
            doc_id = results["ids"][i]
            
            # Keyword matching bonus
            query_keywords = set(query.lower().split())
            doc_keywords = set(doc.lower().split())
            overlap = len(query_keywords & doc_keywords)
            keyword_bonus = overlap * 0.05
            
            # Length penalty (prefer medium-length docs)
            length_penalty = 0.01 * abs(500 - len(doc.split())) / 500
            
            # Final score (lower distance = better, so subtract bonuses)
            final_score = distance - keyword_bonus + length_penalty
            
            scored_docs.append((final_score, doc_id, doc))
        
        # Sort by final score
        scored_docs.sort(key=lambda x: x[0])
        
        # Return top_k
        top_docs = [doc for _, _, doc in scored_docs[:top_k]]
        top_ids = [doc_id for _, doc_id, _ in scored_docs[:top_k]]
        
        return top_ids, top_docs
    
    def delete_by_ids(self, ids: List[str]) -> bool:
        """Xóa documents bằng IDs"""
        try:
            self.collection.delete(ids=ids)
            print(f"✅ Deleted {len(ids)} documents")
            return True
        except Exception as e:
            print(f"❌ Error deleting: {e}")
            return False
    
    def clear_collection(self) -> bool:
        """Xóa toàn bộ collection"""
        try:
            self.client.delete_collection(name=self.collection.name)
            self.collection = self.client.get_or_create_collection(
                name=self.collection.name,
                metadata={"hnsw:space": "cosine"}
            )
            print("✅ Cleared collection")
            return True
        except Exception as e:
            print(f"❌ Error clearing: {e}")
            return False



============================================


VÍ DỤ SỬ DỤNG COMPLETE RAG PIPELINE


============================================

from rag_client import DeepSeekRAGClient
from chunker import DocumentChunker


Initialize

api_key = "YOUR_HOLYSHEEP_API_KEY"
rag_client = DeepSeekRAGClient(api_key=api_key)
chunker = DocumentChunker(chunk_size=1000, chunk_overlap=200)
vector_store = VectorRAGStore(persist_directory="./my_rag_db")


Sample documents

documents = [
    "DeepSeek V3 có context window 640K tokens, giá chỉ $0.42/MTok.",
    "HolySheep API hỗ trợ thanh toán qua WeChat và Alipay.",
    "RAG là Retrieval Augmented Generation, kết hợp retrieval với generation.",
    "Vector database như ChromaDB lưu trữ embeddings để semantic search."
]


Process documents

all_chunks = []
all_embeddings = []
all_metadata = []

for i, doc in enumerate(documents):
    chunks = chunker.smart_chunk(doc, metadata={"doc_id": i, "source": "sample"})
    for chunk in chunks:
        all_chunks.append(chunk["text"])
        all_metadata.append(chunk["metadata"])
        
        # Create embedding
        embedding = rag_client.create_embedding(chunk["text"])
        all_embeddings.append(embedding)


Add to vector store

ids = vector_store.add_documents(all_chunks, all_embeddings, all_metadata)


Query

query = "DeepSeek V3 giá bao nhiêu?"
query_embedding = rag_client.create_embedding(query)


Retrieve

results = vector_store.retrieve(query_embedding, n_results=2)
print(f"\n🔍 Query: {query}")
print(f"📄 Top documents:")
for i, doc in enumerate(results["documents"]):
    print(f"  {i+1}. {doc[:100]}... (distance: {results['distances'][i]:.4f})")


Generate response

response = rag_client.generate_response(query, results["documents"])
print(f"\n🤖 Response:\n{response}")

3.4 Complete RAG Pipeline Production-Ready

"""
Complete RAG Pipeline với DeepSeek V3 qua HolySheep API
- Input: User query
- Output: AI-generated response based on retrieved context
- Features: Streaming, retry logic, caching, metrics
"""

import time
import hashlib
from typing import Generator, Optional
from functools import lru_cache
from openai import APIError, RateLimitError
import json

class ProductionRAGPipeline:
    """
    Production-ready RAG pipeline với:
    - Retry logic cho API calls
    - Simple caching
    - Latency metrics
    - Streaming response
    """
    
    def __init__(
        self,
        api_key: str,
        model: str = "deepseek-chat",
        embedding_model: str = "text-embedding-3-small",
        cache_ttl: int = 3600  # Cache TTL in seconds
    ):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"
        )
        self.model = model
        self.embedding_model = embedding_model
        self.cache = {}
        self.cache_ttl = cache_ttl
        self.metrics = {
            "total_requests": 0,
            "cache_hits": 0,
            "total_latency_ms": 0,
            "errors": 0
        }
    
    def _get_cache_key(self, text: str, prefix: str = "") -> str:
        """Generate cache key từ text hash"""
        content = f"{prefix}:{text}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def _is_cache_valid(self, cache_entry: dict) -> bool:
        """Check if cache entry còn valid"""
        if not cache_entry:
            return False
        return time.time() - cache_entry["timestamp"] < self.cache_ttl
    
    @lru_cache(maxsize=1000)
    def _cached_embedding(self, text: str) -> Optional[list]:
        """Cached embedding với LRU cache"""
        cache_key = self._get_cache_key(text, "embedding")
        
        if cache_key in self.cache and self._is_cache_valid(self.cache[cache_key]):
            self.metrics["cache_hits"] += 1
            return self.cache[cache_key]["data"]
        
        return None
    
    def create_embedding(self, text: str, use_cache: bool = True) -> list:
        """Tạo embedding với caching và retry"""
        
        # Check cache first
        if use_cache:
            cached = self._cached_embedding(text)
            if cached:
                return cached
        
        for attempt in range(3):
            try:
                start_time = time.time()
                response = self.client.embeddings.create(
                    model=self.embedding_model,
                    input=text
                )
                latency = (time.time() - start_time) * 1000
                
                embedding = response.data[0].embedding
                
                # Store in cache
                cache_key = self._get_cache_key(text, "embedding")
                self.cache[cache_key] = {
                    "data": embedding,
                    "timestamp": time.time()
                }
                
                self.metrics["total_latency_ms"] += latency
                return embedding
                
            except RateLimitError:
                wait_time = 2 ** attempt
                print(f"⏳ Rate limited, waiting {wait_time}s...")
                time.sleep(wait_time)
            except APIError as e:
                print(f"❌ API Error: {e}")
                self.metrics["errors"] += 1
                if attempt == 2:
                    return []
        
        return []
    
    def generate_response(
        self,
        query: str,
        context: list[str],
        system_prompt: str = None,
        temperature: float = 0.3,
        max_tokens: int = 2000,
        stream: bool = False
    ) -> str | Generator:
        """Generate response với retry và streaming support"""
        
        if system_prompt is None:
            system_prompt = """Bạn là trợ lý AI chuyên nghiệp. 
Trả lời dựa trên ngữ cảnh được cung cấp. Nếu không có thông tin, nói rõ.
Luôn trả lời bằng tiếng Việt, ngắn gọn và hữu ích."""

        context_text = "\n\n".join([
            f"[Document {i+1}]:\n{doc}" 
            for i, doc in enumerate(context)
        ])

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"""Ngữ cảnh:
{context_text}

Câu hỏi: {query}

Hãy trả lời dựa trên ngữ cảnh trên."""}
        ]
        
        for attempt in range(3):
            try:
                start_time = time.time()
                self.metrics["total_requests"] += 1
                
                if stream:
                    return self._stream_response(
                        messages, temperature, max_tokens
                    )
                else:
                    response = self.client.chat.completions.create(
                        model=self.model,
                        messages=messages,
                        temperature=temperature,
                        max_tokens=max_tokens
                    )
                    
                    latency = (time.time() - start_time) * 1000
                    self.metrics["total_latency_ms"] += latency
                    
                    return response.choices[0].message.content
                    
            except RateLimitError:
                wait_time = 2 ** attempt
                print(f"⏳ Rate limited, waiting {wait_time}s...")
                time.sleep(wait_time)
            except APIError as e:
                print(f"❌ API Error: {e}")
                self.metrics["errors"] += 1
                if attempt == 2:
                    return "Xin lỗi, đã xảy ra lỗi khi xử lý yêu cầu."
        
        return "Xin lỗi, không thể xử lý yêu cầu sau nhiều lần thử."
    
    def _stream_response(self, messages: list, temperature: float, max_tokens: int):
        """Streaming response generator"""
        try:
            stream = self.client.chat.completions.create

Tài nguyên liên quan
Bài viết liên quan