จากประสบการณ์การสร้าง production AI system มากกว่า 50 โปรเจกต์ในปีที่ผ่านมา ผมเห็นว่า LangChain วิวัฒนาการไปไกลมากจาก framework ทดลองเล่นสู่เครื่องมือที่พร้อมสำหรับ enterprise scale บทความนี้จะเจาะลึกสถาปัตยกรรมที่แท้จริง การ optimize performance และ cost efficiency รวมถึง production-ready patterns ที่ใช้งานได้จริง

LangChain Architecture Deep Dive

LangChain v0.1 ขึ้นมาปัจจุบันมีการเปลี่ยนแปลงครั้งใหญ่เรื่อง component abstraction โดยเฉพาะ LCEL (LangChain Expression Language) ที่ทำให้การ compose chain มีความยืดหยุ่นมากขึ้น สถาปัตยกรรมหลักประกอบด้วย:

ใน production environment สิ่งที่สำคัญที่สุดคือการแยก concerns อย่างชัดเจน ผมแนะนำให้ใช้ modular approach ตามนี้:

# Project Structure for Production LangChain App
"""
langchain_production/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI entry point
│   ├── config.py            # Environment & settings
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── chat.py          # Chat endpoints
│   │   └── embeddings.py    # Embedding endpoints
│   ├── services/
│   │   ├── __init__.py
│   │   ├── llm_service.py   # LLM abstraction layer
│   │   ├── retriever.py     # RAG retrieval logic
│   │   └── chain_builder.py # Chain composition
│   ├── models/
│   │   ├── __init__.py
│   │   ├── schemas.py       # Pydantic models
│   │   └── requests.py      # Request/Response models
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── logger.py        # Structured logging
│   │   └── metrics.py      # Prometheus metrics
│   └── dependencies.py      # Dependency injection
├── prompts/
│   ├── system_prompts/      # System prompt templates
│   └── few_shot_examples/   # Few-shot learning examples
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
└── docker-compose.yml       # Local development
"""


Config Management with Pydantic Settings

from pydantic_settings import BaseSettings
from typing import Literal
from functools import lru_cache

class Settings(BaseSettings):
    """Production configuration with type safety"""
    
    # API Settings
    api_host: str = "0.0.0.0"
    api_port: int = 8000
    debug: bool = False
    
    # LLM Provider Settings (HolySheep AI)
    llm_provider: Literal["holysheep", "openai", "anthropic"] = "holysheep"
    llm_model: str = "gpt-4.1"
    llm_temperature: float = 0.7
    llm_max_tokens: int = 4096
    
    # HolySheep API Configuration
    # base_url ต้องเป็น https://api.holysheep.ai/v1 เท่านั้น
    holysheep_api_base: str = "https://api.holysheep.ai/v1"
    holysheep_api_key: str = ""  # Load from env
    
    # Embedding Settings
    embedding_model: str = "text-embedding-3-small"
    embedding_batch_size: int = 100
    
    # Vector Store Settings
    vector_store_provider: Literal["chroma", "pinecone", "qdrant"] = "chroma"
    vector_store_path: str = "./data/chroma"
    
    # RAG Settings
    retrieval_k: int = 5
    retrieval_score_threshold: float = 0.7
    rerank_enabled: bool = True
    
    # Performance Settings
    request_timeout: int = 120
    max_concurrent_requests: int = 100
    cache_enabled: bool = True
    cache_ttl: int = 3600
    
    # Cost Optimization
    enable_budget_alerts: bool = True
    monthly_budget_limit: float = 500.0
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        case_sensitive = False

@lru_cache()
def get_settings() -> Settings:
    return Settings()


Environment variables (.env)

"""
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
LLM_MODEL=gpt-4.1
EMBEDDING_MODEL=text-embedding-3-small
DEBUG=false
"""

LLM Service Abstraction สำหรับ HolySheep AI

ในการพัฒนา production system สิ่งที่จำเป็นมากคือ abstraction layer สำหรับ LLM provider เพื่อให้สามารถ switch provider ได้ง่ายและเป็นอิสระ ผมใช้ HolySheep AI สำหรับโปรเจกต์ส่วนใหญ่เพราะราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI โดยอัตราแลกเปลี่ยน ¥1=$1 ทำให้ค่าใช้จ่ายลดลงอย่างมาก ความหน่วงต่ำกว่า 50ms ทำให้ response time เร็วมาก

# LLM Service Abstraction Layer with HolySheep AI
from abc import ABC, abstractmethod
from typing import AsyncIterator, Optional, List, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime
import asyncio
import logging

import httpx
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain.callbacks.base import AsyncCallbackHandler

logger = logging.getLogger(__name__)

@dataclass
class LLMUsageStats:
    """Track token usage for cost optimization"""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    cost_usd: float = 0.0
    latency_ms: float = 0.0
    provider: str = ""
    model: str = ""
    timestamp: datetime = field(default_factory=datetime.utcnow)

@dataclass
class LLMResponse:
    """Standardized LLM response"""
    content: str
    usage: LLMUsageStats
    model: str
    finish_reason: str
    metadata: Dict[str, Any] = field(default_factory=dict)

class BaseLLMService(ABC):
    """Abstract LLM service interface"""
    
    @abstractmethod
    async def invoke(
        self,
        messages: List[BaseMessage],
        **kwargs
    ) -> LLMResponse:
        pass
    
    @abstractmethod
    async def stream(
        self,
        messages: List[BaseMessage],
        **kwargs
    ) -> AsyncIterator[str]:
        pass
    
    @abstractmethod
    async def get_embeddings(self, texts: List[str]) -> List[List[float]]:
        pass

class HolySheepLLMService(BaseLLMService):
    """
    HolySheep AI LLM Service Implementation
    ราคา: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok, 
          Gemini 2.5 Flash $2.50/MTok, DeepSeek V3.2 $0.42/MTok
    ความหน่วง: <50ms
    """
    
    # Pricing per million tokens (USD)
    PRICING = {
        "gpt-4.1": {"input": 8.0, "output": 8.0},
        "gpt-4o": {"input": 5.0, "output": 15.0},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
        "gemini-2.5-flash": {"input": 2.50, "output": 2.50},
        "deepseek-v3.2": {"input": 0.42, "output": 0.42},
    }
    
    def __init__(
        self,
        api_key: str,
        model: str = "gpt-4.1",
        temperature: float = 0.7,
        max_tokens: int = 4096,
        timeout: int = 120,
        enable_caching: bool = True,
    ):
        # ต้องใช้ base_url = https://api.holysheep.ai/v1 เท่านั้น
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.model = model
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.timeout = timeout
        self.enable_caching = enable_caching
        
        # Initialize LangChain ChatOpenAI with HolySheep
        self._llm = ChatOpenAI(
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            streaming=True,
            api_key=api_key,
            base_url=self.base_url,  # HolySheep API endpoint
            timeout=timeout,
            default_headers={
                "HTTP-Referer": "https://www.holysheep.ai",
                "X-Title": "Production AI Application"
            }
        )
        
        # Initialize embeddings
        self._embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            api_key=api_key,
            base_url=f"{self.base_url}/embeddings"
        )
        
        # Usage tracking
        self.total_usage = LLMUsageStats(provider="holysheep", model=model)
    
    async def invoke(
        self,
        messages: List[BaseMessage],
        **kwargs
    ) -> LLMResponse:
        """Non-streaming LLM invocation with full metrics"""
        start_time = datetime.utcnow()
        
        try:
            response = await self._llm.ainvoke(messages)
            latency_ms = (datetime.utcnow() - start_time).total_seconds() * 1000
            
            # Calculate cost based on pricing
            pricing = self.PRICING.get(self.model, {"input": 8.0, "output": 8.0})
            # Approximate token calculation (LangChain returns usage in response)
            estimated_input_tokens = sum(len(str(m.content)) // 4 for m in messages)
            estimated_output_tokens = len(str(response.content)) // 4
            cost = (estimated_input_tokens * pricing["input"] + 
                   estimated_output_tokens * pricing["output"]) / 1_000_000
            
            usage = LLMUsageStats(
                prompt_tokens=estimated_input_tokens,
                completion_tokens=estimated_output_tokens,
                total_tokens=estimated_input_tokens + estimated_output_tokens,
                cost_usd=cost,
                latency_ms=latency_ms,
                provider="holysheep",
                model=self.model
            )
            
            # Update tracking
            self._update_total_usage(usage)
            logger.info(
                f"LLM Response | Model: {self.model} | "
                f"Latency: {latency_ms:.2f}ms | Cost: ${cost:.6f}"
            )
            
            return LLMResponse(
                content=response.content,
                usage=usage,
                model=self.model,
                finish_reason="completed",
                metadata={"raw_response": response}
            )
            
        except Exception as e:
            logger.error(f"LLM invocation failed: {str(e)}")
            raise
    
    async def stream(
        self,
        messages: List[BaseMessage],
        **kwargs
    ) -> AsyncIterator[str]:
        """Streaming LLM invocation for real-time responses"""
        try:
            async for chunk in self._llm.astream(messages):
                if chunk.content:
                    yield chunk.content
        except Exception as e:
            logger.error(f"Streaming failed: {str(e)}")
            yield f"[Error: {str(e)}]"
    
    async def get_embeddings(self, texts: List[str]) -> List[List[float]]:
        """Get embeddings for texts"""
        try:
            return await self._embeddings.aembed_documents(texts)
        except Exception as e:
            logger.error(f"Embedding generation failed: {str(e)}")
            raise
    
    def _update_total_usage(self, usage: LLMUsageStats):
        """Update cumulative usage statistics"""
        self.total_usage.prompt_tokens += usage.prompt_tokens
        self.total_usage.completion_tokens += usage.completion_tokens
        self.total_usage.total_tokens += usage.total_tokens
        self.total_usage.cost_usd += usage.cost_usd
        self.total_usage.latency_ms += usage.latency_ms
    
    def get_usage_report(self) -> Dict[str, Any]:
        """Generate cost usage report"""
        return {
            "total_prompt_tokens": self.total_usage.prompt_tokens,
            "total_completion_tokens": self.total_usage.completion_tokens,
            "total_tokens": self.total_usage.total_tokens,
            "total_cost_usd": self.total_usage.cost_usd,
            "average_latency_ms": (
                self.total_usage.latency_ms / 
                max(1, self.total_usage.total_tokens)
            ),
            "model": self.total_usage.model,
            "provider": "HolySheep AI"
        }


Factory Pattern for LLM Service

class LLMServiceFactory:
    """Factory for creating LLM service instances"""
    
    @staticmethod
    def create_service(
        provider: str,
        api_key: str,
        model: str = "gpt-4.1",
        **kwargs
    ) -> BaseLLMService:
        
        if provider == "holysheep":
            return HolySheepLLMService(
                api_key=api_key,
                model=model,
                **kwargs
            )
        elif provider == "openai":
            return OpenAILLMService(api_key=api_key, model=model, **kwargs)
        else:
            raise ValueError(f"Unsupported provider: {provider}")


Usage in FastAPI

"""
from fastapi import FastAPI, Depends
from app.services.llm_service import LLMServiceFactory, get_settings

app = FastAPI()

def get_llm_service():
    settings = get_settings()
    return LLMServiceFactory.create_service(
        provider="holysheep",
        api_key=settings.holysheep_api_key,
        model=settings.llm_model,
        temperature=settings.llm_temperature,
        max_tokens=settings.llm_max_tokens
    )

@app.post("/api/chat")
async def chat(request: ChatRequest, llm_service = Depends(get_llm_service)):
    response = await llm_service.invoke(request.messages)
    return {"response": response.content, "usage": response.usage}
"""

RAG Pipeline Optimization

สำหรับ RAG (Retrieval Augmented Generation) ใน production สิ่งที่ต้องควบคุมอย่างเข้มงวดคือ retrieval quality และ latency ผมใช้เทคนิคหลายระดับ:

# Production RAG Pipeline with Advanced Retrieval
from typing import List, Tuple, Optional, AsyncIterator
from dataclasses import dataclass
from enum import Enum
import hashlib

from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

import httpx

class ChunkingStrategy(Enum):
    SEMANTIC = "semantic"
    RECURSIVE = "recursive"
    SEMANTIC_WITH_OVERLAP = "semantic_with_overlap"

@dataclass
class RetrievalResult:
    """Enhanced retrieval result with metadata"""
    content: str
    score: float
    source: str
    page: Optional[int] = None
    chunk_id: str = ""

@dataclass
class RAGConfig:
    """RAG configuration for production use"""
    chunk_size: int = 1024
    chunk_overlap: int = 128
    chunking_strategy: ChunkingStrategy = ChunkingStrategy.SEMANTIC_WITH_OVERLAP
    retrieval_k: int = 10  # Top-k for retrieval
    final_k: int = 5      # Final k after re-ranking
    score_threshold: float = 0.65
    enable_rerank: bool = True
    enable_query_expansion: bool = True
    hybrid_search_alpha: float = 0.7  # Weight for semantic search

class ProductionRAGPipeline:
    """
    Production-grade RAG pipeline with:
    - Semantic chunking
    - Hybrid search (vector + keyword)
    - Re-ranking
    - Query expansion
    - Response generation
    """
    
    def __init__(
        self,
        llm_service,  # HolySheepLLMService instance
        vector_store: Chroma,
        config: RAGConfig = None
    ):
        self.llm_service = llm_service
        self.vector_store = vector_store
        self.config = config or RAGConfig()
        
        # Initialize components
        self._setup_chunking()
        self._setup_retriever()
        self._setup_reranker()
        self._setup_query_expander()
    
    def _setup_chunking(self):
        """Configure text chunking strategy"""
        if self.config.chunking_strategy == ChunkingStrategy.RECURSIVE:
            self.text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=self.config.chunk_size,
                chunk_overlap=self.config.chunk_overlap,
                separators=["\n\n", "\n", " ", ""],
                length_function=len
            )
        else:
            # Semantic chunking with overlap
            self.text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=self.config.chunk_size,
                chunk_overlap=self.config.chunk_overlap,
                separators=["\n\n", "\n", "。", "！", "？", " ", ""]
            )
    
    def _setup_retriever(self):
        """Setup retriever with hybrid search capability"""
        # Base vector store retriever
        self.base_retriever = self.vector_store.as_retriever(
            search_type="similarity_score_threshold",
            search_kwargs={
                "k": self.config.retrieval_k,
                "score_threshold": self.config.score_threshold
            }
        )
        
        # If hybrid search is supported
        if hasattr(self.vector_store, "search"):
            self.retriever = self._hybrid_retriever
        else:
            self.retriever = self.base_retriever
    
    def _setup_reranker(self):
        """Setup re-ranking for improved relevance"""
        if self.config.enable_rerank:
            # Use Cohere or similar for re-ranking
            # In production, you might use a dedicated re-ranking service
            self.compressor = CohereRerank(
                top_n=self.config.final_k,
                cohere_api_key=self.config.cohere_api_key
            )
            self.reranker = ContextualCompressionRetriever(
                base_compressor=self.compressor,
                base_retriever=self.base_retriever
            )
        else:
            self.reranker = self.base_retriever
    
    def _setup_query_expander(self):
        """Setup query expansion for better recall"""
        if self.config.enable_query_expansion:
            self.expansion_prompt = """Given the following query, generate 3 alternative 
            phrasings that capture the same intent but use different words.
            Return only the expanded queries, one per line.
            
            Original Query: {query}
            
            Expanded Queries:"""
    
    async def _expand_query(self, query: str) -> List[str]:
        """Expand query for better retrieval"""
        if not self.config.enable_query_expansion:
            return [query]
        
        messages = [
            SystemMessage(content="You are a query expansion assistant."),
            HumanMessage(content=self.expansion_prompt.format(query=query))
        ]
        
        response = await self.llm_service.invoke(messages)
        expanded = [query] + response.content.strip().split("\n")
        return [q.strip() for q in expanded if q.strip()]
    
    def _generate_chunk_id(self, content: str, index: int) -> str:
        """Generate deterministic chunk ID"""
        hash_input = f"{content[:100]}_{index}".encode()
        return hashlib.md5(hash_input).hexdigest()[:16]
    
    async def _hybrid_retriever(
        self,
        query: str,
        alpha: float = None
    ) -> List[Document]:
        """Hybrid search combining vector and keyword search"""
        alpha = alpha or self.config.hybrid_search_alpha
        
        # Semantic search
        semantic_results = await self.base_retriever.ainvoke(query)
        
        # Keyword search (BM25-style)
        # In production, integrate with Elasticsearch or similar
        keyword_results = await self._keyword_search(query)
        
        # Combine results with Reciprocal Rank Fusion
        fused_scores = self._reciprocal_rank_fusion(
            [semantic_results, keyword_results],
            [alpha, 1 - alpha]
        )
        
        return fused_scores[:self.config.final_k]
    
    def _reciprocal_rank_fusion(
        self,
        result_lists: List[List[Document]],
        weights: List[float]
    ) -> List[Document]:
        """Reciprocal Rank Fusion for combining multiple retrieval results"""
        doc_scores = {}
        
        for results, weight in zip(result_lists, weights):
            for rank, doc in enumerate(results):
                doc_id = doc.page_content[:100]  # Use content hash as ID
                if doc_id not in doc_scores:
                    doc_scores[doc_id] = {"doc": doc, "score": 0}
                doc_scores[doc_id]["score"] += weight * (1 / (rank + 60))
        
        # Sort by fused score
        sorted_docs = sorted(
            doc_scores.values(),
            key=lambda x: x["score"],
            reverse=True
        )
        
        return [item["doc"] for item in sorted_docs]
    
    async def _keyword_search(self, query: str) -> List[Document]:
        """BM25-style keyword search"""
        # In production, query Elasticsearch or similar
        # Placeholder implementation
        return []
    
    async def retrieve_with_expansion(self, query: str) -> List[Document]:
        """Retrieve documents using expanded queries"""
        expanded_queries = await self._expand_query(query)
        
        all_results = []
        for eq in expanded_queries:
            if hasattr(self, '_hybrid_retriever'):
                results = await self._hybrid_retriever(eq)
            else:
                results = await self.base_retriever.ainvoke(eq)
            all_results.extend(results)
        
        # Deduplicate and re-rank
        seen = set()
        unique_results = []
        for doc in all_results:
            doc_id = doc.page_content[:100]
            if doc_id not in seen:
                seen.add(doc_id)
                unique_results.append(doc)
        
        return unique_results[:self.config.final_k]
    
    def format_context(self, docs: List[Document]) -> str:
        """Format retrieved documents as context"""
        context_parts = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("source", "unknown")
            page = doc.metadata.get("page", "")
            context_parts.append(
                f"[{i}] Source: {source}" + 
                (f", Page: {page}" if page else "") +
                f"\n{doc.page_content}"
            )
        return "\n\n".join(context_parts)
    
    async def generate_response(
        self,
        query: str,
        include_sources: bool = True
    ) -> Tuple[str, List[RetrievalResult]]:
        """Full RAG pipeline: retrieve + generate"""
        
        # Step 1: Retrieve relevant documents
        docs = await self.retrieve_with_expansion(query)
        
        if not docs:
            return "ไม่พบข้อมูลที่เกี่ยวข้องกับคำถามของคุณในฐานข้อมูล", []
        
        # Step 2: Format context
        context = self.format_context(docs)
        
        # Step 3: Generate response with RAG prompt
        rag_prompt = f"""คุณเป็นผู้ช่วยที่ให้ข้อมูลที่ถูกต้องโดยอ้างอิงจากเอกสารที่ให้มา

เอกสารที่เกี่ยวข้อง:
{context}

คำถาม: {query}

คำตอบ (อ้างอิงแหล่งที่มาในรูปแบบ [1], [2], ฯลฯ):
"""
        
        messages = [
            SystemMessage(content="คุณเป็นผู้ช่วยที่ให้ข้อมูลแม่นยำและอ้างอิงแหล่งที่มา"),
            HumanMessage(content=rag_prompt)
        ]
        
        response = await self.llm_service.invoke(messages)
        
        # Step 4: Format retrieval results
        retrieval_results = [
            RetrievalResult(
                content=doc.page_content,
                score=0.9 - (i * 0.1),  # Approximate scoring
                source=doc.metadata.get("source", "unknown"),
                page=doc.metadata.get("page"),
                chunk_id=self._generate_chunk_id(doc.page_content, i)
            )
            for i, doc in enumerate(docs)
        ]
        
        return response.content, retrieval_results
    
    async def stream_response(self, query: str) -> AsyncIterator[str]:
        """Streaming RAG response"""
        docs = await self.retrieve_with_expansion(query)
        context = self.format_context(docs)
        
        rag_prompt = f"""เอกสารที่เกี่ยวข้อง:
{context}

คำถาม: {query}

คำตอบ:"""
        
        messages = [
            SystemMessage(content="คุณเป็นผู้ช่วยที่ให้ข้อมูลแม่นยำ"),
            HumanMessage(content=rag_prompt)
        ]
        
        async for chunk in self.llm_service.stream(messages):
            yield chunk


Document Processing Pipeline

class DocumentProcessor:
    """Process documents for RAG indexing"""
    
    def __init__(
        self,
        text_splitter: RecursiveCharacterTextSplitter,
        embeddings: OpenAIEmbeddings
    ):
        self.text_splitter = text_splitter
        self.embeddings = embeddings
    
    def process_document(
        self,
        content: str,
        metadata: dict
    ) -> List[Document]:
        """Split document into chunks with metadata"""
        chunks = self.text_splitter.split_text(content)
        
        documents = []
        for i, chunk in enumerate(chunks):
            doc = Document(
                page_content=chunk,
                metadata={
                    **metadata,
                    "chunk_index": i,
                    "total_chunks": len(chunks),
                    "chunk_id": hashlib.md5(chunk.encode()).hexdigest()[:16]
                }
            )
            documents.append(doc)
        
        return documents
    
    async def process_batch(
        self,
        documents: List[dict],  # List of {"content": str, "metadata": dict}
        batch_size: int = 100
    ) -> List[Document]:
        """Process documents in batches for efficiency"""
        all_docs = []
        
        for doc in documents:
            chunks = self.process_document(doc["content"], doc["metadata"])
            all_docs.extend(chunks)
        
        # Add embeddings in batches
        texts = [doc.page_content for doc in all_docs]
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            await self.embeddings.aembed_documents(batch)
        
        return all_docs


Usage Example

"""

Initialize with HolySheep AI

settings = get_settings()
llm_service = HolySheepLLMService(
    api_key=settings.holysheep_api_key,
    model="deepseek-v3.2",  # Most cost-effective at $0.42/MTok
    temperature=0.3
)


Initialize vector store

vector_store = Chroma(
    collection_name="production_rag",
    embedding_function=OpenAIEmbeddings(
        api_key=settings.holysheep_api_key,
        base_url="https://api.holysheep.ai/v1/embeddings"
    ),
    persist_directory="./data/chroma"
)


Create RAG pipeline

rag_config = RAGConfig(
    chunk_size=1024,
    chunk_overlap=128,
    retrieval_k=10,
    final_k=5,
    enable_rerank=True,
    enable_query_expansion=True
)

rag_pipeline = ProductionRAGPipeline(
    llm_service=llm_service,
    vector_store=vector_store,
    config=rag_config
)


Generate response

answer, sources = await rag_pipeline.generate_response(
    "อธิบายสถาปัตยกรรมของ LangChain"
)
"""

Concurrency Control และ Rate Limiting

ใน production environment การจัดการ concurrency เป็นสิ่งสำคัญมาก โดยเฉพาะเมื่อใช้ rate-limited API เช่น LLM providers ผมใช้ combination ของ semaphore, token bucket และ priority queue สำหรับ request management

# Concurrency Control and Rate Limiting for Production
import asyncio
import time
from typing import Optional, Callable, Any, TypeVar
from dataclasses import dataclass, field
from collections import defaultdict
from enum import Enum
from functools import wraps
import logging
from contextlib import asynccontextmanager

logger = logging.getLogger(__name__)

T = TypeVar('T')

class Priority(Enum):
    HIGH = 1

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