Day 30: Quantization & Deploy Local Model API

Day 25: quyết định RAG/fine-tune/tool/prompt
Day 26: dataset instruction tuning
Day 27: LoRA/QLoRA hands-on
Day 28: evaluation trước/sau fine-tune
Day 29: local LLM runtime: Ollama, llama.cpp, vLLM
Day 30: quantization + local model API + benchmark

weights_memory_gb ~= params_billion * bytes_per_param

kv_cache_bytes ~= layers * 2 * kv_heads * head_dim * seq_len * concurrent_sequences * bytes_per_kv_element

Một model 7B INT4 có thể chỉ cần ~5GB weights,
nhưng với context dài và nhiều request song song,
KV cache + overhead có thể làm GPU 8GB OOM.

required_memory =
  weights_memory
  + kv_cache_memory
  + runtime_overhead
  + fragmentation_and_safety_margin

Client / Product service
  -> FastAPI Local Model Gateway
      -> auth / API key / tenant policy
      -> request validation
      -> prompt template version
      -> timeout and max_tokens
      -> concurrency limiter
      -> structured logging / trace_id
      -> readiness check
      -> LocalLLMClient
          -> Ollama / llama.cpp server / vLLM / TGI
  -> Response

from __future__ import annotations

import asyncio
import json
import logging
import os
import time
import uuid
from contextlib import asynccontextmanager
from typing import Any

import httpx
import psutil
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field

logger = logging.getLogger("local_model_api")
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))


class Settings(BaseModel):
    runtime_base_url: str = os.getenv("LOCAL_LLM_BASE_URL", "http://localhost:11434/v1")
    api_key: str = os.getenv("LOCAL_LLM_API_KEY", "local")
    model: str = os.getenv("LOCAL_LLM_MODEL", "llama3.2")
    runtime: str = os.getenv("LOCAL_LLM_RUNTIME", "ollama")
    request_timeout_s: float = float(os.getenv("REQUEST_TIMEOUT_S", "60"))
    max_concurrency: int = int(os.getenv("MAX_CONCURRENCY", "4"))


settings = Settings()
semaphore = asyncio.Semaphore(settings.max_concurrency)
ready_state = {"ready": False, "last_error": None}


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=8000)
    system: str = Field(default="You are a concise internal assistant.", max_length=2000)
    temperature: float = Field(default=0.2, ge=0.0, le=1.5)
    max_tokens: int = Field(default=512, ge=1, le=2048)


class ChatResponse(BaseModel):
    answer: str
    model: str
    runtime: str
    latency_ms: float
    memory_rss_mb: float
    trace_id: str


async def call_openai_compatible(req: ChatRequest) -> str:
    headers = {"Authorization": f"Bearer {settings.api_key}"}
    payload: dict[str, Any] = {
        "model": settings.model,
        "messages": [
            {"role": "system", "content": req.system},
            {"role": "user", "content": req.message},
        ],
        "temperature": req.temperature,
        "max_tokens": req.max_tokens,
    }
    timeout = httpx.Timeout(settings.request_timeout_s)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.post(
            f"{settings.runtime_base_url.rstrip('/')}/chat/completions",
            headers=headers,
            json=payload,
        )
        response.raise_for_status()
        data = response.json()
        return data["choices"][0]["message"]["content"] or ""


@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        async with httpx.AsyncClient(timeout=5) as client:
            await client.get(f"{settings.runtime_base_url.rstrip('/')}/models")
        ready_state["ready"] = True
    except Exception as exc:
        ready_state["last_error"] = str(exc)
        logger.warning(json.dumps({"event": "model_runtime_not_ready", "error": str(exc)}))
    yield


app = FastAPI(title="Local Model API", version="1.0.0", lifespan=lifespan)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/ready")
async def ready():
    if not ready_state["ready"]:
        raise HTTPException(status_code=503, detail=ready_state)
    return {"status": "ready", "model": settings.model, "runtime": settings.runtime}


@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest, request: Request):
    trace_id = request.headers.get("x-request-id", str(uuid.uuid4()))
    start = time.perf_counter()

    try:
        async with semaphore:
            answer = await asyncio.wait_for(
                call_openai_compatible(req),
                timeout=settings.request_timeout_s + 2,
            )
    except TimeoutError as exc:
        raise HTTPException(status_code=504, detail={"trace_id": trace_id, "error": "timeout"}) from exc
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=502, detail={"trace_id": trace_id, "error": str(exc)}) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail={"trace_id": trace_id, "error": type(exc).__name__}) from exc

    latency_ms = (time.perf_counter() - start) * 1000
    memory_rss_mb = psutil.Process().memory_info().rss / 1024 / 1024

    logger.info(
        json.dumps(
            {
                "event": "chat_completed",
                "trace_id": trace_id,
                "model": settings.model,
                "runtime": settings.runtime,
                "latency_ms": round(latency_ms, 2),
                "input_chars": len(req.message),
                "output_chars": len(answer),
                "memory_rss_mb": round(memory_rss_mb, 2),
            },
            ensure_ascii=False,
        )
    )

    return ChatResponse(
        answer=answer,
        model=settings.model,
        runtime=settings.runtime,
        latency_ms=round(latency_ms, 2),
        memory_rss_mb=round(memory_rss_mb, 2),
        trace_id=trace_id,
    )

pip install -U httpx psutil
python benchmark_local_api.py --url http://localhost:9000/chat --concurrency 4 --repeat 5

1. Xác định task và SLA
2. Chọn baseline model FP16/BF16 hoặc hosted model mạnh
3. Tạo golden set và metric
4. Chọn runtime theo hardware
5. Thử quantization theo memory target
6. Benchmark latency + throughput + memory
7. Eval quality regression
8. Quyết định production/canary/rollback

weights_gb ~= params_billion * bytes_per_param

kv_cache_bytes ~= layers * 2 * kv_heads * head_dim * seq_len * concurrent_sequences * bytes

Nếu tăng max context từ 4k lên 16k,
KV cache tăng xấp xỉ 4 lần.

Nếu tăng concurrency từ 2 lên 8,
KV cache tăng xấp xỉ 4 lần.

pip install -U fastapi uvicorn httpx pydantic psutil

export LOCAL_LLM_BASE_URL=http://localhost:11434/v1
export LOCAL_LLM_API_KEY=local
export LOCAL_LLM_MODEL=llama3.2
export LOCAL_LLM_RUNTIME=ollama
export REQUEST_TIMEOUT_S=60
export MAX_CONCURRENCY=4

uvicorn app:app --host 0.0.0.0 --port 9000

from __future__ import annotations

import asyncio
import json
import logging
import os
import time
import uuid
from contextlib import asynccontextmanager
from typing import Any

import httpx
import psutil
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field


logging.basicConfig(
    level=os.getenv("LOG_LEVEL", "INFO"),
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("local_model_api")


class Settings(BaseModel):
    runtime_base_url: str = os.getenv("LOCAL_LLM_BASE_URL", "http://localhost:11434/v1")
    api_key: str = os.getenv("LOCAL_LLM_API_KEY", "local")
    model: str = os.getenv("LOCAL_LLM_MODEL", "llama3.2")
    runtime: str = os.getenv("LOCAL_LLM_RUNTIME", "ollama")
    request_timeout_s: float = float(os.getenv("REQUEST_TIMEOUT_S", "60"))
    max_concurrency: int = int(os.getenv("MAX_CONCURRENCY", "4"))
    max_input_chars: int = int(os.getenv("MAX_INPUT_CHARS", "8000"))
    max_output_tokens: int = int(os.getenv("MAX_OUTPUT_TOKENS", "2048"))


settings = Settings()
semaphore = asyncio.Semaphore(settings.max_concurrency)
ready_state: dict[str, Any] = {"ready": False, "last_error": None}


class ChatRequest(BaseModel):
    message: str = Field(min_length=1)
    system: str = Field(default="You are a concise internal assistant.", max_length=2000)
    temperature: float = Field(default=0.2, ge=0.0, le=1.5)
    max_tokens: int = Field(default=512, ge=1)


class ChatResponse(BaseModel):
    answer: str
    model: str
    runtime: str
    latency_ms: float
    memory_rss_mb: float
    trace_id: str


class ErrorResponse(BaseModel):
    trace_id: str
    error: str


def validate_limits(req: ChatRequest) -> None:
    if len(req.message) > settings.max_input_chars:
        raise HTTPException(status_code=413, detail="message too large")
    if req.max_tokens > settings.max_output_tokens:
        raise HTTPException(status_code=422, detail="max_tokens exceeds server limit")


async def call_model(req: ChatRequest) -> str:
    headers = {"Authorization": f"Bearer {settings.api_key}"}
    payload: dict[str, Any] = {
        "model": settings.model,
        "messages": [
            {"role": "system", "content": req.system},
            {"role": "user", "content": req.message},
        ],
        "temperature": req.temperature,
        "max_tokens": req.max_tokens,
    }

    async with httpx.AsyncClient(timeout=httpx.Timeout(settings.request_timeout_s)) as client:
        response = await client.post(
            f"{settings.runtime_base_url.rstrip('/')}/chat/completions",
            headers=headers,
            json=payload,
        )
        response.raise_for_status()
        data = response.json()

    return data["choices"][0]["message"]["content"] or ""


@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        async with httpx.AsyncClient(timeout=5) as client:
            response = await client.get(f"{settings.runtime_base_url.rstrip('/')}/models")
            response.raise_for_status()
        ready_state["ready"] = True
    except Exception as exc:
        ready_state["last_error"] = str(exc)
        logger.warning(json.dumps({"event": "model_runtime_not_ready", "error": str(exc)}))

    yield

    ready_state["ready"] = False


app = FastAPI(
    title="Local Model API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/ready")
async def ready() -> dict[str, Any]:
    if not ready_state["ready"]:
        raise HTTPException(status_code=503, detail=ready_state)
    return {
        "status": "ready",
        "model": settings.model,
        "runtime": settings.runtime,
        "max_concurrency": settings.max_concurrency,
    }


@app.post(
    "/chat",
    response_model=ChatResponse,
    responses={502: {"model": ErrorResponse}, 504: {"model": ErrorResponse}},
)
async def chat(req: ChatRequest, request: Request) -> ChatResponse:
    validate_limits(req)

    trace_id = request.headers.get("x-request-id", str(uuid.uuid4()))
    start = time.perf_counter()

    try:
        async with semaphore:
            answer = await asyncio.wait_for(
                call_model(req),
                timeout=settings.request_timeout_s + 2,
            )
    except TimeoutError as exc:
        raise HTTPException(
            status_code=504,
            detail={"trace_id": trace_id, "error": "model timeout"},
        ) from exc
    except httpx.HTTPStatusError as exc:
        raise HTTPException(
            status_code=502,
            detail={"trace_id": trace_id, "error": f"runtime returned {exc.response.status_code}"},
        ) from exc
    except Exception as exc:
        logger.exception(json.dumps({"event": "chat_failed", "trace_id": trace_id}))
        raise HTTPException(
            status_code=500,
            detail={"trace_id": trace_id, "error": type(exc).__name__},
        ) from exc

    latency_ms = (time.perf_counter() - start) * 1000
    memory_rss_mb = psutil.Process().memory_info().rss / 1024 / 1024

    logger.info(
        json.dumps(
            {
                "event": "chat_completed",
                "trace_id": trace_id,
                "model": settings.model,
                "runtime": settings.runtime,
                "latency_ms": round(latency_ms, 2),
                "input_chars": len(req.message),
                "output_chars": len(answer),
                "max_tokens": req.max_tokens,
                "memory_rss_mb": round(memory_rss_mb, 2),
            },
            ensure_ascii=False,
        )
    )

    return ChatResponse(
        answer=answer,
        model=settings.model,
        runtime=settings.runtime,
        latency_ms=round(latency_ms, 2),
        memory_rss_mb=round(memory_rss_mb, 2),
        trace_id=trace_id,
    )

from __future__ import annotations

import argparse
import asyncio
import statistics
import time
from dataclasses import dataclass

import httpx
import psutil


PROMPTS = [
    "Tóm tắt local LLM trong 3 bullet.",
    "So sánh INT8 và INT4 cho production serving.",
    "Trả lời JSON với keys: risks, metrics, rollback.",
    "Giải thích vì sao KV cache tăng theo context length.",
]


@dataclass
class Result:
    latency_ms: float
    ok: bool
    output_chars: int
    error: str | None = None


async def one_request(client: httpx.AsyncClient, url: str, prompt: str, max_tokens: int) -> Result:
    start = time.perf_counter()
    try:
        response = await client.post(
            url,
            json={"message": prompt, "max_tokens": max_tokens},
        )
        elapsed_ms = (time.perf_counter() - start) * 1000
        response.raise_for_status()
        data = response.json()
        return Result(latency_ms=elapsed_ms, ok=True, output_chars=len(data.get("answer", "")))
    except Exception as exc:
        elapsed_ms = (time.perf_counter() - start) * 1000
        return Result(latency_ms=elapsed_ms, ok=False, output_chars=0, error=type(exc).__name__)


async def run(url: str, concurrency: int, repeat: int, timeout_s: float, max_tokens: int) -> None:
    prompts = (PROMPTS * repeat)[: len(PROMPTS) * repeat]
    limits = httpx.Limits(max_connections=concurrency, max_keepalive_connections=concurrency)

    async with httpx.AsyncClient(timeout=timeout_s, limits=limits) as client:
        pending = []
        for prompt in prompts:
            pending.append(one_request(client, url, prompt, max_tokens))
            if len(pending) == concurrency:
                yield_results = await asyncio.gather(*pending)
                for item in yield_results:
                    results.append(item)
                pending = []
        if pending:
            for item in await asyncio.gather(*pending):
                results.append(item)


def percentile(values: list[float], pct: float) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    index = max(0, min(len(ordered) - 1, int(len(ordered) * pct) - 1))
    return ordered[index]


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--url", default="http://localhost:9000/chat")
    parser.add_argument("--concurrency", type=int, default=4)
    parser.add_argument("--repeat", type=int, default=5)
    parser.add_argument("--timeout-s", type=float, default=90)
    parser.add_argument("--max-tokens", type=int, default=300)
    args = parser.parse_args()

    results: list[Result] = []
    process = psutil.Process()
    rss_before_mb = process.memory_info().rss / 1024 / 1024
    started = time.perf_counter()

    asyncio.run(run(args.url, args.concurrency, args.repeat, args.timeout_s, args.max_tokens))

    total_s = time.perf_counter() - started
    rss_after_mb = process.memory_info().rss / 1024 / 1024
    ok_latencies = [r.latency_ms for r in results if r.ok]
    errors = [r.error for r in results if not r.ok]

    print(
        {
            "count": len(results),
            "ok": len(ok_latencies),
            "error_count": len(errors),
            "error_types": sorted(set(e for e in errors if e)),
            "total_s": round(total_s, 2),
            "requests_per_s": round(len(results) / total_s, 2) if total_s else 0,
            "p50_ms": round(statistics.median(ok_latencies), 2) if ok_latencies else 0,
            "p95_ms": round(percentile(ok_latencies, 0.95), 2),
            "p99_ms": round(percentile(ok_latencies, 0.99), 2),
            "avg_ms": round(statistics.mean(ok_latencies), 2) if ok_latencies else 0,
            "avg_output_chars": round(statistics.mean([r.output_chars for r in results if r.ok]), 2)
            if ok_latencies
            else 0,
            "client_rss_before_mb": round(rss_before_mb, 2),
            "client_rss_after_mb": round(rss_after_mb, 2),
        }
    )

ps -o pid,rss,comm -p <PID>
top -p <PID>

nvidia-smi --query-gpu=timestamp,name,memory.used,memory.total,utilization.gpu --format=csv -l 1

ollama ps

Chấp nhận INT4 nếu:
- p95 latency giảm hoặc memory fit rõ ràng,
- format accuracy giảm không quá 1%,
- task accuracy giảm không quá 2%,
- không tăng critical hallucination,
- error rate < 0.5% dưới benchmark target.

ollama pull llama3.2
ollama serve

export LOCAL_LLM_BASE_URL=http://localhost:11434/v1
export LOCAL_LLM_API_KEY=local
export LOCAL_LLM_MODEL=llama3.2
export LOCAL_LLM_RUNTIME=ollama

./llama-server -m model.gguf --host 0.0.0.0 --port 8080 --ctx-size 4096

export LOCAL_LLM_BASE_URL=http://localhost:8080/v1
export LOCAL_LLM_API_KEY=local
export LOCAL_LLM_MODEL=local-gguf
export LOCAL_LLM_RUNTIME=llama.cpp

python -m vllm.entrypoints.openai.api_server \
  --model <model-id> \
  --host 0.0.0.0 \
  --port 8000

export LOCAL_LLM_BASE_URL=http://localhost:8000/v1
export LOCAL_LLM_API_KEY=local
export LOCAL_LLM_MODEL=<model-id>
export LOCAL_LLM_RUNTIME=vllm

pip install -U fastapi uvicorn httpx pydantic psutil

uvicorn app:app --host 0.0.0.0 --port 9000

curl http://localhost:9000/health
curl http://localhost:9000/ready
curl -X POST http://localhost:9000/chat \
  -H 'content-type: application/json' \
  -d '{"message":"Giải thích INT4 trong 3 bullet.","max_tokens":200}'

python benchmark_local_api.py --concurrency 1 --repeat 5 --max-tokens 200
python benchmark_local_api.py --concurrency 4 --repeat 5 --max-tokens 300
python benchmark_local_api.py --concurrency 8 --repeat 5 --max-tokens 300

ollama ps
nvidia-smi

nvidia-smi --query-gpu=timestamp,name,memory.used,memory.total,utilization.gpu --format=csv -l 1

ps -o pid,rss,comm -p <PID>
top -p <PID>

# Production Decision

## Context

- Use case:
- Traffic target:
- SLA:
- Hardware:
- Runtime:
- Model:
- Quantization:

## Benchmark Summary

- p50:
- p95:
- p99:
- req/s:
- RAM/VRAM peak:
- error rate:

## Quality Summary

- Golden set size:
- Pass rate baseline:
- Pass rate quantized:
- Regression:

## Decision

Dùng được trong production không?

## Điều kiện

- Điều kiện 1:
- Điều kiện 2:
- Điều kiện 3:

## Rollback/Fallback

- Khi timeout:
- Khi OOM:
- Khi quality regression: