Semantic Caching: Cách Tiết Kiệm 85%+ Chi Phí LLM Bằng Việc Tái Sử Dụng Câu Trả Lời

Bạn có biết rằng 70% câu hỏi gửi đến chatbot của mình là những biến thể của cùng một vấn đề? "Giá sản phẩm bao nhiêu?", "Cho hỏi giá", "Sản phẩm này bao tiền?" - tất cả đều cần một câu trả lời giống nhau, nhưng mô hình LLM phải xử lý lại từ đầu. Semantic Caching chính là giải pháp cho bài toán này.

So Sánh Chi Phí: HolySheep vs API Chính Thức vs Dịch Vụ Relay

Semantic Caching Là Gì?

Semantic Caching là kỹ thuật lưu trữ câu trả lời của LLM theo nghĩa ngữ nghĩa, không phải theo từng chữ. Khi người dùng hỏi một câu hỏi mới, hệ thống sẽ:

• Tính toán embedding vector của câu hỏi mới
• So sánh với các câu hỏi đã lưu trong cache
• Nếu độ tương đồng (similarity) > ngưỡng, trả về câu trả lời đã lưu
• Nếu không, gọi LLM và lưu kết quả vào cache

Cài Đặt Semantic Cache Với HolySheep AI

Dưới đây là ví dụ triển khai Semantic Caching sử dụng HolySheep AI với tỷ giá ¥1=$1, giúp bạn tiết kiệm tối đa chi phí.

Cài Đặt Thư Viện

pip install requests numpy scikit-learn redis openai

Triển Khai Semantic Cache Server

import os
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
import requests
from collections import OrderedDict
import hashlib
import time

class SemanticCache:
    def __init__(self, similarity_threshold=0.85, max_size=10000):
        self.similarity_threshold = similarity_threshold
        self.max_size = max_size
        self.cache_store = OrderedDict()
        self.cache_vectors = {}
        self.cache_hits = 0
        self.cache_misses = 0
        
        # HolySheep AI Configuration
        self.holysheep_api_key = os.getenv("YOUR_HOLYSHEEP_API_KEY")
        self.holysheep_base_url = "https://api.holysheep.ai/v1"
        
    def _get_embedding(self, text):
        """Lấy embedding vector từ HolySheep AI sử dụng model embedding"""
        url = f"{self.holysheep_base_url}/embeddings"
        headers = {
            "Authorization": f"Bearer {self.holysheep_api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "text-embedding-3-small",
            "input": text
        }
        
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        return np.array(data["data"][0]["embedding"])
    
    def _get_cache_key(self, text):
        """Tạo cache key duy nhất cho text"""
        return hashlib.md5(text.encode()).hexdigest()
    
    def _find_similar(self, query_vector):
        """Tìm câu trả lời có độ tương đồng cao nhất"""
        if not self.cache_vectors:
            return None, 0
        
        cached_vectors = np.array(list(self.cache_vectors.values()))
        similarities = cosine_similarity([query_vector], cached_vectors)[0]
        
        max_idx = np.argmax(similarities)
        max_similarity = similarities[max_idx]
        
        if max_similarity >= self.similarity_threshold:
            cache_keys = list(self.cache_vectors.keys())
            return cache_keys[max_idx], max_similarity
        
        return None, max_similarity
    
    def get_or_generate(self, prompt, model="gpt-4.1"):
        """Lấy từ cache hoặc gọi LLM qua HolySheep AI"""
        start_time = time.time()
        
        # Bước 1: Tính embedding của câu hỏi
        query_vector = self._get_embedding(prompt)
        
        # Bước 2: Tìm câu trả lời tương tự trong cache
        cache_key, similarity = self._find_similar(query_vector)
        
        if cache_key and cache_key in self.cache_store:
            # Cache HIT - trả về câu trả lời đã lưu
            self.cache_hits += 1
            cached_item = self.cache_store[cache_key]
            
            # Di chuyển item lên đầu (LRU)
            self.cache_store.move_to_end(cache_key)
            
            elapsed = time.time() - start_time
            return {
                "response": cached_item["response"],
                "cached": True,
                "similarity": float(similarity),
                "latency_ms": elapsed * 1000
            }
        
        # Bước 3: Cache MISS - gọi HolySheep AI
        self.cache_misses += 1
        
        url = f"{self.holysheep_base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.holysheep_api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}]
        }
        
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        llm_response = data["choices"][0]["message"]["content"]
        
        # Bước 4: Lưu vào cache
        new_cache_key = self._get_cache_key(prompt)
        
        # Kiểm tra kích thước cache, xóa item cũ nếu cần
        if len(self.cache_store) >= self.max_size:
            oldest_key = next(iter(self.cache_store))
            del self.cache_store[oldest_key]
            del self.cache_vectors[oldest_key]
        
        # Lưu cache mới
        self.cache_store[new_cache_key] = {
            "response": llm_response,
            "prompt": prompt,
            "timestamp": time.time()
        }
        self.cache_vectors[new_cache_key] = query_vector.tolist()
        
        elapsed = time.time() - start_time
        return {
            "response": llm_response,
            "cached": False,
            "similarity": 0,
            "latency_ms": elapsed * 1000
        }
    
    def get_stats(self):
        """Lấy thống kê cache"""
        total = self.cache_hits + self.cache_misses
        hit_rate = (self.cache_hits / total * 100) if total > 0 else 0
        return {
            "cache_size": len(self.cache_store),
            "hits": self.cache_hits,
            "misses": self.cache_misses,
            "hit_rate": f"{hit_rate:.2f}%"
        }


============== SỬ DỤNG ==============
if __name__ == "__main__":
    cache = SemanticCache(similarity_threshold=0.90)
    
    # Câu hỏi gốc
    result1 = cache.get_or_generate(
        "Cách làm bánh mì bơ tỏi?",
        model="gpt-4.1"
    )
    print(f"Câu hỏi 1: {'(CACHE)' if result1['cached'] else '(LLM)'} {result1['latency_ms']:.0f}ms")
    print(result1['response'][:100] + "...")
    print()
    
    # Câu hỏi tương tự - sẽ được cache
    result2 = cache.get_or_generate(
        "Công thức làm bánh mì bơ tỏi như thế nào?",
        model="gpt-4.1"
    )
    print(f"Câu hỏi 2: {'(CACHE)' if result2['cached'] else '(LLM)'} {result2['latency_ms']:.0f}ms")
    print(f"Độ tương đồng: {result2['similarity']:.2%}")
    print()
    
    # Thống kê
    print("Thống kê cache:", cache.get_stats())

Triển Khai Với Redis Để Scale

import os
import json
import numpy as np
import redis
import requests
from sklearn.metrics.pairwise import cosine_similarity

class DistributedSemanticCache:
    """Semantic Cache sử dụng Redis cho hệ thống phân tán"""
    
    def __init__(self, similarity_threshold=0.85):
        self.similarity_threshold = similarity_threshold
        
        # Kết nối Redis
        self.redis_client = redis.Redis(
            host=os.getenv("REDIS_HOST", "localhost"),
            port=int(os.getenv("REDIS_PORT", 6379)),
            decode_responses=True
        )
        
        # HolySheep AI Configuration
        self.api_key = os.getenv("YOUR_HOLYSHEEP_API_KEY")
        self.base_url = "https://api.holysheep.ai/v1"
        
    def _get_embedding(self, text):
        """Lấy embedding từ HolySheep AI"""
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={"model": "text-embedding-3-small", "input": text}
        )
        response.raise_for_status()
        return np.array(response.json()["data"][0]["embedding"])
    
    def _call_llm(self, prompt, model="gpt-4.1"):
        """Gọi LLM qua HolySheep AI"""
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}]
            }
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]
    
    def _vector_to_string(self, vector):
        """Chuyển vector thành string để lưu trong Redis"""
        return json.dumps(vector.tolist())
    
    def _string_to_vector(self, string):
        """Chuyển string thành vector"""
        return np.array(json.loads(string))
    
    def query(self, prompt, model="gpt-4.1"):
        """
        Query với Semantic Cache
        Trả về: {"response": str, "cached": bool, "similarity": float}
        """
        # Lấy embedding của prompt
        query_vector = self._get_embedding(prompt)
        
        # Tìm kiếm trong Redis sử dụng vector similarity
        candidates = self.redis_client.zrevrange(
            "semantic_cache:vectors",
            0,
            99,  # Top 100 candidates
            withscores=True
        )
        
        best_match = None
        best_similarity = 0
        
        for cache_key, _ in candidates:
            cached_vector_str = self.redis_client.get(f"semantic_cache:vec:{cache_key}")
            if cached_vector_str:
                cached_vector = self._string_to_vector(cached_vector_str)
                similarity = cosine_similarity(
                    [query_vector], [cached_vector]
                )[0][0]
                
                if similarity > best_similarity:
                    best_similarity = similarity
                    best_match = cache_key
        
        # Cache HIT
        if best_match and best_similarity >= self.similarity_threshold:
            cached_response = self.redis_client.get(f"semantic_cache:resp:{best_match}")
            
            # Cập nhật access time (LRU)
            self.redis_client.zadd(
                "semantic_cache:access",
                {best_match: self.redis_client.time()[0]}
            )
            
            return {
                "response": cached_response,
                "cached": True,
                "similarity": float(best_similarity)
            }
        
        # Cache MISS - gọi L
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
vi retrieval augmented generationrag2026nianzuijiashi 2026 0