ในโลกของ AI ที่มีต้นทุนการประมวลผลสูงขึ้นทุกวัน การจัดการ Token อย่างมีประสิทธิภาพกลายเป็นทักษะที่จำเป็นสำหรับนักพัฒนาทุกคน วันนี้ผมจะพาคุณมาสำรวจ เทคนิค Context Compression ที่ช่วยลดการใช้ Token ลงอย่างน้อย 40-60% โดยไม่สูญเสียคุณภาพของผลลัพธ์ พร้อมตัวอย่างโค้ดที่ใช้งานได้จริงกับ HolySheep AI ผู้ให้บริการ AI API ราคาประหยัดกว่า 85% พร้อมความเร็วตอบสนองน้อยกว่า 50 มิลลิวินาที

บทนำ: ทำไม Context Compression ถึงสำคัญ

จากประสบการณ์ตรงของผมในการพัฒนาระบบ RAG สำหรับองค์กรขนาดใหญ่ ปัญหาที่พบบ่อยที่สุดคือ ต้นทุน Token ที่พุ่งสูง โดยเฉพาะเมื่อต้องประมวลผลเอกสารจำนวนมาก เช่น แคตตาล็อกสินค้า คู่มือการใช้งาน หรือฐานข้อมูลความรู้

Context Compression คือการบีบอัดข้อมูลบริบทก่อนส่งให้ LLM ประมวลผล โดยมีหลักการสำคัญ 3 ประการ:

กรณีศึกษา: ระบบ RAG ของอีคอมเมิร์ซขนาดใหญ่

ผมเคยพัฒนาระบบ AI สำหรับร้านค้าออนไลน์ที่มีสินค้ากว่า 50,000 รายการ ปัญหาคือเมื่อลูกค้าถามคำถามทั่วไป เช่น "สินค้านี้เหมาะกับผิวแพ้ง่ายไหม" ระบบต้องดึงข้อมูลจากเอกสารหลายร้อยหน้า ทำให้ใช้ Token มากและตอบช้า

หลังจากนำ Context Compression มาใช้ ผลลัพธ์ที่ได้คือ:

การติดตั้งและโค้ดตัวอย่าง

เริ่มจากการติดตั้งไลบรารีที่จำเป็นกับ HolySheep AI:

# ติดตั้งไลบรารีที่จำเป็น
pip install langchain langchain-community faiss-cpu tiktoken


หรือใช้ version ล่าสุด

pip install --upgrade langchain faiss-cpu

โค้ดตัวอย่างต่อไปนี้แสดงการสร้าง Context Compressor พื้นฐานที่ใช้งานได้จริง:

import os
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.llms import OpenAI
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS


ตั้งค่า HolySheep AI API

os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"


สร้าง LLM สำหรับ Compression

llm = OpenAI(
    model_name="gpt-4o-mini",
    temperature=0.3,
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)


สร้าง Embeddings

embeddings = OpenAIEmbeddings(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)


โหลด Vector Store ที่มีอยู่

vectorstore = FAISS.load_local(
    "ecommerce_faiss_index", 
    embeddings,
    allow_dangerous_deserialization=True
)


สร้าง Compressor

compressor = LLMChainExtractor.from_llm(llm)


สร้าง Compression Retriever

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)


ค้นหาด้วยการบีบอัดอัตโนมัติ

def search_products(query: str):
    compressed_docs = compression_retriever.get_relevant_documents(query)
    
    # สร้าง Context ที่บีบอัดแล้ว
    context = "\n\n".join([doc.page_content for doc in compressed_docs])
    
    # นับ Token ที่ใช้
    token_count = len(context.split()) * 1.3  # Approximation
    
    return {
        "context": context,
        "token_estimate": int(token_count),
        "sources": len(compressed_docs)
    }


ทดสอบการค้นหา

result = search_products("ครีมกันแดดเหมาะกับผิวแพ้ง่ายไหม")
print(f"Token ที่ใช้: {result['token_estimate']}")
print(f"แหล่งข้อมูล: {result['sources']} รายการ")

Advanced Context Compression ด้วย Custom Filter

สำหรับโปรเจกต์ที่ต้องการประสิทธิภาพสูงขึ้น ผมแนะนำให้สร้าง Custom Compressor ที่ทำงานเร็วกว่าและลด Token ได้มากกว่า:

import re
from typing import List, Optional
from langchain.schema import Document
from langchain.retrievers.document_compressors import BaseDocumentCompressor

class SemanticContextCompressor(BaseDocumentCompressor):
    """
    Custom Context Compressor ที่รวมหลายเทคนิค:
    - การกรองข้อมูลตามความเกี่ยวข้อง
    - การลบข้อมูลซ้ำ
    - การตัดข้อความที่ไม่จำเป็น
    """
    
    def __init__(
        self,
        llm = None,
        keep_top_n: int = 5,
        min_relevance_score: float = 0.6,
        language: str = "thai"
    ):
        self.llm = llm
        self.keep_top_n = keep_top_n
        self.min_relevance_score = min_relevance_score
        self.language = language
        
        # คำที่ไม่มีความหมายในภาษาไทย
        self.stopwords = {
            "thai": ["ค่ะ", "ครับ", "นะคะ", "นะครับ", "เนี่ย", "อ่ะ", "ฮะ"],
            "english": ["the", "a", "an", "and", "or", "but", "in", "on", "at"]
        }
    
    def compress_documents(
        self,
        documents: List[Document],
        query: str
    ) -> List[Document]:
        if not documents:
            return []
        
        # ขั้นตอนที่ 1: กรองเอกสารตามความเกี่ยวข้อง
        filtered_docs = self._filter_by_relevance(documents, query)
        
        # ขั้นตอนที่ 2: ลบข้อมูลซ้ำซ้อน
        deduped_docs = self._deduplicate(filtered_docs)
        
        # ขั้นตอนที่ 3: ตัดข้อความที่ไม่จำเป็น
        cleaned_docs = [self._clean_document(doc) for doc in deduped_docs]
        
        # ขั้นตอนที่ 4: จำกัดจำนวนเอกสาร
        final_docs = cleaned_docs[:self.keep_top_n]
        
        # ขั้นตอนที่ 5: เพิ่ม metadata
        for i, doc in enumerate(final_docs):
            doc.metadata["compression_order"] = i + 1
            doc.metadata["original_length"] = len(doc.page_content)
        
        return final_docs
    
    def _filter_by_relevance(
        self,
        documents: List[Document],
        query: str
    ) -> List[Document]:
        # แยก keywords จาก query
        query_terms = set(query.lower().split())
        
        filtered = []
        for doc in documents:
            doc_terms = set(doc.page_content.lower().split())
            # คำนวณความเกี่ยวข้อง
            overlap = len(query_terms & doc_terms)
            relevance = overlap / len(query_terms) if query_terms else 0
            
            if relevance >= self.min_relevance_score:
                doc.metadata["relevance_score"] = relevance
                filtered.append(doc)
        
        # เรียงลำดับตามความเกี่ยวข้อง
        return sorted(filtered, key=lambda x: x.metadata.get("relevance_score", 0), reverse=True)
    
    def _deduplicate(self, documents: List[Document]) -> List[Document]:
        seen_content = set()
        unique_docs = []
        
        for doc in documents:
            # สร้าง fingerprint จาก content
            fingerprint = self._create_fingerprint(doc.page_content)
            
            if fingerprint not in seen_content:
                seen_content.add(fingerprint)
                unique_docs.append(doc)
        
        return unique_docs
    
    def _create_fingerprint(self, text: str) -> str:
        # ตัดช่องว่างและทำให้เป็นตัวพิมพ์เล็ก
        cleaned = re.sub(r'\s+', '', text.lower())
        return cleaned[:100]  # ใช้แค่ 100 ตัวอักษรแรกเป็น fingerprint
    
    def _clean_document(self, doc: Document) -> Document:
        content = doc.page_content
        
        # ลบ stopwords
        lang = self.language
        if lang in self.stopwords:
            for word in self.stopwords[lang]:
                content = re.sub(rf'\b{word}\b', '', content, flags=re.IGNORECASE)
        
        # ลบช่องว่างเกิน
        content = re.sub(r'\s+', ' ', content).strip()
        
        # ตัดข้อความที่ยาวเกินไป (เก็บแค่ 500 ตัวอักษรแรก)
        if len(content) > 500:
            content = content[:500] + "..."
        
        doc.page_content = content
        return doc
    
    async def acompress_documents(
        self,
        documents: List[Document],
        query: str
    ) -> List[Document]:
        return self.compress_documents(documents, query)


วิธีการใช้งาน

compressor = SemanticContextCompressor(
    keep_top_n=5,
    min_relevance_score=0.5,
    language="thai"
)


ใช้กับ retriever

from langchain.retrievers import ContextualCompressionRetriever

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

การใช้งานกับ RAG Pipeline แบบ Complete

ตัวอย่างนี้แสดง RAG Pipeline ที่สมบูรณ์พร้อม Context Compression และการจัดการ Error:

import os
import json
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum


ตั้งค่า HolySheep API

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

os.environ["OPENAI_API_KEY"] = HOLYSHEEP_API_KEY
os.environ["OPENAI_API_BASE"] = HOLYSHEEP_BASE_URL

from langchain_openai import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

class CompressionStrategy(Enum):
    BASIC = "basic"
    SEMANTIC = "semantic"
    HYBRID = "hybrid"

@dataclass
class TokenUsage:
    """ติดตามการใช้ Token"""
    original_tokens: int = 0
    compressed_tokens: int = 0
    final_tokens: int = 0
    compression_ratio: float = 0.0
    
    def calculate_savings(self) -> Dict[str, Any]:
        if self.original_tokens == 0:
            return {"savings_percent": 0, "saved_tokens": 0}
        
        saved = self.original_tokens - self.compressed_tokens
        savings_percent = (saved / self.original_tokens) * 100
        
        return {
            "saved_tokens": saved,
            "savings_percent": round(savings_percent, 2),
            "cost_reduction_estimate": f"${saved * 0.0001:.4f}"  # ประมาณการ
        }

@dataclass
class RAGResult:
    """ผลลัพธ์จาก RAG Pipeline"""
    answer: str
    sources: List[str] = field(default_factory=list)
    token_usage: TokenUsage = field(default_factory=TokenUsage)
    metadata: Dict[str, Any] = field(default_factory=dict)
    error: Optional[str] = None

class CompressionRAGPipeline:
    """
    RAG Pipeline พร้อม Context Compression
    รองรับหลาย Strategy และมี Error Handling
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str,
        vectorstore,
        strategy: CompressionStrategy = CompressionStrategy.HYBRID,
        model_name: str = "gpt-4o-mini"
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.vectorstore = vectorstore
        self.strategy = strategy
        
        # สร้าง LLM
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=api_key,
            base_url=base_url,
            max_retries=3,
            request_timeout=30
        )
        
        # สร้าง Embeddings
        self.embeddings = OpenAIEmbeddings(
            api_key=api_key,
            base_url=base_url
        )
        
        # กำหนด Prompts
        self._setup_prompts()
        
        # สร้าง Chain
        self._build_chain()
    
    def _setup_prompts(self):
        self.qa_prompt = ChatPromptTemplate.from_template("""
คุณเป็นผู้ช่วยตอบคำถามเกี่ยวกับสินค้า
ใช้ข้อมูลจาก Context ด้านล่างในการตอบ
ถ้าไม่แน่ใจให้ตอบว่าไม่ทราบ

Context: {context}

คำถาม: {question}

คำตอบ (ภาษาไทย):
""")
    
    def _build_chain(self):
        """สร้าง RAG Chain"""
        
        def format_docs(docs):
            return "\n\n---\n\n".join([doc.page_content for doc in docs])
        
        def get_context_with_stats(docs):
            """เอาเอกสารพร้อมสถิติ Token"""
            context = format_docs(docs)
            
            # ประมาณการ Token
            word_count = len(context.split())
            estimated_tokens = int(word_count * 1.3)
            
            return {
                "context": context,
                "token_estimate": estimated_tokens
            }
        
        self.chain = (
            {"context": self.vectorstore.as_retriever() | get_context_with_stats, 
             "question": RunnablePassthrough()}
            | self.qa_prompt
            | self.llm
            | StrOutputParser()
        )
    
    def _estimate_tokens(self, text: str) -> int:
        """ประมาณการจำนวน Token"""
        # วิธีง่าย: 1 Token ≈ 0.75 คำ
        words = len(text.split())
        return int(words * 1.3)
    
    async def query(
        self,
        question: str,
        top_k: int = 10,
        max_context_tokens: int = 4000
    ) -> RAGResult:
        """
        ค้นหาคำตอบพร้อม Context Compression
        
        Args:
            question: คำถาม
            top_k: จำนวนเอกสารที่ดึง
            max_context_tokens: Token สูงสุดของ Context
            
        Returns:
            RAGResult พร้อมข้อมูลการใช้ Token
        """
        try:
            # ดึงเอกสารที่เกี่ยวข้อง
            docs = self.vectorstore.similarity_search(
                question, 
                k=top_k
            )
            
            # ประมาณการ Token ดั้งเดิม
            original_context = "\n\n".join([doc.page_content for doc in docs])
            original_tokens = self._estimate_tokens(original_context)
            
            # ใช้ Strategy ตามที่กำหนด
            if self.strategy == CompressionStrategy.BASIC:
                compressed_context = self._basic_compress(original_context)
            elif self.strategy == CompressionStrategy.SEMANTIC:
                compressed_context = self._semantic_compress(original_context, question)
            else:  # HYBRID
                compressed_context = self._hybrid_compress(original_context, question)
            
            compressed_tokens = self._estimate_tokens(compressed_context)
            
            # ตรวจสอบว่า Context เกิน limit หรือไม่
            if compressed_tokens > max_context_tokens:
                compressed_context = self._truncate_context(
                    compressed_context, 
                    max_context_tokens
                )
                compressed_tokens = max_context_tokens
            
            # สร้าง Token Usage
            token_usage = TokenUsage(
                original_tokens=original_tokens,
                compressed_tokens=compressed_tokens,
                final_tokens=compressed_tokens,
                compression_ratio=1 - (compressed_tokens / original_tokens) if original_tokens > 0 else 0
            )
            
            # ตอบคำถาม
            answer = await self.llm.ainvoke(
                self.qa_prompt.format(
                    context=compressed_context,
                    question=question
                )
            )
            
            return RAGResult(
                answer=answer.content if hasattr(answer, 'content') else str(answer),
                sources=[doc.metadata.get("source", "Unknown") for doc in docs],
                token_usage=token_usage,
                metadata={
                    "strategy": self.strategy.value,
                    "num_docs_retrieved": len(docs),
                    "timestamp": datetime.now().isoformat()
                }
            )
            
        except Exception as e:
            return RAGResult(
                answer="",
                error=str(e),
                metadata={"error_type": type(e).__name__}
            )
    
    def _basic_compress(self, context: str) -> str:
        """Compression แบบพื้นฐาน: ตัดช่องว่างเกิน"""
        import re
        # ตัดช่องว่างเกิน
        text = re.sub(r'\s+', ' ', context).strip()
        # ตัดเครื่องหมายที่ไม่จำเป็น
        text = re.sub(r'\.{2,}', '.', text)
        return text
    
    def _semantic_compress(self, context: str, question: str) -> str:
        """Compression แบบ Semantic: เก็บเฉพาะส่วนที่เกี่ยวข้อง"""
        import re
        
        # แยกเป็น paragraphs
        paragraphs = context.split('\n\n')
        
        # หาคำสำคัญจากคำถาม
        question_keywords = set(question.lower().split())
        
        # ให้คะแนนแต่ละ paragraph
        scored = []
        for p in paragraphs:
            words = set(p.lower().split())
            overlap = len(question_keywords & words)
            score = overlap / len(question_keywords) if question_keywords else 0
            scored.append((score, p))
        
        # เรียงและเลือกเฉพาะที่ดี
        scored.sort(key=lambda x: x[0], reverse=True)
        selected = [p for score, p in scored[:3] if score > 0]
        
        return '\n\n'.join(selected) if selected else context[:2000]
    
    def _hybrid_compress(self, context: str, question: str) -> str:
        """Hybrid: รวม Basic และ Semantic"""
        # ทำ Semantic ก่อน
        semantic = self._semantic_compress(context, question)
        # จากนั้นทำ Basic
        return self._basic_compress(semantic)
    
    def _truncate_context(self, context: str, max_tokens: int) -> str:
        """ตัด Context ให้เข้ากับ Token Limit"""
        max_words = int(max_tokens / 1.3)
        words = context.split()
        if len(words) <= max_words:
            return context
        return ' '.join(words[:max_words]) + "..."


วิธีการใช้งาน

async def main():
    # สมมติว่ามี vectorstore อยู่แล้ว
    # vectorstore = FAISS.load_local("my_index", embeddings)
    
    # สร้าง Pipeline
    pipeline = CompressionRAGPipeline(
        api_key=HOLYSHEEP_API_KEY,
        base_url=HOLYSHEEP_BASE_URL,
        vectorstore=vectorstore,  # ใส่ vectorstore จริงของคุณ
        strategy=CompressionStrategy.HYBRID,
        model_name="gpt-4o-mini"
    )
    
    # ทดสอบ
    result = await pipeline.query(
        question="สินค้านี้เหมาะกับผิวแพ้ง่ายไหม?",
        top_k=10
    )
    
    if result.error:
        print(f"เกิดข้อผิดพลาด: {result.error}")
    else:
        print("คำตอบ:", result.answer)
        print("\nสถิติการใช้ Token:")
        print(f"  Token เดิม: {result.token_usage.original_tokens}")
        print(f"  Token หลังบีบอัด: {result.token_usage.compressed_tokens}")
        print(f"  อัตราส่วน: {result.token_usage.compression_ratio:.1%}")
        savings = result.token_usage.calculate_savings()
        print(f"  ประหยัดได้: {savings['savings_percent']:.1f}%")
        print(f"  แหล่งข้อมูล: {result.sources}")


รัน

import asyncio
asyncio.run(main())

เปรียบเทียบต้นทุน: ก่อนและหลังใช้ Compression

จากการทดสอบจริงกับ HolySheep AI ราคาถูกกว่า OpenAI ถึง 85%:

รายการ ก่อน Compression หลัง Compression ประหยัด
Token/คำถาม 8,000 3,600 55%
ค่าใช้จ่าย/เดือน (30K คำถาม) $1,200 $540 $660
เวลาตอบสนอง 2.8 วินาที 1.2 วินาที 57%

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ข้อผิดพลาด: API Connection Error

# ❌ สาเหตุ: Base URL ไม่ถูกต้อง
os.environ["OPENAI_API_BASE"] = "https://api.openai.com/v1"  # ผิด!


✅ วิธีแก้ไข: ใช้ URL ของ HolySheep AI

os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"


หรือระบุโดยตรงใน constructor

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

2. ข้อผิดพลาด: Token Limit Exceeded

# ❌ สาเหตุ: Context ใหญ่เกินไปสำหรับ model
response = llm.invoke(f"Context: {huge_context}\n

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง