Production Patterns

from functools import lru_cache
import hashlib

# In-memory cache: savolni hash qilib, embedding ni saqlaymiz
_embedding_cache: dict[str, list[float]] = {}

def get_cached_embedding(text: str) -> list[float] | None:
    cache_key = hashlib.sha256(text.encode()).hexdigest()
    if cache_key in _embedding_cache:
        return _embedding_cache[cache_key]
    return None

def set_cached_embedding(text: str, embedding: list[float]) -> None:
    cache_key = hashlib.sha256(text.encode()).hexdigest()
    _embedding_cache[cache_key] = embedding

def generate_query_embedding(text_input: str) -> list[float] | None:
    # Avval cache da bor-yo'qligini tekshiramiz
    cached = get_cached_embedding(text_input)
    if cached is not None:
        return cached

    # Cache da yo'q — API ga so'rov yuboramiz
    embedding = _call_embedding_api(text_input)
    if embedding is not None:
        set_cached_embedding(text_input, embedding)
    return embedding

import time

_response_cache: dict[str, dict] = {}
CACHE_TTL = 3600  # 1 soat

def get_cached_response(question: str) -> str | None:
    cache_key = hashlib.sha256(question.encode()).hexdigest()
    entry = _response_cache.get(cache_key)
    if entry is None:
        return None
    if time.time() - entry["timestamp"] > CACHE_TTL:
        del _response_cache[cache_key]
        return None
    return entry["response"]

def set_cached_response(question: str, response: str) -> None:
    cache_key = hashlib.sha256(question.encode()).hexdigest()
    _response_cache[cache_key] = {
        "response": response,
        "timestamp": time.time(),
    }

import time
from collections import defaultdict
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests: int = 10, window_seconds: int = 60):
        super().__init__(app)
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: dict[str, list[float]] = defaultdict(list)

    async def dispatch(self, request: Request, call_next):
        # Faqat API endpoint lariga rate limit qo'yamiz
        if not request.url.path.startswith("/api/"):
            return await call_next(request)

        client_ip = request.client.host
        now = time.time()
        window_start = now - self.window_seconds

        # Eskirgan so'rovlarni tozalaymiz
        self.requests[client_ip] = [
            t for t in self.requests[client_ip] if t > window_start
        ]

        if len(self.requests[client_ip]) >= self.max_requests:
            raise HTTPException(
                status_code=429,
                detail="Too many requests. Please wait before trying again.",
            )

        self.requests[client_ip].append(now)
        return await call_next(request)

# main.py da:
app.add_middleware(RateLimitMiddleware, max_requests=10, window_seconds=60)

import logging
import time

logger = logging.getLogger("rag")
logger.setLevel(logging.INFO)

handler = logging.FileHandler("rag_queries.log")
handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)s | %(message)s"
))
logger.addHandler(handler)

def process_chat(session, question: str) -> str:
    start = time.time()

    # Embedding
    embedding = generate_query_embedding(question)
    embed_time = time.time() - start

    # Retrieval
    chunks = retrieve_chunks(session, embedding)
    retrieval_time = time.time() - start - embed_time
    top_sim = chunks[0]["similarity"] if chunks else 0

    # Chunk tanlash + Generation
    chunks = select_context_chunks(chunks, question)
    messages = build_prompt(chunks, question)
    answer = "".join(stream_response(messages))
    total_time = time.time() - start

    # Log
    logger.info(
        f"Q: {question[:80]} | "
        f"Chunks: {len(chunks)} | "
        f"TopSim: {top_sim:.3f} | "
        f"EmbedTime: {embed_time:.2f}s | "
        f"RetrTime: {retrieval_time:.2f}s | "
        f"TotalTime: {total_time:.2f}s"
    )

    # Alert: similarity juda past bo'lsa
    if top_sim < 0.3:
        logger.warning(f"LOW SIMILARITY: {question[:80]} -> {top_sim:.3f}")

    return answer

2025-07-15 14:32:01 | INFO | Q: Расскажите про проект HUMO | Chunks: 5 | TopSim: 0.847 | EmbedTime: 0.23s | RetrTime: 0.05s | TotalTime: 2.41s
2025-07-15 14:32:15 | WARNING | LOW SIMILARITY: Какая погода в Ташкенте? -> 0.187

def _write_analytics_chat(**kwargs):
    s = SessionLocal()
    try:
        s.add(LgAnalyticsChat(**kwargs))
        s.commit()
    except Exception:
        s.rollback()   # analytics xatosi asosiy oqimni buzmaydi
    finally:
        s.close()

WDE (Local)                    Chef (Cloud)
Port 8005                      Port 8006
OLLAMA (qwen3:8b)              OpenRouter (gemini-2.0-flash)
qwen3-embedding:8b             text-embedding-3-small
LLM-based reranker             Jina Reranker v2

def embed_all(session: Session) -> dict:
    # Faqat embedding si yo'q chunk larni olamiz
    pending = session.query(LvContentChunks).filter(
        LvContentChunks.embedding_local.is_(None),
    ).all()

    embedded = 0
    errors = []

    for chunk in pending:
        try:
            vector = _generate_embedding(chunk.chunk_text)
            if vector is None:
                errors.append({"chunk_id": chunk.id, "error": "empty response"})
                continue
            chunk.embedding_local = json.dumps(vector)
            session.commit()
            embedded += 1
        except Exception as e:
            errors.append({"chunk_id": chunk.id, "error": str(e)})
            session.rollback()

    already = session.query(LvContentChunks).filter(
        LvContentChunks.embedding_local.isnot(None),
    ).count()

    return {
        "embedded": embedded,
        "already_done": already - embedded,
        "errors": errors,
    }

# crontab -e
0 3 * * * curl -X POST http://localhost:8005/api/admin/embed-all

class AuthMiddleware(BaseHTTPMiddleware):
    ALLOWED_PATHS = {"/login", "/api/auth/login", "/api/auth/logout"}
    USER_PATHS = {"/chat", "/api/chat", "/api/rubrics", "/api/feedback", ...}
    GUEST_PATHS = {"/chat", "/api/chat", "/api/rubrics", ...}  # feedback POST yo'q
    ANALYTICS_PATHS = {"/analytics", "/api/questions", "/api/analytics/users"}

    async def dispatch(self, request, call_next):
        path = request.url.path
        if path in self.ALLOWED_PATHS or path.startswith("/static"):
            return await call_next(request)
        role = request.session.get("role")
        if role == "admin":
            return await call_next(request)                    # hamma narsa ochiq
        if role == "user" and path in self.USER_PATHS:
            return await call_next(request)                    # faqat chat + feedback
        if role == "analytics" and path in self.ANALYTICS_PATHS | self.USER_PATHS:
            return await call_next(request)                    # + analytics sahifalar
        if not request.session.get("authenticated") and path in self.GUEST_PATHS:
            return await call_next(request)                    # mehmon: cheklangan
        if request.method == "GET":
            return RedirectResponse("/login", status_code=303)
        return JSONResponse({"detail": "Not authenticated"}, status_code=401)

def safe_chat(session: Session, question: str) -> str:
    # 1. Embedding service down
    try:
        embedding = generate_query_embedding(question)
    except Exception:
        return "[Embedding xizmati vaqtincha ishlamayapti. Keyinroq urinib ko'ring.]"

    if embedding is None:
        return "[Embedding yaratib bo'lmadi. Xizmat yuklanmoqda.]"

    # 2. Database connection failed
    try:
        chunks = retrieve_chunks(session, embedding)
    except Exception:
        return "[Ma'lumotlar bazasiga ulanib bo'lmadi.]"

    # 3. Chunk tanlash — heuristika, tashqi servisga bog'liq emas
    chunks = select_context_chunks(chunks, question)

    # 4. LLM timeout
    try:
        messages = build_prompt(chunks, question)
        answer = "".join(stream_response(messages))
    except Exception:
        return "[Javob yaratishda xatolik. Iltimos, qayta urinib ko'ring.]"

    return answer

def rerank_chunks_with_fallback(chunks, question, top_n=None):
    # 1-urinish: Jina Reranker (tashqi API)
    if settings.JINA_API_KEY:
        try:
            return _jina_rerank(chunks, question, top_n)
        except Exception:
            pass

    # 2-urinish: LLM-based reranking (lokal)
    try:
        return _llm_rerank(chunks, question, top_n)
    except Exception:
        pass

    # 3-fallback: oddiy similarity tartibi
    return chunks[:top_n]

-- Brute-force: barcha chunk larni tekshiradi
SELECT id, chunk_text, 1 - (embedding_local <=> :vec) AS similarity
FROM wde.lv_content_chunks
WHERE embedding_local IS NOT NULL
ORDER BY embedding_local <=> :vec
LIMIT 15;

-- IVFFlat index: tez, lekin approximate
CREATE INDEX idx_chunks_embedding ON wde.lv_content_chunks
USING ivfflat (embedding_cloud vector_cosine_ops)
WITH (lists = 100);

-- HNSW index: aniqroq, lekin ko'proq xotira
CREATE INDEX idx_chunks_embedding_hnsw ON wde.lv_content_chunks
USING hnsw (embedding_cloud vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

# Bir nechta uvicorn worker bilan ishga tushirish
# uvicorn app.main:app --workers 4 --host 0.0.0.0 --port 8005

# SQLAlchemy connection pool sozlamalari
engine = create_engine(
    settings.DATABASE_URL,
    pool_size=5,       # Har bir worker uchun 5 ta connection
    max_overflow=10,   # Yuklama oshganda qo'shimcha 10 ta
    pool_pre_ping=True, # Har bir so'rovdan oldin connection tirikligini tekshirish
)